Merge branch 'main' into NPI-4067-check-sp3-column-alignment

Author: treefern

gnssanalysis/gn_datetime.py

@@ -17,7 +17,7 @@
 logger = logging.getLogger(__name__)
 
 
-def gpsweekD(yr, doy, wkday_suff=False):
+def derive_gps_week(year: Union[int, str], day_of_year: Union[int, str], weekday_suffix: bool = False) -> str:
     """
     Convert year, day-of-year to GPS week format: WWWWD or WWWW
     Based on code from Kristine Larson's gps.py
@@ -32,30 +32,37 @@ def gpsweekD(yr, doy, wkday_suff=False):
     """
 
     # Set up the date and time variables
-    yr = int(yr)
-    doy = int(doy)
-    dt = _datetime.strptime(f"{yr}-{doy:03d} 01", "%Y-%j %H")
+    year = int(year)
+    day_of_year = int(day_of_year)
+    dt: _datetime = _datetime.strptime(f"{year}-{day_of_year:03d} 01", "%Y-%j %H")
 
     wkday = dt.weekday() + 1
 
     if wkday == 7:
         wkday = 0
 
-    mn, dy, hr = dt.month, dt.day, dt.hour
+    month, day, hour = dt.month, dt.day, dt.hour
 
-    if mn <= 2:
-        yr = yr - 1
-        mn = mn + 12
+    if month <= 2:
+        year = year - 1
+        month = month + 12
 
-    JD = _np.floor(365.25 * yr) + _np.floor(30.6001 * (mn + 1)) + dy + hr / 24.0 + 1720981.5
-    GPS_wk = int(_np.floor((JD - 2444244.5) / 7.0))
+    julian_day: float = _np.floor(365.25 * year) + _np.floor(30.6001 * (month + 1)) + day + hour / 24.0 + 1720981.5
+    GPS_wk = int(_np.floor((julian_day - 2444244.5) / 7.0))
 
-    if wkday_suff:
+    if weekday_suffix:
         return str(GPS_wk) + str(wkday)
     else:
         return str(GPS_wk)
 
 
+def gpsweekD(yr, doy, wkday_suff=False) -> str:
+    """
+    TODO DEPRECATED. Legacy wrapper for the above
+    """
+    return derive_gps_week(yr, doy, wkday_suff)
+
+
 class GPSDate:
     """
     Representation of datetime that provides easy access to
@@ -94,24 +101,49 @@ def as_datetime(self) -> _datetime:
         return _pd.Timestamp(self._internal_dt64).to_pydatetime()
 
     @property
-    def yr(self) -> str:
+    def as_date(self) -> _date:
+        """Convert to Python `date` object."""
+        return self._internal_dt64.astype(_date)
+
+    @property
+    def year(self) -> str:
         """Year"""
         return self.as_datetime.strftime("%Y")
 
+    @property  # For backwards compatibility
+    def yr(self) -> str:
+        """Year DEPRECATED, use GPSDate.year"""
+        return self.year
+
     @property
-    def dy(self) -> str:
+    def day_of_year(self) -> str:
         """Day of year"""
         return self.as_datetime.strftime("%j")
 
+    @property  # For backwards compatibility
+    def dy(self) -> str:
+        """Day of year. DEPRECATED, use GPSDate.day_of_year"""
+        return self.day_of_year
+
     @property
-    def gpswk(self) -> str:
+    def gps_week(self) -> str:
         """GPS week"""
-        return gpsweekD(self.yr, self.dy, wkday_suff=False)
+        return derive_gps_week(self.yr, self.dy, weekday_suffix=False)
+
+    @property  # For backwards compatibility
+    def gpswk(self) -> str:
+        """GPS week. DEPRECATED, use GPSDate.gps_week"""
+        return self.gps_week
 
     @property
-    def gpswkD(self) -> str:
+    def gps_week_and_day_of_week(self) -> str:
         """GPS week with weekday suffix"""
-        return gpsweekD(self.yr, self.dy, wkday_suff=True)
+        return derive_gps_week(self.yr, self.dy, weekday_suffix=True)
+
+    @property  # For backwards compatibility
+    def gpswkD(self) -> str:
+        """GPS week with weekday suffix. DEPRECATED, use GPSDate.gps_week_and_day_of_week"""
+        return self.gps_week_and_day_of_week
 
     @property
     def next(self):
@@ -128,24 +160,27 @@ def __str__(self) -> str:
         return str(self._internal_dt64)
 
 
-def dt2gpswk(dt, wkday_suff=False, both=False) -> Union[str, tuple[str, str]]:
+def datetime_to_gps_week(dt: _datetime, wkday_suff: bool = False) -> str:
     """
