Last active
October 8, 2021 11:42
-
-
Save system123/3c2c3896c0d542999b15a02d19f82cb8 to your computer and use it in GitHub Desktop.
Download Sentinel 1 POE and RES orbit files from aux.sentinel1.eo.esa.int
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
"""Get Sentinel-1 orbit file(s) from ESA archive website""" | |
import argparse | |
import logging | |
import os | |
import re | |
import sys | |
from datetime import datetime | |
from pyroSAR.examine import ExamineSnap | |
import requests | |
from pathlib import Path | |
from typing import Union | |
from lxml import html | |
def download_file(url: str, directory: Union[Path, str] = '.', chunk_size=None, retries=1, backoff_factor=1) -> str: | |
"""Download a file | |
Args: | |
url: URL of the file to download | |
directory: Directory location to place files into | |
chunk_size: Size to chunk the download into | |
retries: Number of retries to attempt | |
backoff_factor: Factor for calculating time between retries | |
Returns: | |
download_path: The path to the downloaded file | |
""" | |
logging.info(f'Downloading {url}') | |
try: | |
download_path = Path(directory) / url.split("/")[-1] | |
except AttributeError: | |
raise requests.exceptions.InvalidURL(f'Invalid URL provided: {url}') | |
with requests.get(url, stream=True) as s: | |
if s.status_code == 200: | |
with open(download_path, "wb") as f: | |
for chunk in s.iter_content(chunk_size=chunk_size): | |
if chunk: | |
f.write(chunk) | |
return str(download_path) | |
def _get_asf_orbit_url(orbit_type, platform, timestamp): | |
ot = orbit_type.upper().split('_')[1] | |
search_url = f"http://aux.sentinel1.eo.esa.int/{ot}/{timestamp[:4]}/{timestamp[4:6]}/{timestamp[6:8]}/" | |
response = requests.get(search_url) | |
if response.status_code == 200: | |
tree = html.fromstring(response.content) | |
file_list = [file for file in tree.xpath('//a[@href]//@href') | |
if file.startswith(platform) and file.endswith('.EOF')] | |
d1 = 0 | |
best = None | |
for file in file_list: | |
file = file.strip() | |
t = re.split('_', file.replace('T', '').replace('V', '')) | |
if len(t) > 7: | |
start = t[6] | |
end = t[7].replace('.EOF', '') | |
if start < timestamp < end: | |
d = ((int(timestamp) - int(start)) + (int(end) - int(timestamp))) / 2 | |
if d > d1: | |
best = file | |
d1 = d | |
if best is not None: | |
return search_url + best | |
return None | |
def downloadSentinelOrbitFile( | |
granule: str, directory = None, orbit_types=('AUX_POEORB', 'AUX_RESORB') | |
): | |
"""Download a Sentinel-1 Orbit file | |
Args: | |
granule: Granule name to find an orbit file for | |
directory: Directory to save the orbit files into | |
orbit_types: Iterable of orbit file types to attempt to download, in order of preference | |
Returns: Tuple of: | |
orbit_file: The downloaded orbit file | |
""" | |
if not directory: | |
try: | |
auxdatapath = ExamineSnap().auxdatapath | |
except AttributeError: | |
auxdatapath = os.path.join(os.path.expanduser('~'), '.snap', 'auxdata') | |
osvdir = os.path.join(auxdatapath, 'Orbits', 'Sentinel-1') | |
outdir_poe = os.path.join(osvdir, 'POEORB') | |
outdir_res = os.path.join(osvdir, 'RESORB') | |
else: | |
outdir_poe = directory | |
outdir_res = directory | |
outdir = { | |
'AUX_POEORB': outdir_poe, | |
'AUX_RESORB': outdir_res | |
} | |
orbit_file = None | |
for orbit_type in orbit_types: | |
platform = granule[0:3] | |
time_stamps = re.split('_+', granule)[4:6] | |
orbit_url = _get_asf_orbit_url(orbit_type.lower(), platform, time_stamps[0].replace('T', '')) | |
try: | |
orbit_file = download_file(orbit_url, directory=outdir[orbit_type]) | |
break | |
except: | |
continue | |
return orbit_file |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment