The **gen_new_list.py** generates a list of new product IDs in the DHuS database added since the last run

of this script and saves them. Otherwise, generates list of IDs from the last 30 days. The list of IDs is then

stored in a file.

**Authentication**: Basic auth is resolved automatically by the Requests library by reading a **~/.netrc** file. Make sure

to set up the correct entry (DHuS URL) there.

**Configuration file**: The **sentinel_config.yml** contains the URL to the Sentinel data hub host.

**Command line arguments**: The command line options supersede the configuration file settings. Run help to list all

configurable parameters: `./gen_new_list.py -h`

The **register_stack.py** script fetches Sentinel 1, 2, 3 and 5P metadata from a data hub

and transforms it into a [STAC](https://stacspec.org/en) format. It can publish the results to a STAC catalogue.

The transformation is done by imported [stactools](https://github.com/stac-utils/stactools) modules.

There are several ways to configure the script's behaviour:

**Preconfigured mappings**: The **stac_collections.py** file contains general constants that should be updated

by developers if necessary. You can modify the options if needed.

**Configuration file**: The **sentinel_config.yml** contains the URLs to the Sentinel data hub host and the STAC catalogue

host, log file prefixes and path to a location to save the data.

**Command line arguments**: The command line options supersede the configuration file settings. Run help to list all

configurable parameters: `./register_stack.py -h`

**Authentication**: Basic auth is resolved automatically by the Requests library by reading a **~/.netrc** file. Make sure

to set up the correct entries (Sentinel and STAC host URL) there.

import argparse

import os

from datetime import datetime, timedelta

import requests

import yaml

DEBUG = False

DATE_FORMAT = "%Y-%m-%dT%H:%M:%S.%f"

CONFIG_FILE = "sentinel_config.yml"

LIST_FILENAME = "gen_new_list_processed.txt"

TIMESTAMP_FILENAME = "gen_new_list_timestamp.txt"

def parse_arguments():

"""

parser = argparse.ArgumentParser(

description='Generates a list of Sentinel products recently published at DHuS endpoint.'

'Example usage: ./gen_new_list.py')

parser.add_argument('-r',

'--dryRun',

required=False,

action='store_true',

help='Do not store results')

parser.add_argument('-f',

"--fromTimestamp",

required=False,

type=lambda d: datetime.strptime(d, '%Y-%m-%d'),

help="Alternative start date to use instead of stored timestamp.")

parser.add_argument('-e',

'--sentinelHost',

required=False,

help='URL of server to fetch Sentinel data from, for example https://dhr1.cesnet.cz/.'

'Overwrites SENTINEL_HOST configuration option.')

parser.add_argument('-d',

'--debug',

required=False,

action="store_true",

help='Enable to enable extended progress messages.')

args = parser.parse_args()

return args

def print_debug(msg):

"""

if DEBUG:

print(msg)

def read_configuration():

"""

with open(CONFIG_FILE, "r") as f:

return yaml.safe_load(f)

def get_timestamp(local_dir):

"""

timestamp_filepath = os.path.join(local_dir, TIMESTAMP_FILENAME)

fallback_timestamp = (datetime.now() - timedelta(days=31)).strftime(DATE_FORMAT)[:-3]

if not os.path.isfile(timestamp_filepath) or not os.path.getsize(timestamp_filepath):

return fallback_timestamp

with open(timestamp_filepath, "r") as f:

content = f.read().strip()

try:

timestamp = content[:-3]

print_debug(f"Using stored timestamp {timestamp}")

return timestamp

except ValueError:

print("Timestamp file exists but is formatted incorrectly")

return fallback_timestamp

def create_missing_dir(dir_path):

"""

if not os.path.exists(dir_path):

os.makedirs(dir_path, exist_ok=True)

def fetch_products(host_url, timestamp):

"""

fetched_count = 100

page_size = 100

result = []

# header is included in response

while fetched_count + 1 >= page_size:

url = f"{host_url}/odata/v1/Products"

params = {

'$format': 'text/csv',

'$select': 'Id',

'$skip': len(result),

'$top': page_size,

'$filter': f"CreationDate ge datetime'{timestamp}'"

}

response = requests.get(url, params=params)

if not response.ok:

raise Exception(f"Request to fetch products file failed with {response.status_code}.\n{response.text}")

product_ids = response.text.splitlines()[1:]

result.extend(product_ids)

fetched_count = len(product_ids)

print_debug(f"Fetched {len(result)} products.")

return result

def load_cached_products(local_dir):

"""

filepath = os.path.join(local_dir, LIST_FILENAME)

if not os.path.exists(filepath):

return []

else:

with open(filepath, "r") as f:

return f.readlines()

def store_new_timestamp(local_dir, new_timestamp):

"""

timestamp_filepath = os.path.join(local_dir, TIMESTAMP_FILENAME)

with open(timestamp_filepath, 'w') as f:

f.write(new_timestamp)

def store_new_list(local_dir, missing_products):

"""

list_filepath = os.path.join(local_dir, LIST_FILENAME)

with open(list_filepath, 'w') as f:

f.write("\n".join(missing_products))

def main():

args = parse_arguments()

config = read_configuration()

global DEBUG

DEBUG = args.debug

sentinel_host = args.sentinelHost or config.get("SENTINEL_HOST")

if not sentinel_host:

raise Exception("SENTINEL_HOST is not defined and sentinelHost parameter not passed!")

local_dir = config.get("LOCAL_DIR")

timestamp = args.fromTimestamp or get_timestamp(local_dir)

new_timestamp = datetime.now().strftime(DATE_FORMAT)

fetched_products = fetch_products(sentinel_host, timestamp)

stored_products = load_cached_products(local_dir)

missing_products = list(set(fetched_products) - set(stored_products))

print_debug(f"There are {len(missing_products)} unprocessed products.")

if not args.dryRun:

store_new_timestamp(local_dir, new_timestamp)

store_new_list(local_dir, missing_products)

if __name__ == "__main__":

main()