Update orbit module

sar_antarctica/nci/filesystem.py

@@ -0,0 +1,47 @@
+from pathlib import Path
+
+from sar_antarctica.nci.preparation.orbits import find_orbits
+
+
+def get_orbits_nci(orbit_type: str | None, sensor: str) -> list[Path]:
+    """For a given orbit type and sensor, compile the relevant orbit files
+
+    Parameters
+    ----------
+    orbit_type : str | None
+        One of 'POE', 'RES', or None. If None, both POE and RES orbits will be included
+    sensor : str
+        Sensor (e.g. S1A or S1B) to search. Typically extracted from the scene ID
+
+    Returns
+    -------
+    list[Path]
+        List of orbit files on NCI matching search criteria
+
+    Raises
+    ------
+    ValueError
+        Invalid orbit type. Must be one of 'POE', 'RES' or None
+    """
+
+    # Constants for NCI
+    S1_DIR = Path("/g/data/fj7/Copernicus/Sentinel-1/")
+    POE_DIR = "POEORB"
+    RES_DIR = "RESORB"
+
+    if orbit_type == "POE":
+        orbit_type_directories = [POE_DIR]
+    elif orbit_type == "RES":
+        orbit_type_directories = [RES_DIR]
+    elif orbit_type is None:
+        orbit_type_directories = [RES_DIR, POE_DIR]
+    else:
+        raise ValueError("orbit_type must be one of 'POE', 'RES', or None")
+
+    nci_orbit_directories = [
+        S1_DIR / orbit_dir / sensor for orbit_dir in orbit_type_directories
+    ]
+
+    orbits = find_orbits(nci_orbit_directories)
+
+    return orbits

sar_antarctica/nci/preparation/create_config.py

@@ -3,7 +3,12 @@
 from pyroSAR import identify
 import rasterio
 
-from sar_antarctica.nci.preparation.scenes import find_scene_file_from_id
+from sar_antarctica.nci.filesystem import get_orbits_nci
+
+from sar_antarctica.nci.preparation.scenes import (
+    find_scene_file_from_id,
+    parse_scene_file_sensor,
+)
 from sar_antarctica.nci.preparation.orbits import find_latest_orbit_for_scene
 from sar_antarctica.nci.preparation.dem import get_cop30_dem_for_bounds
 
