Skip to content

rslearn.utils.m2m_api

m2m_api

API client for the USGS M2M API.

APIException

Bases: Exception

Exception raised for M2M API errors.

Source code in rslearn/utils/m2m_api.py 
17
18
19
20
class APIException(Exception):
    """Exception raised for M2M API errors."""

    pass

M2MAPIClient

An API client for interacting with the USGS M2M API.

Source code in rslearn/utils/m2m_api.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
class M2MAPIClient:
    """An API client for interacting with the USGS M2M API."""

    api_url = "https://m2m.cr.usgs.gov/api/api/json/stable/"
    pagination_size = 1000

    def __init__(
        self,
        username: str | None = None,
        token: str | None = None,
        timeout: timedelta = timedelta(seconds=120),
        session: requests.Session | None = None,
    ) -> None:
        """Initialize a new M2MAPIClient.

        Args:
            username: the EROS username. If None, uses M2M_USERNAME environment variable.
            token: the application token. If None, uses M2M_TOKEN environment variable.
            timeout: timeout for requests.
            session: optional requests session to use. If None, a default session is
                created.
        """
        if username is None:
            username = os.environ.get("M2M_USERNAME")
            if username is None:
                raise ValueError(
                    "username must be provided or M2M_USERNAME environment variable must be set"
                )
        if token is None:
            token = os.environ.get("M2M_TOKEN")
            if token is None:
                raise ValueError(
                    "token must be provided or M2M_TOKEN environment variable must be set"
                )

        self.username = username
        self.timeout = timeout
        self.session = session if session is not None else requests.Session()

        json_data = json.dumps({"username": username, "token": token})
        response = self.session.post(
            self.api_url + "login-token",
            data=json_data,
            timeout=self.timeout.total_seconds(),
        )

        response.raise_for_status()
        self.auth_token = response.json()["data"]

    def request(
        self, endpoint: str, data: dict[str, Any] | None = None
    ) -> dict[str, Any] | None:
        """Make a request to the API.

        Args:
            endpoint: the endpoint to call
            data: POST data to pass

        Returns:
            JSON response data if any
        """
        response = self.session.post(
            self.api_url + endpoint,
            headers={"X-Auth-Token": self.auth_token},
            data=json.dumps(data),
            timeout=self.timeout.total_seconds(),
        )
        response.raise_for_status()
        if response.text:
            response_dict = response.json()

            if response_dict["errorMessage"]:
                raise APIException(response_dict["errorMessage"])
            return response_dict
        return None

    def get_filters(self, dataset_name: str) -> list[dict[str, Any]]:
        """Returns filters available for the given dataset.

        Args:
            dataset_name: the dataset name e.g. landsat_ot_c2_l1

        Returns:
            list of filter objects
        """
        response_dict = self.request("dataset-filters", {"datasetName": dataset_name})
        if response_dict is None:
            raise APIException("No response from API")
        return response_dict["data"]

    def scene_search(
        self,
        dataset_name: str,
        acquisition_time_range: tuple[datetime, datetime] | None = None,
        cloud_cover_range: tuple[int, int] | None = None,
        bbox: tuple[float, float, float, float] | None = None,
        metadata_filter: dict[str, Any] | None = None,
    ) -> list[dict[str, Any]]:
        """Search for scenes matching the arguments.

        Args:
            dataset_name: the dataset name e.g. landsat_ot_c2_l1
            acquisition_time_range: optional filter on the acquisition time
            cloud_cover_range: optional filter on the cloud cover
            bbox: optional spatial filter
            metadata_filter: optional metadata filter dict
        """
        base_data: dict[str, Any] = {"datasetName": dataset_name, "sceneFilter": {}}
        if acquisition_time_range:
            base_data["sceneFilter"]["acquisitionFilter"] = {
                "start": acquisition_time_range[0].isoformat(),
                "end": acquisition_time_range[1].isoformat(),
            }
        if cloud_cover_range:
            base_data["sceneFilter"]["cloudCoverFilter"] = {
                "min": cloud_cover_range[0],
                "max": cloud_cover_range[1],
                "includeUnknown": False,
            }
        if bbox:
            base_data["sceneFilter"]["spatialFilter"] = {
                "filterType": "mbr",
                "lowerLeft": {"longitude": bbox[0], "latitude": bbox[1]},
                "upperRight": {"longitude": bbox[2], "latitude": bbox[3]},
            }
        if metadata_filter:
            base_data["sceneFilter"]["metadataFilter"] = metadata_filter

        starting_number = 1
        results = []
        while True:
            cur_data = base_data.copy()
            cur_data["startingNumber"] = starting_number
            cur_data["maxResults"] = self.pagination_size
            response_dict = self.request("scene-search", cur_data)
            if response_dict is None:
                raise APIException("No response from API")
            data = response_dict["data"]
            results.extend(data["results"])
            if data["recordsReturned"] < self.pagination_size:
                break
            starting_number += self.pagination_size

        return results

    def get_scene_metadata(self, dataset_name: str, entity_id: str) -> dict[str, Any]:
        """Get detailed metadata for a scene.

        Args:
            dataset_name: the dataset name in which to search
            entity_id: the entity ID of the scene

        Returns:
            full scene metadata
        """
        response_dict = self.request(
            "scene-metadata",
            {
                "datasetName": dataset_name,
                "entityId": entity_id,
                "metadataType": "full",
            },
        )
        if response_dict is None:
            raise APIException("No response from API")
        return response_dict["data"]

    def get_downloadable_products(
        self, dataset_name: str, entity_id: str
    ) -> list[dict[str, Any]]:
        """Get the downloadable products for a given scene.

        Args:
            dataset_name: the dataset name
            entity_id: the entity ID of the scene

        Returns:
            list of downloadable products
        """
        data = {"datasetName": dataset_name, "entityIds": [entity_id]}
        response_dict = self.request("download-options", data)
        if response_dict is None:
            raise APIException("No response from API")
        return response_dict["data"]

    def get_download_url(self, entity_id: str, product_id: str) -> str:
        """Get the download URL for a given product.

        Args:
            entity_id: the entity ID of the product
            product_id: the product ID of the product

        Returns:
            the download URL
        """
        label = str(uuid.uuid4())
        data = {
            "downloads": [
                {"label": label, "entityId": entity_id, "productId": product_id}
            ]
        }
        response_dict = self.request("download-request", data)
        if response_dict is None:
            raise APIException("No response from API")
        response = response_dict["data"]

        # Check if URL is immediately available in the response
        if response.get("availableDownloads"):
            return response["availableDownloads"][0]["url"]

        # Otherwise poll download-retrieve until the URL is ready
        while True:
            # We sleep first because we have observed that if we immediately call
            # download-retrieve it sometimes has a response that includes neither
            # available nor requested list.
            time.sleep(10)
            response_dict = self.request("download-retrieve", {"label": label})
            if response_dict is None:
                raise APIException("No response from API")
            response = response_dict["data"]
            if len(response["available"]) > 0:
                return response["available"][0]["url"]
            if len(response["requested"]) == 0:
                logger.debug(
                    f"Response to download-retrieve did not include our requested download under label {label}: {response}"
                )
                raise APIException("Did not get download URL")
            if response["requested"][0].get("url"):
                return response["requested"][0]["url"]

