202507 antex file discovery

.gitignore

@@ -16,3 +16,5 @@ __pycache__
 **/tmp_test*/*
 **.profile
 **.pstat
+src/prx/sp3/sp3_files/*
+src/prx/precise_corrections/sp3/sp3_files/2023/001/IGS0OPSFIN_20230010000_01D_15M_ORB.SP3

src/prx/precise_corrections/antex/antex_file_discovery.py

@@ -0,0 +1,162 @@
+from ftplib import FTP
+import ftplib
+from pathlib import Path
+import re
+
+import georinex
+import pandas as pd
+import urllib
+import fnmatch
+
+from prx import converters
+from prx import util
+from prx.util import timestamp_to_mid_day
+
+"""
+Logic overview:
+- The code first extracts the GPS week from the provided RINEX observation file.
+- It then identifies the most recent ANTEX file available both locally and remotely.
+- If both local and remote ANTEX files are the same, the local version is reused, as the local database is considered up to date.
+- Otherwise, the latest remote file is downloaded and used.
+- Note: If the most recent ANTEX file has a GPS week that is earlier than the observation file's GPS week 
+(e.g., the remote database has not yet been updated), a warning is displayed to indicate that the most recent ANTEX file 
+was still used despite being older than the observation file.
+"""
+
+log = util.get_logger(__name__)
+
+atx_filename = "igs20_????.atx"
+
+
+def date_to_gps_week(date: pd.Timestamp):
+    """
+    Convert a Timestamp to GPS week
+
+    GPS week starts on 01/06/1980 (sunday)
+    """
+    gps_epoch = pd.Timestamp("1980-01-06T00:00:00Z")
+    delta = date.tz_localize("UTC") - gps_epoch
+    gps_week = delta.days // 7
+    return gps_week
+
+
+def extract_gps_week(filename: str) -> int:
+    """
+    Extract GPS week from filenames like 'igsxx_WWWW.atx'.
+    """
+    match = re.search(r"igs...(\d{4})\.atx", filename)
+    if match:
+        return int(match.group(1))
+    return -1
+
+
+def atx_file_database_folder():
+    """
+    Returns the path to the folder where ATX database files are stored.
+    """
+    db_folder = (
+        util.prx_repository_root() / "src/prx/precise_corrections/antex/atx_files"
+    )
+    db_folder.mkdir(exist_ok=True)
+    return db_folder
+
+
+def find_latest_local_antex_file(db_folder=atx_file_database_folder()):
+    candidates = list(db_folder.glob(atx_filename))
+    if not candidates:
+        return None
+    return max(candidates, key=lambda c: extract_gps_week(c))
+
+
+def list_ftp_directory(server, folder):
+    ftp = FTP(server)
+    ftp.login()
+    ftp.cwd(folder)
+    dir_list = []
+    ftp.dir(dir_list.append)
+    return [c.split()[-1].strip() for c in dir_list]
+
+
+def fetch_latest_remote_antex_file():
+    """
+    List the ANTEX files available online and returns the latest
+    """
+    server = "gssc.esa.int"
+    remote_folder = "/igs/station/general"
+    candidates = list_ftp_directory(server, remote_folder)
+    candidates = [c for c in candidates if fnmatch.fnmatch(c, atx_filename)]
+    if not candidates:
+        return None
+    return max(candidates, key=lambda c: extract_gps_week(c))
+
+
+def check_online_availability(file: str, folder: Path) -> Path | None:
+    """
+    Need to keep the same inputs as try_downloading_sp3_ftp, in order to be able to use `unittest.mock.patch` in tests
+    """
+    server = "gssc.esa.int"
+    remote_folder = "/igs/station/general"
+    ftp = FTP(server)
+    ftp.login()
+    ftp.cwd(remote_folder)
+    try:
+        ftp.size(file)
+        return folder.joinpath(file)
+    except ftplib.error_perm:
+        log.warning(f"{file} not available on {server}")
+        return None
+
+
+def try_downloading_atx_ftp(file: str, folder: Path):
+    """
+    Download the wanted remote file
+    """
+    server = "gssc.esa.int"
+    remote_folder = "/igs/station/general"
+    ftp_file = f"ftp://{server}/{remote_folder}/{file}"
+    local_file = folder / file
+    urllib.request.urlretrieve(ftp_file, local_file)
+    if not local_file.exists():
+        log.warning(f"Could not download {ftp_file}")
+        return None
+    log.info(f"Downloaded ANTEX file {ftp_file}")
+    return local_file
+
+
+def get_atx_file(date: pd.Timestamp, db_folder=atx_file_database_folder()):
+    gps_week = date_to_gps_week(date)
+    latest_atx_local = find_latest_local_antex_file(db_folder)
+    latest_atx_remote = fetch_latest_remote_antex_file()
+    if latest_atx_remote == latest_atx_local:
+        return latest_atx_local
+    elif latest_atx_remote and (
+        not latest_atx_local
+        or extract_gps_week(latest_atx_remote) > extract_gps_week(latest_atx_local)
+    ):
+        # Download the latest file
+        atx_file = try_downloading_atx_ftp(latest_atx_remote, db_folder)
+    if atx_file is not None:
+        if gps_week > extract_gps_week(atx_file.name):
+            log.warning(
+                f"No ANTEX file found for the target GPS week {gps_week} — using the most recent available instead."
+            )
+        return atx_file
+
+    raise FileNotFoundError("No file ANTEX found locally or online.")
+
+
+def discover_or_download_atx_file(observation_file_path=Path()):
+    """
+    Returns the path to a valid antes file (local or downloaded) corresponding to the observation file.
+    """
+    log.info(f"Finding auxiliary files for {observation_file_path} ...")
+    rinex_3_obs_file = converters.anything_to_rinex_3(observation_file_path)
+    header = georinex.rinexheader(rinex_3_obs_file)
+
+    t_start = timestamp_to_mid_day(
+        util.rinex_header_time_string_2_timestamp_ns(header["TIME OF FIRST OBS"])
+        - pd.Timedelta(200, unit="milliseconds")
+    )
+
+    atx_file = get_atx_file(t_start)
+    return atx_file

src/prx/precise_corrections/antex/test/datasets/test_antex_file_discovery.py

@@ -0,0 +1,139 @@
+import os
+from pathlib import Path
+import shutil
+import georinex
+import pandas as pd
+import pytest
+from unittest.mock import patch
+
+import prx
+from prx import util
+from prx.precise_corrections.antex import antex_file_discovery as atx
+
+
+@pytest.fixture
+def set_up_test():
+    test_directory = Path(f"./tmp_test_directory_{__name__}").resolve()
+    if test_directory.exists():
+        # Make sure the expected files has not been generated before and is still on disk due to e.g. a previous
+        # test run having crashed:
+        shutil.rmtree(test_directory)
+    os.makedirs(test_directory)
+    test_obs_file = test_directory.joinpath("TLSE00FRA_R_20230010100_10S_01S_MO.crx.gz")
+    # local_database = atx.atx_file_folder(pd.Timestamp('2023-01-01'), Path('src/prx/precise_corrections/atx/test/datasets'))
+
+    shutil.copy(
+        util.prx_repository_root()
+        / f"src/prx/test/datasets/TLSE_2023001/{test_obs_file.name}",
+        test_obs_file,
+    )
+    assert test_obs_file.exists()
+
+    yield {"test_obs_file": test_obs_file}
+    shutil.rmtree(test_directory)
+
+
+def test_extract_gps_week():
+    filenames = [
+        "igs20_2134.atx",
+        "igs_10.atx",
+        "igs14_abc.atx",
+    ]
+    expected_returns = [2134, -1, -1]
+    gps_week = [atx.extract_gps_week(f) for f in filenames]
+    assert gps_week == expected_returns
+
+
+def test_download_if_not_local(set_up_test):
+    """
+    Tests that the ANTEX file is downloaded when no local file is available,
+    but a valid remote file exists.
+
+    Scenario :
+    - No local ANTEX file available
+    - A remote ANTEX file exists
+    -> The function should trigger the download of the remote file.
+    """
+    obs_file = set_up_test["test_obs_file"]
+    header = georinex.rinexheader(obs_file)
+    t_start = util.rinex_header_time_string_2_timestamp_ns(
+        header["TIME OF FIRST OBS"]
+    ) - pd.Timedelta(200, unit="milliseconds")
+
+    db_folder = set_up_test["test_obs_file"].parent
+    downloaded_atx = atx.get_atx_file(t_start, db_folder)
+
+    # Ensure the file was downloaded and exists
+    assert downloaded_atx is not None
+    assert downloaded_atx.exists()
+
+
+def test_download_if_remote_is_newer(set_up_test):
+    """
+    Tests that the ANTEX file is downloaded when the remote file has a newer GPS week
+    than the local one.
+
+    Scenario :
+    - local file = igs20_2370.atx
+    - remote file = igs20_2375.atx
+    -> The function should download igs20_2375.atx
+    """
+    latest_local = "igs20_2370.atx"
+    latest_remote = "igs20_2375.atx"
+    date = pd.Timestamp("2025-01-01 12:00:00")
+    db_folder = set_up_test["test_obs_file"].parent
+
+    with (
+        patch(  # replace download function by online availability check
+            "prx.precise_corrections.antex.antex_file_discovery.try_downloading_atx_ftp",
+            new=prx.precise_corrections.antex.antex_file_discovery.check_online_availability,
+        ),
+        patch(  # simulate that the latest local file found is 'igs20_2370.atx'
+            "prx.precise_corrections.antex.antex_file_discovery.find_latest_local_antex_file",
+            return_value=latest_local,
+        ),
+        patch(  # simulate that the latest remote file found is 'igs20_2375.atx'
+            "prx.precise_corrections.antex.antex_file_discovery.fetch_latest_remote_antex_file",
+            return_value=latest_remote,
+        ),
+    ):
+        result = atx.get_atx_file(date, db_folder)
+        assert result is not None
+        assert result.name == latest_remote
+
+
+def test_skip_download_if_same_week(set_up_test):
+    """
+    Tests that the ANTEX file is not downloaded when the remote and local files
+    are from the same GPS week.
+
+    Scenario:
+    - local file = igs20_2375.atx
+    - remote file = igs20_2375.atx
+    → The function should skip download and return the local file
+    """
+    latest_local = "igs20_2375.atx"
+    latest_remote = "igs20_2375.atx"
+    date = pd.Timestamp("2025-01-01 12:00:00")
+    db_folder = set_up_test["test_obs_file"].parent
+
+    with (
+        patch(  # mock download function
+            "prx.precise_corrections.antex.antex_file_discovery.try_downloading_atx_ftp"
+        ) as mock_download,
+        patch(  # simulate that the latest local file found is 'igs20_2375.atx'
+            "prx.precise_corrections.antex.antex_file_discovery.find_latest_local_antex_file",
+            return_value=latest_local,
+        ),
+        patch(  # simulate that the latest remote file found is 'igs20_2375.atx'
+            "prx.precise_corrections.antex.antex_file_discovery.fetch_latest_remote_antex_file",
+            return_value=latest_remote,
+        ),
+    ):
+        result = atx.get_atx_file(date, db_folder)
+
+        assert result is not None
+        # Ensure that the download function was not called
+        mock_download.assert_not_called()
+        # Ensure the returned file is the local one
+        assert result == latest_local