@@ -47,6 +52,15 @@ def write_file_paths(
 @click.argument("scene_id", nargs=1)
 @click.argument("scene_config", nargs=1)
 def main(scene_id: str, scene_config: str):
+    """Generate a configuration file for a scene ID
+
+    Parameters
+    ----------
+    scene_id : str
+        ID of scene to process
+    scene_config : str
+        where to store the output configuration file
+    """
     print(f"Processing scene: {scene_id} \n")
 
     # Set the data path for outputs
@@ -59,7 +73,9 @@ def main(scene_id: str, scene_config: str):
     scene_file = find_scene_file_from_id(scene_id)
 
     # Identify location of latest orbit file on GADI
-    latest_poe_file = find_latest_orbit_for_scene(scene_id, orbit_type="POE")
+    scene_sensor = parse_scene_file_sensor(scene_id)
+    poe_orbits = get_orbits_nci("POE", scene_sensor)
+    latest_poe_file = find_latest_orbit_for_scene(scene_id, poe_orbits)
 
     # Identify bounds of scene and use bounding box to build DEM
     scene = identify(str(scene_file))

sar_antarctica/nci/preparation/orbits.py

@@ -1,17 +1,159 @@
 from datetime import datetime
 from pathlib import Path
 import re
-from typing import Optional
 
 from sar_antarctica.nci.preparation.scenes import (
     parse_scene_file_dates,
-    parse_scene_file_sensor,
 )
 
-# Constants for NCI
-S1_DIR = Path("/g/data/fj7/Copernicus/Sentinel-1/")
-POE_DIR = "POEORB"
-RES_DIR = "RESORB"
+
+def find_latest_orbit_for_scene(scene_id: str, orbit_files: list[Path]) -> Path:
+    """Identifies the most recent orbit file available for a given scene, based
+    on the scene's start and end date.
+
+        Parameters
+        ----------
+        scene_id : str
+            Sentinel-1 scene ID
+            e.g. S1A_EW_GRDM_1SDH_20220612T120348_20220612T120452_043629_053582_0F6
+        orbit_directories : list[Path]
+            directories to search for the latest orbit file
+
+        Returns
+        -------
+        Path
+            File path to latest orbit file on NCI
+    """
+
+    scene_start, scene_stop = parse_scene_file_dates(scene_id)
+
+    latest_orbit = find_latest_orbit_covering_window(
+        orbit_files, scene_start, scene_stop
+    )
+
+    return latest_orbit
+
+
+def find_orbits(directories: list[Path], extension: str = ".EOF") -> list[Path]:
+    """_summary_
+
+    Parameters
+    ----------
+    directories : list[Path]
+        A list of directories to search for orbit files
+    extension : str, optional
+        The extension for orbit files, by default ".EOF"
+
+    Returns
+    -------
+    list[Path]
+        A list of orbit files for every directory searched
+    """
+
+    matching_files = []
+    for directory in directories:
+        if directory.is_dir():
+            matching_files.extend(directory.glob(f"*{extension}"))
+    return matching_files
+
+
+def find_latest_orbit_covering_window(
+    orbit_files: list[Path], window_start: datetime, window_stop: datetime
+) -> Path:
+    """For a list of orbit files, finds the file with the latest publish date that
+    covers the time window specified by a start and stop datetime.
+
+    Parameters
+    ----------
+    orbit_files : list[Path]
+        A list of orbit files
+    window_start : datetime
+        The start of the window the orbit must cover
+    window_stop : datetime
+        The end of the window the orbit must cover
+
+    Returns
+    -------
+    Path
+        the orbit file with the latest published date that covers the window
+    """
+
+    orbits_files_in_window = filter_orbits_to_cover_time_window(
+        orbit_files, window_start, window_stop
+    )
+
+    latest_orbit = filter_orbits_to_latest(orbits_files_in_window)
+
+    return latest_orbit
+
+
+def filter_orbits_to_cover_time_window(
+    orbit_files: list[Path],
+    window_start: datetime,
+    window_stop: datetime,
+) -> list[dict[str, Path | datetime]]:
+    """For a list of orbit files, finds all files that cover the time window
+    specified by a start and stop datetime.
+
+    Parameters
+    ----------
+    orbit_files : list[Path]
+        A list of orbit files
+    window_start : datetime
+        The start of the window the orbit must cover
+    window_stop : datetime
+        The end of the window the orbit must cover
+
+    Returns
+    -------
+    list[dict[str, Path | datetime]]
+        _description_
+
+    Raises
+    ------
+    ValueError
+        _description_
+    """
+
+    matching_orbits = []
+    for orbit_file in orbit_files:
+        orbit_published, orbit_start, orbit_stop = parse_orbit_file_dates(orbit_file)
+
+        if window_start >= orbit_start and window_stop <= orbit_stop:
+            orbit_metadata = {"orbit": orbit_file, "published_date": orbit_published}
+            matching_orbits.append(orbit_metadata)
+
+    if not matching_orbits:
+        raise ValueError("No orbits were found within the specified time widow.")
+
+    return matching_orbits
+
+
+def filter_orbits_to_latest(orbits: list[dict[str, Path | datetime]]) -> Path:
+    """For a list of orbit files and published dates, find the orbit file with the latest published date.
+
+    Parameters
+    ----------
+    orbits : list[dict[str, Path  |  datetime]]
+        List of orbits, where each orbit is a dictionary of
+        {"orbit": Path, "published_date": datetime}
+
+    Returns
+    -------
+    Path
+        The path to the orbit file with the latest published date
+
+    Raises
+    ------
+    ValueError
+        _description_
+    """
+
+    latest_orbit = max(orbits, key=lambda x: x["published_date"])
+
+    latest_orbit_file = latest_orbit["orbit"]
+
+    return latest_orbit_file
 
 
 def parse_orbit_file_dates(orbit_file_name: str) -> tuple[datetime, datetime, datetime]:
@@ -55,72 +197,3 @@ def parse_orbit_file_dates(orbit_file_name: str) -> tuple[datetime, datetime, da
     stop_date = datetime.strptime(match.group("stop_date"), "%Y%m%dT%H%M%S")
 
     return (published_date, start_date, stop_date)
-
-
-def find_latest_orbit_for_scene(
-    scene_id: str, orbit_type: Optional[str] = None
-) -> Path:
-    """Identifies the most recent orbit file available for a given scene, based
-    on the scene's start and end date.
-
-    Parameters
-    ----------
-    scene_id : str
-        Sentinel-1 scene ID
-        e.g. S1A_EW_GRDM_1SDH_20220612T120348_20220612T120452_043629_053582_0F6
-    orbit_type : Optional[str], optional
-        Any of "POE" for POE orbits, "RES" for RES orbits, or None, by default None
-
-    Returns
-    -------
-    Path
-        Full file path to latest orbit file on NCI
-
-    Raises
-    ------
-    ValueError
-        orbit_type must be one of "POE", "RES" or None
-    ValueError
-        No valid orbit file was found
-    """
-
-    scene_start, scene_stop = parse_scene_file_dates(scene_id)
-    scene_sensor = parse_scene_file_sensor(scene_id)
-
-    relevant_orbits = []
-
-    if orbit_type == "POE":
-        orbit_directories = [POE_DIR]
-    elif orbit_type == "RES":
-        orbit_directories = [RES_DIR]
-    elif orbit_type is None:
-        orbit_directories = [RES_DIR, POE_DIR]
-    else:
-        raise ValueError("orbit_type must be one of 'POE', 'RES', or None")
-
-    # Find all orbits for the sensor that fall within the date range of the scene
-    for orbit_dir in orbit_directories:
-        orbit_dir_path = S1_DIR / orbit_dir
-        orbit_files_path = orbit_dir_path / scene_sensor
-        orbit_files = orbit_files_path.glob("*.EOF")
-
-        for orbit_file in orbit_files:
-
-            orbit_published, orbit_start, orbit_stop = parse_orbit_file_dates(
-                orbit_file
-            )
-
-            # Check if scene falls within orbit
-            if scene_start >= orbit_start and scene_stop <= orbit_stop:
-                orbit_metadata = (orbit_file, orbit_dir, orbit_published)
-                relevant_orbits.append(orbit_metadata)
-
-    # If relevant_orbits is empty, set latest_orbit to None
-    latest_orbit = max(relevant_orbits, key=lambda x: x[2]) if relevant_orbits else None
-
-    if latest_orbit is None:
-        raise ValueError("No valid orbit was found.")
-    else:
-        latest_orbit_file = latest_orbit[0]
-
-    return latest_orbit_file