request 

request(endpoint: str, data: dict[str, Any] | None = None) -> dict[str, Any] | None

Make a request to the API.

Parameters:

Name Type Description Default
endpoint str

the endpoint to call

 required
data dict[str, Any] | None

POST data to pass

 None

Returns:

Type Description
dict[str, Any] | None

JSON response data if any
Source code in rslearn/utils/m2m_api.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def request(
    self, endpoint: str, data: dict[str, Any] | None = None
) -> dict[str, Any] | None:
    """Make a request to the API.

    Args:
        endpoint: the endpoint to call
        data: POST data to pass

    Returns:
        JSON response data if any
    """
    response = self.session.post(
        self.api_url + endpoint,
        headers={"X-Auth-Token": self.auth_token},
        data=json.dumps(data),
        timeout=self.timeout.total_seconds(),
    )
    response.raise_for_status()
    if response.text:
        response_dict = response.json()

        if response_dict["errorMessage"]:
            raise APIException(response_dict["errorMessage"])
        return response_dict
    return None

get_filters 

get_filters(dataset_name: str) -> list[dict[str, Any]]

Returns filters available for the given dataset.

Parameters:

Name Type Description Default
dataset_name str

the dataset name e.g. landsat_ot_c2_l1

 required

Returns:

Type Description
list[dict[str, Any]]

list of filter objects
Source code in rslearn/utils/m2m_api.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
def get_filters(self, dataset_name: str) -> list[dict[str, Any]]:
    """Returns filters available for the given dataset.

    Args:
        dataset_name: the dataset name e.g. landsat_ot_c2_l1

    Returns:
        list of filter objects
    """
    response_dict = self.request("dataset-filters", {"datasetName": dataset_name})
    if response_dict is None:
        raise APIException("No response from API")
    return response_dict["data"]

scene_search(dataset_name: str, acquisition_time_range: tuple[datetime, datetime] | None = None, cloud_cover_range: tuple[int, int] | None = None, bbox: tuple[float, float, float, float] | None = None, metadata_filter: dict[str, Any] | None = None) -> list[dict[str, Any]]

Search for scenes matching the arguments.

Parameters:

Name Type Description Default
dataset_name str

the dataset name e.g. landsat_ot_c2_l1

 required
acquisition_time_range tuple[datetime, datetime] | None

optional filter on the acquisition time

 None
cloud_cover_range tuple[int, int] | None

optional filter on the cloud cover

 None
bbox tuple[float, float, float, float] | None

optional spatial filter

 None
metadata_filter dict[str, Any] | None

optional metadata filter dict

 None
