s1_orbit: return S1A/S1B when parsing filename (#104)

* s1_orbit: return S1A/S1B when parsing filename

This fixes #103 to get the right orbit file.

Currently we are returning "A" or "B" as a "sensor_id", then searching if the orbit file has that identifier (the single letter, which all orbit files have).

This 1. changes the parser to return the full S1A/S1B, 2. calls it the "mission_id" to match the language in the product spec https://sentinels.copernicus.eu/documents/247904/351187/Copernicus_Sentinels_POD_Service_File_Format_Specification

3. does further simplficication and cleanup of logic to make pylance/flake8 errors go away

* add unit test for diffing the missions in one orbit dir

* use fstring to add EOF extension

* codacy trailing whitespace

src/s1reader/s1_orbit.py

@@ -37,17 +37,17 @@ def download_orbit(safe_file: str, orbit_dir: str):
     check_internet_connection()
 
     # Parse info from SAFE file name
-    sensor_id, _, start_time, end_time, _ = parse_safe_filename(safe_file)
+    mission_id, _, start_time, end_time, _ = parse_safe_filename(safe_file)
 
     # Find precise orbit first
-    orbit_dict = get_orbit_dict(sensor_id, start_time,
+    orbit_dict = get_orbit_dict(mission_id, start_time,
                                 end_time, 'AUX_POEORB')
     # If orbit dict is empty, find restituted orbits
     if orbit_dict is None:
-        orbit_dict = get_orbit_dict(sensor_id, start_time,
+        orbit_dict = get_orbit_dict(mission_id, start_time,
                                     end_time, 'AUX_RESORB')
     # Download orbit file
-    orbit_file = os.path.join(orbit_dir, orbit_dict["orbit_name"] + '.EOF')
+    orbit_file = os.path.join(orbit_dir, f"{orbit_dict['orbit_name']}.EOF")
     if not os.path.exists(orbit_file):
         download_orbit_file(orbit_dir, orbit_dict['orbit_url'])
 
@@ -76,9 +76,9 @@ def parse_safe_filename(safe_filename):
 
     Returns
     -------
-    List of [sensor_id, mode_id, start_datetime,
+    List of [mission_id, mode_id, start_datetime,
                 end_datetime, abs_orbit_num]
-       sensor_id: sensor identifier (S1A or S1B)
+       mission_id: sensor identifier (S1A or S1B)
        mode_id: mode/beam (e.g. IW)
        start_datetime: acquisition start datetime
        stop_datetime: acquisition stop datetime
@@ -88,29 +88,51 @@ def parse_safe_filename(safe_filename):
     ---------
     parse_safe_filename('S1A_IW_SLC__1SDV_20150224T114043_20150224T114111_004764_005E86_AD02.SAFE')
     returns
-    ['A', 'IW', datetime.datetime(2015, 2, 24, 11, 40, 43),\
+    ['S1A', 'IW', datetime.datetime(2015, 2, 24, 11, 40, 43),\
     datetime.datetime(2015, 2, 24, 11, 41, 11), 4764]
     '''
 
     safe_name = os.path.basename(safe_filename)
-    sensor_id = safe_name[2]
+    mission_id = safe_name[:3]
     sensor_mode = safe_name[4:6]
     start_datetime = datetime.datetime.strptime(safe_name[17:32],
                                                 FMT)
     end_datetime = datetime.datetime.strptime(safe_name[33:48],
                                               FMT)
     abs_orb_num = int(safe_name[49:55])
 
-    return [sensor_id, sensor_mode, start_datetime, end_datetime, abs_orb_num]
+    return [mission_id, sensor_mode, start_datetime, end_datetime, abs_orb_num]
 
 
-def get_orbit_dict(sensor_id, start_time, end_time, orbit_type):
+def get_file_name_tokens(zip_path: str) -> [str, list[datetime.datetime]]:
+    '''Extract swath platform ID and start/stop times from SAFE zip file path.
+
+    Parameters
+    ----------
+    zip_path: list[str]
+        List containing orbit path strings.
+        Orbit files required to adhere to naming convention found here:
+        https://sentinels.copernicus.eu/documents/247904/351187/Copernicus_Sentinels_POD_Service_File_Format_Specification
+
+    Returns
+    -------
+    mission_id: ('S1A', 'S1B')
+    orbit_path : str
+        Path the orbit file.
+    t_swath_start_stop: list[datetime.datetime]
+        Swath start/stop times
+    '''
+    mission_id, _, start_time, end_time, _ = parse_safe_filename(zip_path)
+    return mission_id, [start_time, end_time]
+
+
+def get_orbit_dict(mission_id, start_time, end_time, orbit_type):
     '''
     Query Copernicus GNSS API to find latest orbit file
     Parameters
     ----------
-    sensor_id: str
-        Sentinel satellite identifier ('A' or 'B')
+    mission_id: str
+        Sentinel satellite identifier ('S1A' or 'S1B')
     start_time: datetime object
         Sentinel start acquisition time
     end_time: datetime object
@@ -140,7 +162,7 @@ def get_orbit_dict(sensor_id, start_time, end_time, orbit_type):
     pad_end_time = end_time + pad_30_min
     new_start_time = pad_start_time.strftime('%Y-%m-%dT%H:%M:%S')
     new_end_time = pad_end_time.strftime('%Y-%m-%dT%H:%M:%S')
-    query_string = f"startswith(Name,'S1{sensor_id}') and substringof('{orbit_type}',Name) " \
+    query_string = f"startswith(Name,'{mission_id}') and substringof('{orbit_type}',Name) " \
                    f"and ContentDate/Start lt datetime'{new_start_time}' and ContentDate/End gt datetime'{new_end_time}'"
     query_params = {'$top': 1, '$orderby': 'ContentDate/Start asc',
                     '$filter': query_string}
@@ -193,32 +215,6 @@ def download_orbit_file(output_folder, orbit_url):
     return orbit_file
 
 
-def get_file_name_tokens(zip_path: str) -> [str, list[datetime.datetime]]:
-    '''Extract swath platform ID and start/stop times from SAFE zip file path.
-
-    Parameters
-    ----------
-    zip_path: list[str]
-        List containing orbit path strings. Orbit files required to adhere to
-        naming convention found here:
-        https://s1qc.asf.alaska.edu/aux_poeorb/
-
-    Returns
-    -------
-    platform_id: ('S1A', 'S1B')
-    orbit_path : str
-        Path the orbit file.
-    t_swath_start_stop: list[datetime.datetime]
-        Swath start/stop times
-    '''
-    platform_id, _, start_time, end_time, _ = parse_safe_filename(zip_path)
-    return platform_id,[start_time, end_time]
-
-
-# lambda to check if file exists if desired sat_id in basename
-item_valid = lambda item, sat_id: os.path.isfile(item) and sat_id in os.path.basename(item)
-
-
 def get_orbit_file_from_dir(zip_path: str, orbit_dir: str, auto_download: bool = False) -> str:
     '''Get orbit state vector list for a given swath.
 
@@ -252,12 +248,6 @@ def get_orbit_file_from_dir(zip_path: str, orbit_dir: str, auto_download: bool =
             print(f"{orbit_dir} not found, creating directory.")
             os.makedirs(orbit_dir, exist_ok=True)
 
-    # extract platform id, start and end times from swath file name
-    platform_id, t_swath_start_stop = get_file_name_tokens(zip_path)
-
-    # initiate output
-    orbit_file = ''
-
     # search for orbit file
     orbit_file_list = glob.glob(os.path.join(orbit_dir, 'S1*.EOF'))
 
@@ -292,23 +282,24 @@ def get_orbit_file_from_list(zip_path: str, orbit_file_list: list) -> str:
     Returns:
     --------
     orbit_file : str
-        Path to the orbit file.
+        Path to the orbit file, or an empty string if no orbit file was found.
     '''
-
     # check the existence of input file path and directory
     if not os.path.exists(zip_path):
         raise FileNotFoundError(f"{zip_path} does not exist")
 
     # extract platform id, start and end times from swath file name
-    platform_id, t_swath_start_stop = get_file_name_tokens(zip_path)
+    mission_id, t_swath_start_stop = get_file_name_tokens(zip_path)
 
     # initiate output
     orbit_file_final = ''
 
     # search for orbit file
     for orbit_file in orbit_file_list:
         # check if file validity
-        if not item_valid(orbit_file, platform_id):
+        if not os.path.isfile(orbit_file):
+            continue
+        if mission_id not in os.path.basename(orbit_file):
             continue
 
         # get file name and extract state vector start/end time strings
@@ -332,4 +323,3 @@ def get_orbit_file_from_list(zip_path: str, orbit_file_list: list) -> str:
         warnings.warn(msg)
 
     return orbit_file_final
-

tests/test_orbit.py

@@ -6,12 +6,32 @@
 
 from s1reader.s1_orbit import get_orbit_file_from_dir
 
+
 def test_get_orbit_file(test_paths):
     orbit_file = get_orbit_file_from_dir(test_paths.safe, test_paths.orbit_dir)
 
     expected_orbit_path = f'{test_paths.orbit_dir}/{test_paths.orbit_file}'
     assert orbit_file == expected_orbit_path
 
+
+def test_get_orbit_file_multi_mission(tmp_path):
+    orbit_a = tmp_path / "S1A_OPER_AUX_POEORB_OPOD_20210314T131617_V20191007T225942_20191009T005942.EOF"
+    orbit_a.write_text("")
+    orbit_b = tmp_path / "S1B_OPER_AUX_POEORB_OPOD_20210304T232500_V20191007T225942_20191009T005942.EOF"
+    orbit_b.write_text("")
+
+    zip_path = tmp_path / "zips"
+    zip_path.mkdir()
+    zip_a = zip_path / "S1A_IW_SLC__1SDV_20191008T005936_20191008T010003_018377_0229E5_909C.zip"
+    zip_a.write_text("")
+    zip_b = zip_path / "S1B_IW_SLC__1SDV_20191008T005936_20191008T010003_018377_0229E5_909C.zip"
+    zip_b.write_text("")
+
+    # Test S1A zip file
+    assert get_orbit_file_from_dir(zip_a, orbit_dir=tmp_path) == str(orbit_a)
+    assert get_orbit_file_from_dir(zip_b, orbit_dir=tmp_path) == str(orbit_b)
+
+
 def test_orbit_datetime(bursts):
     # pad in seconds used in orbit_reader
     pad = datetime.timedelta(seconds=60)