Download Sentinel 1 POE and RES orbit files from aux.sentinel1.eo.esa.int

download_orbit.py

"""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