Source code in rslearn/utils/m2m_api.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def scene_search(
    self,
    dataset_name: str,
    acquisition_time_range: tuple[datetime, datetime] | None = None,
    cloud_cover_range: tuple[int, int] | None = None,
    bbox: tuple[float, float, float, float] | None = None,
    metadata_filter: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    """Search for scenes matching the arguments.

    Args:
        dataset_name: the dataset name e.g. landsat_ot_c2_l1
        acquisition_time_range: optional filter on the acquisition time
        cloud_cover_range: optional filter on the cloud cover
        bbox: optional spatial filter
        metadata_filter: optional metadata filter dict
    """
    base_data: dict[str, Any] = {"datasetName": dataset_name, "sceneFilter": {}}
    if acquisition_time_range:
        base_data["sceneFilter"]["acquisitionFilter"] = {
            "start": acquisition_time_range[0].isoformat(),
            "end": acquisition_time_range[1].isoformat(),
        }
    if cloud_cover_range:
        base_data["sceneFilter"]["cloudCoverFilter"] = {
            "min": cloud_cover_range[0],
            "max": cloud_cover_range[1],
            "includeUnknown": False,
        }
    if bbox:
        base_data["sceneFilter"]["spatialFilter"] = {
            "filterType": "mbr",
            "lowerLeft": {"longitude": bbox[0], "latitude": bbox[1]},
            "upperRight": {"longitude": bbox[2], "latitude": bbox[3]},
        }
    if metadata_filter:
        base_data["sceneFilter"]["metadataFilter"] = metadata_filter

    starting_number = 1
    results = []
    while True:
        cur_data = base_data.copy()
        cur_data["startingNumber"] = starting_number
        cur_data["maxResults"] = self.pagination_size
        response_dict = self.request("scene-search", cur_data)
        if response_dict is None:
            raise APIException("No response from API")
        data = response_dict["data"]
        results.extend(data["results"])
        if data["recordsReturned"] < self.pagination_size:
            break
        starting_number += self.pagination_size

    return results

get_scene_metadata 

get_scene_metadata(dataset_name: str, entity_id: str) -> dict[str, Any]

Get detailed metadata for a scene.

Parameters:

Name Type Description Default
dataset_name str

the dataset name in which to search

 required
entity_id str

the entity ID of the scene

 required

Returns:

Type Description
dict[str, Any]

full scene metadata
Source code in rslearn/utils/m2m_api.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def get_scene_metadata(self, dataset_name: str, entity_id: str) -> dict[str, Any]:
    """Get detailed metadata for a scene.

    Args:
        dataset_name: the dataset name in which to search
        entity_id: the entity ID of the scene

    Returns:
        full scene metadata
    """
    response_dict = self.request(
        "scene-metadata",
        {
            "datasetName": dataset_name,
            "entityId": entity_id,
            "metadataType": "full",
        },
    )
    if response_dict is None:
        raise APIException("No response from API")
    return response_dict["data"]

get_downloadable_products 

get_downloadable_products(dataset_name: str, entity_id: str) -> list[dict[str, Any]]

Get the downloadable products for a given scene.

Parameters:

Name Type Description Default
dataset_name str

the dataset name

 required
entity_id str

the entity ID of the scene

 required

Returns:

Type Description
list[dict[str, Any]]

list of downloadable products
Source code in rslearn/utils/m2m_api.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def get_downloadable_products(
    self, dataset_name: str, entity_id: str
) -> list[dict[str, Any]]:
    """Get the downloadable products for a given scene.

    Args:
        dataset_name: the dataset name
        entity_id: the entity ID of the scene

    Returns:
        list of downloadable products
    """
    data = {"datasetName": dataset_name, "entityIds": [entity_id]}
    response_dict = self.request("download-options", data)
    if response_dict is None:
        raise APIException("No response from API")
    return response_dict["data"]

get_download_url 

get_download_url(entity_id: str, product_id: str) -> str

Get the download URL for a given product.

Parameters:

Name Type Description Default
entity_id str

the entity ID of the product

 required
product_id str

the product ID of the product

 required

Returns:

Type Description
str

the download URL
Source code in rslearn/utils/m2m_api.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def get_download_url(self, entity_id: str, product_id: str) -> str:
    """Get the download URL for a given product.

    Args:
        entity_id: the entity ID of the product
        product_id: the product ID of the product

    Returns:
        the download URL
    """
    label = str(uuid.uuid4())
    data = {
        "downloads": [
            {"label": label, "entityId": entity_id, "productId": product_id}
        ]
    }
    response_dict = self.request("download-request", data)
    if response_dict is None:
        raise APIException("No response from API")
    response = response_dict["data"]

    # Check if URL is immediately available in the response
    if response.get("availableDownloads"):
        return response["availableDownloads"][0]["url"]

    # Otherwise poll download-retrieve until the URL is ready
    while True:
        # We sleep first because we have observed that if we immediately call
        # download-retrieve it sometimes has a response that includes neither
        # available nor requested list.
        time.sleep(10)
        response_dict = self.request("download-retrieve", {"label": label})
        if response_dict is None:
            raise APIException("No response from API")
        response = response_dict["data"]
        if len(response["available"]) > 0:
            return response["available"][0]["url"]
        if len(response["requested"]) == 0:
            logger.debug(
                f"Response to download-retrieve did not include our requested download under label {label}: {response}"
            )
            raise APIException("Did not get download URL")
        if response["requested"][0].get("url"):
            return response["requested"][0]["url"]