Merge pull request #89 from GeoscienceAustralia/NPI-3470-misc-clarity-and-hint-improvements

Npi 3470 misc clarity and hint improvements

Author: treefern

gnssanalysis/gn_download.py

@@ -432,7 +432,7 @@ def generate_product_filename(
         )
     else:
         if file_ext.lower() == "snx":
-            product_filename = f"igs{gps_date.yr[2:]}P{gps_date.gpswk}.snx.Z"
+            product_filename = f"igs{gps_date.year[2:]}P{gps_date.gps_week}.snx.Z"
         else:
             hour = f"{reference_start.hour:02}"
             prefix = "igs" if solution_type == "FIN" else "igr" if solution_type == "RAP" else "igu"
@@ -498,7 +498,7 @@ def attempt_ftps_download(
     download_dir: _Path,
     ftps: _ftplib.FTP_TLS,
     filename: str,
-    type_of_file: str = None,
+    type_of_file: Optional[str] = None,
     if_file_present: str = "prompt_user",
 ) -> Union[_Path, None]:
     """Attempt download of file (filename) given the ftps client object (ftps) to chosen location (download_dir)
@@ -742,7 +742,7 @@ def download_file_from_cddis(
     max_retries: int = 3,
     decompress: bool = True,
     if_file_present: str = "prompt_user",
-    note_filetype: str = None,
+    note_filetype: Optional[str] = None,
 ) -> Union[_Path, None]:
     """Downloads a single file from the CDDIS ftp server
 