-    Convert the given datetime object to a GPS week (option to include day suffix)
+    Convert given datetime object to GPS week e.g. '2350', optionally including day suffix e.g. '23500' (Sunday)
+
+    :param datetime dt: datetime for which to calculate a GPS week (and optionally day)
+    :returns str: intput datetime expressed as a GPS week (wwww) or GPS week with day (wwwwd)
     """
     yr = dt.strftime("%Y")
     doy = dt.strftime("%j")
-    if not both:
-        return gpsweekD(yr, doy, wkday_suff=wkday_suff)
-    else:
-        return gpsweekD(yr, doy, wkday_suff=False), gpsweekD(yr, doy, wkday_suff=True)
+    return derive_gps_week(yr, doy, weekday_suffix=wkday_suff)
 
 
-# TODO DEPRECATED
-def gpswkD2dt(gpswkD: str) -> _datetime:
+def dt2gpswk(dt: _datetime, wkday_suff: bool = False, both: bool = False) -> Union[str, tuple[str, str]]:
     """
-    DEPRECATED. This is a compatibility wrapper. Use gps_week_day_to_datetime() instead.
+    TODO DEPRECATED. Please use datetime_to_gps_week()
     """
-    return gps_week_day_to_datetime(gpswkD)
+
+    if both:
+        return (datetime_to_gps_week(dt, wkday_suff=False), datetime_to_gps_week(dt, wkday_suff=True))
+    else:
+        return datetime_to_gps_week(dt, wkday_suff=wkday_suff)
 
 
 def gps_week_day_to_datetime(gps_week_and_weekday: str) -> _datetime:
@@ -178,6 +213,14 @@ def gps_week_day_to_datetime(gps_week_and_weekday: str) -> _datetime:
     return dt_64.astype(_datetime)
 
 
+# TODO DEPRECATED
+def gpswkD2dt(gpswkD: str) -> _datetime:
+    """
+    DEPRECATED. This is a compatibility wrapper. Use gps_week_day_to_datetime() instead.
+    """
+    return gps_week_day_to_datetime(gpswkD)
+
+
 def yydoysec2datetime(
     arr: Union[_np.ndarray, _pd.Series, list], recenter: bool = False, as_j2000: bool = True, delimiter: str = ":"
 ) -> _np.ndarray:

gnssanalysis/gn_download.py

@@ -336,7 +336,7 @@ def generate_long_filename(
     start_epoch: _datetime.datetime,  # YYYYDDDHHMM
     end_epoch: _datetime.datetime = None,
     timespan: _datetime.timedelta = None,  # LEN
-    solution_type: str = "",  # TTT
+    solution_type: str = "",  # TTT # TODO look at updating to formalised SolutionType
     sampling_rate: str = "15M",  # SMP
     version: str = "0",  # V
     project: str = "EXP",  # PPP, e.g. EXP, OPS
@@ -432,12 +432,15 @@ 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"
-            product_filename = f"{prefix}{gps_date.gpswkD}_{hour}.{file_ext}.Z" if solution_type == "ULT" else \
-                f"{prefix}{gps_date.gpswkD}.{file_ext}.Z"
+            product_filename = (
+                f"{prefix}{gps_date.gps_week_and_day_of_week}_{hour}.{file_ext}.Z"
+                if solution_type == "ULT"
+                else f"{prefix}{gps_date.gps_week_and_day_of_week}.{file_ext}.Z"
+            )
     return product_filename, gps_date, reference_start
 
 
@@ -495,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)
@@ -739,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
 
@@ -752,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)
@@ -793,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
@@ -843,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")
@@ -865,14 +868,14 @@ def download_product_from_cddis(
         project=project_type,
     )
     logging.debug(
-        f"Generated filename: {product_filename}, with GPS Date: {gps_date.gpswkD} and reference: {reference_start}"
+        f"Generated filename: {product_filename}, with GPS Date: {gps_date.gps_week_and_day_of_week} and reference: {reference_start}"
     )
 
     ensure_folders([download_dir])
     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}")
@@ -888,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
@@ -923,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_frame.py

@@ -54,7 +54,9 @@ def get_frame_of_day(
     elif isinstance(itrf_path_or_df, str):
         output = _gn_io.sinex._get_snx_vector_gzchunks(filename=itrf_path_or_df, block_name="SOLUTION/ESTIMATE")
         if output is None:
-            raise Exception(f"Output from _get_snx_vector_gzchunks() was None! Filepath: {itrf_path_or_df}")
+            raise Exception(
+                f"Failed to extract SOLUTION/ESTIMATE from Sinex. Check that file was valid: {itrf_path_or_df}"
+            )
     else:
         raise ValueError(f"itrf_path_or_df must be a pandas DataFrame or str, got: {type(itrf_path_or_df)}")
 

gnssanalysis/gn_io/erp.py

@@ -594,7 +594,9 @@ def erp_outfile(datetime_epoch: datetime.datetime, output_dir: pathlib.Path):
     # Restrict the amount data we output
     resampled_df = resampled_df[resampled_df["MJD"] > mjd - 3]
 
-    gps_date = _gn_datetime.gpsweekD(datetime_epoch.strftime("%Y"), datetime_epoch.strftime("%j"), wkday_suff=True)
+    gps_date = _gn_datetime.derive_gps_week(
+        datetime_epoch.strftime("%Y"), datetime_epoch.strftime("%j"), weekday_suffix=True
+    )
     file_suffix = f'_{int(int(str(mjd).split(".")[1].ljust(2,"0"))*0.24):02}'
     file_name = f"igu{gps_date}{file_suffix}.erp"