Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def query_api(self, area, start_date, end_date, qargs):
if not start_date and not end_date:
start_date = self.start_date
end_date = self.end_date
else:
start_date = datetime.strptime(start_date, "%d/%m/%Y").date()
end_date = datetime.strptime(end_date, "%d/%m/%Y").date()
keywords = {}
for k in qargs:
k = k.split(':')
keywords[k[0]] = k[1]
try:
self.api.query(get_coordinates(area), start_date, end_date, **keywords)
except SentinelAPIError, e:
logger.info(e.msg)
exit(1)
search_kwargs["cloudcoverpercentage"] = (0, cloud)
if query is not None:
search_kwargs.update((x.split("=") for x in query))
if geometry is not None:
search_kwargs["area"] = geojson_to_wkt(read_geojson(geometry))
if uuid is not None:
uuid_list = [x.strip() for x in uuid]
products = {}
for productid in uuid_list:
try:
products[productid] = api.get_product_odata(productid)
except SentinelAPIError as e:
if "Invalid key" in e.msg:
logger.error("No product with ID '%s' exists on server", productid)
exit(1)
else:
raise
elif name is not None:
search_kwargs["identifier"] = name[0] if len(name) == 1 else "(" + " OR ".join(name) + ")"
products = api.query(order_by=order_by, limit=limit, **search_kwargs)
else:
start = start or "19000101"
end = end or "NOW"
products = api.query(date=(start, end), order_by=order_by, limit=limit, **search_kwargs)
if footprints is True:
footprints_geojson = api.to_geojson(products)
with open(os.path.join(path, "search_footprints.geojson"), "w") as outfile:
Returns
-------
bool
True if online, False if in LTA
"""
# Check https://scihub.copernicus.eu/userguide/ODataAPI#Products_entity for more information
url = urljoin(self.api_url, "odata/v1/Products('{}')/Online/$value".format(id))
with self.session.get(url, auth=self.session.auth, timeout=self.timeout) as r:
if r.status_code == 200 and r.text == "true":
return True
elif r.status_code == 200 and r.text == "false":
return False
else:
raise SentinelAPIError(
"Could not verify whether product {} is online".format(id), r
)
auth=self.session.auth,
headers={"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"},
timeout=self.timeout,
)
_check_scihub_response(response)
# store last status code (for testing)
self._last_response = response
# parse response content
try:
json_feed = response.json()["feed"]
if json_feed["opensearch:totalResults"] is None:
# We are using some unintended behavior of the server that a null is
# returned as the total results value when the query string was incorrect.
raise SentinelAPIError(
"Invalid query string. Check the parameters and format.", response
)
total_results = int(json_feed["opensearch:totalResults"])
except (ValueError, KeyError):
raise SentinelAPIError("API response not valid. JSON decoding failed.", response)
products = json_feed.get("entry", [])
# this verification is necessary because if the query returns only
# one product, self.products will be a dict not a list
if isinstance(products, dict):
products = [products]
return products, total_results
msg = "Invalid API response."
try:
msg = response.headers["cause-message"]
except:
try:
msg = response.json()["error"]["message"]["value"]
except:
if not response.text.strip().startswith("{"):
try:
h = html2text.HTML2Text()
h.ignore_images = True
h.ignore_anchors = True
msg = h.handle(response.text).strip()
except:
pass
api_error = SentinelAPIError(msg, response)
# Suppress "During handling of the above exception..." message
# See PEP 409
api_error.__cause__ = None
raise api_error
# store last status code (for testing)
self._last_response = response
# parse response content
try:
json_feed = response.json()["feed"]
if json_feed["opensearch:totalResults"] is None:
# We are using some unintended behavior of the server that a null is
# returned as the total results value when the query string was incorrect.
raise SentinelAPIError(
"Invalid query string. Check the parameters and format.", response
)
total_results = int(json_feed["opensearch:totalResults"])
except (ValueError, KeyError):
raise SentinelAPIError("API response not valid. JSON decoding failed.", response)
products = json_feed.get("entry", [])
# this verification is necessary because if the query returns only
# one product, self.products will be a dict not a list
if isinstance(products, dict):
products = [products]
return products, total_results
The response from the server as a `requests.Response` object.
"""
def __init__(self, msg=None, response=None):
self.msg = msg
self.response = response
def __str__(self):
return "HTTP status {0} {1}: {2}".format(
self.response.status_code,
self.response.reason,
("\n" if "\n" in self.msg else "") + self.msg,
)
class SentinelAPILTAError(SentinelAPIError):
""" Error when retrieving a product from the Long Term Archive
Attributes
----------
msg: str
The error message.
response: requests.Response
The response from the server as a `requests.Response` object.
"""
def __init__(self, msg=None, response=None):
self.msg = msg
self.response = response
class InvalidChecksumError(Exception):