@@ -755,7 +755,7 @@ def download_file_from_cddis(
     :param str note_filetype: How to label the file for STDOUT messages, defaults to None
     :raises e: Raise any error that is run into by ftplib
     :return _Path or None: The pathlib.Path of the downloaded file (or decompressed output of it). Returns None if the
-        file already existed and was skipped.
+        file already existed and was skipped, or if the download failed.
     """
     with ftp_tls(CDDIS_FTP) as ftps:
         ftps.cwd(ftp_folder)
@@ -796,7 +796,7 @@ def download_file_from_cddis(
     raise Exception("Failed to download file or raise exception. Some logic is broken.")
 
 
-def download_multiple_files_from_cddis(files: List[str], ftp_folder: str, output_folder: _Path) -> None:
+def download_multiple_files_from_cddis(files: list[str], ftp_folder: str, output_folder: _Path) -> None:
     """Downloads multiple files in a single folder from cddis in a thread pool.
 
     :param files: List of str filenames
@@ -846,7 +846,7 @@ def download_product_from_cddis(
         # We do this because the weekly files are released/dated as Sunday of each GPS week.
         start_epoch_as_gps_date = GPSDate(start_epoch)
         # Get GPS week number *without* DayOfWeek suffix (therefore start of the GPS Week), then convert back to datetime
-        start_epoch = gps_week_day_to_datetime(f"{start_epoch_as_gps_date.gpswk}")
+        start_epoch = gps_week_day_to_datetime(f"{start_epoch_as_gps_date.gps_week}")
         timespan = _datetime.timedelta(days=7)
 
     logging.info("Attempting CDDIS Product download/s")
@@ -875,7 +875,7 @@ def download_product_from_cddis(
     download_filepaths = []
     with ftp_tls(CDDIS_FTP) as ftps:
         try:
-            ftps.cwd(f"gnss/products/{gps_date.gpswk}")
+            ftps.cwd(f"gnss/products/{gps_date.gps_week}")
         except _ftplib.all_errors as e:
             logging.warning(f"{reference_start} too recent")
             logging.warning(f"ftp_lib error: {e}")
@@ -891,11 +891,11 @@ def download_product_from_cddis(
                 version=version,
                 project=project_type,
             )
-            ftps.cwd(f"gnss/products/{gps_date.gpswk}")
+            ftps.cwd(f"gnss/products/{gps_date.gps_week}")
 
             all_files = ftps.nlst()
             if not (product_filename in all_files):
-                logging.warning(f"{product_filename} not in gnss/products/{gps_date.gpswk} - too recent")
+                logging.warning(f"{product_filename} not in gnss/products/{gps_date.gps_week} - too recent")
                 raise FileNotFoundError
 
         # reference_start will be changed in the first run through while loop below
@@ -926,7 +926,7 @@ def download_product_from_cddis(
                     download_filepaths.append(
                         download_file_from_cddis(
                             filename=product_filename,
-                            ftp_folder=f"gnss/products/{gps_date.gpswk}",
+                            ftp_folder=f"gnss/products/{gps_date.gps_week}",
                             output_folder=download_dir,
                             if_file_present=if_file_present,
                             note_filetype=file_ext,

gnssanalysis/gn_io/nanu.py

@@ -2,14 +2,17 @@
 import logging as _logging
 import os as _os
 from typing import Union as _Union
+from datetime import datetime
 
 import numpy as _np
 import pandas as _pd
 
 from .. import gn_io as _gn_io
 
 
-def nanu_path_to_id(nanu_path):
+def nanu_path_to_id(nanu_path: str) -> str:
+    # TODO some examples would be good here.
+
     dir, _, filename = nanu_path.rpartition(_os.sep)
     nanu_id, _, extension = filename.partition(".")  # get filename without extension
     if nanu_id == "nanu":  # celestrak naming convention
@@ -45,17 +48,17 @@ def parse_nanu(nanu_bytes: bytes) -> dict:
     return output_dict
 
 
-def read_nanu(path_or_bytes: _Union[str, bytes]) -> dict:
+def read_nanu(path: str) -> dict:
     """A parser for Notice Advisory to Navstar Users (NANU) files.
     Assumes there is only one message per file, that starts with '1.'
 
     :param _Union[str, bytes] path_or_bytes: path to nanu file or a bytes object
     :return dict: nanu values with parameter names as keys
     """
-    nanu_bytes = _gn_io.common.path2bytes(path_or_bytes)
+    nanu_bytes = _gn_io.common.path2bytes(path)
     output_dict = {}
-    output_dict["FILEPATH"] = path_or_bytes  # TODO change to pathlib
-    output_dict["NANU ID"] = nanu_path_to_id(path_or_bytes)
+    output_dict["FILEPATH"] = path  # TODO change to pathlib
+    output_dict["NANU ID"] = nanu_path_to_id(path)
     output_dict["CONTENT"] = nanu_bytes
     output_dict.update(parse_nanu(nanu_bytes))
     return output_dict
@@ -71,15 +74,22 @@ def collect_nanus_to_df(glob_expr: str) -> _pd.DataFrame:
     return _pd.DataFrame(read_nanu(n) for n in nanus_list if n is not None)
 
 
-def get_bad_sv_from_nanu_df(nanu_df: _pd.DataFrame, datetime: _Union[_np.datetime64, str], offset_days: int) -> list:
-    """A simple function that analyses an input dataframe NANU collection and outputs a list of SVs that should be excluded for the entered epoch+offset
+def get_bad_sv_from_nanu_df(
+    nanu_df: _pd.DataFrame, up_to_epoch: _Union[_np.datetime64, datetime, str], offset_days: int
+) -> list:
+    """A simple function that analyses an input dataframe NANU collection and outputs a list of SVs that should be
+    excluded for the entered epoch+offset
 
-    :param _pd.DataFrame nanu_df: a dataframe returned by the collect_nanus_to_df, effectively a _pd.DataFrame call on a list of parsed dicts or a parsed dict
-    :param _Union[_np.datetime64, str] datetime: epoch to analyse NANUs up to
+    :param _pd.DataFrame nanu_df: a dataframe returned by the collect_nanus_to_df, effectively a _pd.DataFrame call on
+        a list of parsed dicts or a parsed dict
+    :param _Union[_np.datetime64, datetime, str] up_to_epoch: epoch to analyse NANUs up to
     :param int offset_days: an offset or a length of a planned processing session in days
-    :return list: a list of SVs that should not be used for the specified timeperiod. FIXME Potentially needs to be int?
+    :return list[str]: a list of SVs that should not be used for the specified timeperiod. FIXME Potentially needs to
+        be int?
     """
-    datetime = datetime if isinstance(datetime, _np.datetime64) else _np.datetime64(datetime)
+    up_to_epoch_datetime64: _np.datetime64 = (
+        up_to_epoch if isinstance(up_to_epoch, _np.datetime64) else _np.datetime64(up_to_epoch)
+    )
 
     columns_new = [
         "FILEPATH",
@@ -129,25 +139,29 @@ def get_bad_sv_from_nanu_df(nanu_df: _pd.DataFrame, datetime: _Union[_np.datetim
     dt_df = _pd.concat([df.drop(labels=columns_date, axis=1), dates], axis=1)
 
     events_already_started = (
-        (dt_df["START CALENDAR DATE"] <= (datetime + offset_days))
-        | (dt_df["UNUSABLE START CALENDAR DATE"] <= (datetime + offset_days))
-        | (dt_df["LAUNCH START CALENDAR DATE"] <= (datetime + offset_days))
+        (dt_df["START CALENDAR DATE"] <= (up_to_epoch_datetime64 + offset_days))
+        | (dt_df["UNUSABLE START CALENDAR DATE"] <= (up_to_epoch_datetime64 + offset_days))
+        | (dt_df["LAUNCH START CALENDAR DATE"] <= (up_to_epoch_datetime64 + offset_days))
     )
     dt_valid_df = dt_df[events_already_started]
 
     prns_last_nanu_to_date = dt_valid_df.PRN.astype(float).drop_duplicates(keep="last").index
 
     all_the_last_msgs = dt_df.loc[prns_last_nanu_to_date]
 
+    # Filter maneuver related NANU messages down to those with an end date in the future, or no end date at all:
     last_selected = all_the_last_msgs[
-        ((all_the_last_msgs["STOP CALENDAR DATE"] >= datetime) | all_the_last_msgs["STOP CALENDAR DATE"].isna())
+        (
+            (all_the_last_msgs["STOP CALENDAR DATE"] >= up_to_epoch_datetime64)
+            | all_the_last_msgs["STOP CALENDAR DATE"].isna()
+        )
         & (all_the_last_msgs["NANU TYPE"] != "USABINIT")
     ]
 
     if last_selected.empty:
-        return []
+        return []  # No NANUs currently in effect
 
-    _logging.info(msg="NANUs in affect are:\n" + "\n".join(last_selected.FILEPATH.to_list()))
+    _logging.info(msg="NANUs in effect are:\n" + "\n".join(last_selected.FILEPATH.to_list()))
 
     sel_idx = last_selected.index.values
     msg_gaps = dt_df.loc[sel_idx.min() : sel_idx.max()]