Index

GPS data sources package providing unified interface to different providers.

BaseGpsSource

Bases: ABC

Base class for GPS data sources providing standardized interface.

Source code in src/geepers/gps_sources/base.py

class BaseGpsSource(ABC):
    """Base class for GPS data sources providing standardized interface."""

    def timeseries_many(
        self,
        /,
        ids: Iterable[str] | None = None,
        bbox: tuple[float, float, float, float] | None = None,
        mask: gpd.GeoSeries | None = None,
        frame: Literal["ENU", "XYZ"] = "ENU",
        start_date: str | None = None,
        end_date: str | None = None,
        zero_by: Literal["mean", "start", "none"] = "mean",
        download_if_missing: bool = True,
        *,
        max_workers: int = 8,
        skip_errors: bool = True,
    ):
        if bbox is None and mask is None and ids is None:
            msg = "Must provide ids, bbox or mask"
            raise ValueError(msg)
        gdf_stations = self.stations(bbox=bbox, mask=mask)
        if bbox is not None or mask is not None or ids is None:
            ids = gdf_stations["id"]
        # Index by id for O(1) metadata lookups in `_load_one`
        station_rows = gdf_stations.set_index("id")

        # Function to load one id
        def _load_one(sid: str) -> pd.DataFrame | None:
            try:
                df = self.timeseries(
                    sid,
                    frame=frame,
                    start_date=start_date,
                    end_date=end_date,
                    zero_by=zero_by,
                    download_if_missing=download_if_missing,
                )
            except (requests.HTTPError, urllib.error.HTTPError) as e:
                # Some ids in the station lists have no data file on the
                # server (e.g. UNR grid points with too little data)
                if not skip_errors:
                    raise
                logger.warning("Skipping %s: %s", sid, e)
                return None
            df.insert(0, "id", sid)  # keep id as a column for melt/pivot
            row = station_rows.loc[sid]
            for col in ("lon", "lat", "alt", "geometry"):
                df[col] = row[col]
            return df

        # (Optional) parallel map
        if max_workers:
            dfs = thread_map(
                _load_one, ids, max_workers=max_workers, desc="Loading GPS data"
            )
        else:
            dfs = [_load_one(sid) for sid in tqdm(ids)]

        n_failed = sum(df is None for df in dfs)
        if n_failed:
            logger.warning("Skipped %d of %d ids with no data", n_failed, len(dfs))
        dfs = [df for df in dfs if df is not None]
        if not dfs:
            msg = "No time series could be loaded"
            raise ValueError(msg)
        big = pd.concat(dfs, ignore_index=True)

        return gpd.GeoDataFrame(big, geometry="geometry", crs="EPSG:4326")

    def __init__(self, cache_dir: PathOrStr | None = None):
        """Initialize the GPS data source.

        Parameters
        ----------
        cache_dir : PathOrStr, optional
            Base directory to store cached data.
            Default is None, which uses `utils.get_cache_dir()`.
            Subclasses create directories under this base directory.

        """
        if cache_dir is None:
            self._base_cache_dir = utils.get_cache_dir()
        else:
            self._base_cache_dir = Path(cache_dir)
        self._subdir = self.__class__.__name__.lower().replace("source", "")

    @property
    def _cache_dir(self) -> Path:
        """Cache directory for this source, created on first access.

        Creation is deferred so that instantiating a source (e.g. at module
        import time) has no filesystem side effects.
        """
        cache_dir = self._base_cache_dir / self._subdir
        cache_dir.mkdir(exist_ok=True, parents=True)
        return cache_dir

    @abstractmethod
    def timeseries(
        self,
        station_id: str,
        /,
        frame: Literal["ENU", "XYZ"] = "ENU",
        start_date: str | None = None,
        end_date: str | None = None,
        zero_by: Literal["mean", "start", "none"] = "mean",
        download_if_missing: bool = True,
    ) -> pd.DataFrame:
        """Load GPS station time series data.

        Parameters
        ----------
        station_id : str
            The station identifier.
        frame : {"ENU", "XYZ"}, optional
            Coordinate frame for the data. Default is "ENU".
        start_date : str, optional
            Start date for data filtering (ISO format).
        end_date : str, optional
            End date for data filtering (ISO format).
        zero_by : Literal["mean", "start"], optional
            How to zero the data. Either "mean" or "start".
        download_if_missing : bool, optional
            Whether to download data if not found locally.

        Returns
        -------
        pd.DataFrame
            DataFrame validated against StationObservationSchema.

        """

    @abstractmethod
    def _read_station_data(self) -> gpd.GeoDataFrame:
        """Read raw station data from the source.

        Returns
        -------
        gpd.GeoDataFrame
            GeoDataFrame with station metadata including lon, lat, alt columns.

        """

    def stations(
        self,
        bbox: tuple[float, float, float, float] | None = None,
        mask: gpd.GeoSeries | None = None,
    ) -> gpd.GeoDataFrame:
        """Get GPS stations, optionally filtered by spatial bounds.

        Parameters
        ----------
        bbox : tuple[float, float, float, float], optional
            Bounding box as (west, south, east, north) in degrees.
        mask : gpd.GeoSeries, optional
            Spatial mask to filter stations.

        Returns
        -------
        gpd.GeoDataFrame
            GeoDataFrame with station metadata including lon, lat, alt columns.

        """
        # Read data from source
        gdf = self._read_station_data()

        # Apply spatial filters and validate
        gdf = self._apply_spatial_filters(gdf, bbox, mask)

        return gdf

    def _apply_spatial_filters(
        self,
        gdf: gpd.GeoDataFrame,
        bbox: tuple[float, float, float, float] | None = None,
        mask: gpd.GeoSeries | None = None,
    ) -> gpd.GeoDataFrame:
        """Apply spatial filters to a GeoDataFrame.

        Parameters
        ----------
        gdf : gpd.GeoDataFrame
            Input GeoDataFrame to filter.
        bbox : tuple[float, float, float, float], optional
            Bounding box as (west, south, east, north) in degrees.
        mask : gpd.GeoSeries, optional
            Spatial mask to filter stations.

        Returns
        -------
        gpd.GeoDataFrame
            Filtered GeoDataFrame.

        """
        # Apply bbox filter (coordinate slicing: much faster than a
        # geometric clip for point layers)
        if bbox is not None:
            west, south, east, north = bbox
            gdf = gdf.cx[west:east, south:north]

        # Apply mask filter
        if mask is not None:
            gdf = gdf[gdf.geometry.within(mask.union_all())]

        # Reset index for cleaner output
        gdf = gdf.reset_index(drop=True)

        # Validate basic point schema
        return PointSchema.validate(gdf, lazy=True)

    def _filter_by_date(
        self,
        df: pd.DataFrame,
        start_date: str | None = None,
        end_date: str | None = None,
    ) -> pd.DataFrame:
        """Filter DataFrame by date range.

        Accepts tz-aware date strings (e.g. UTC ISO timestamps); the tz is
        dropped so the bound compares against the tz-naive ``date`` column.
        """

        def _naive(value: str) -> pd.Timestamp:
            ts = pd.to_datetime(value)
            return ts.tz_localize(None) if ts.tzinfo is not None else ts

        if start_date:
            df = df[df["date"] >= _naive(start_date)]
        if end_date:
            df = df[df["date"] <= _naive(end_date)]
        return df

    def _zero_data(
        self,
        df: pd.DataFrame,
        zero_by: Literal["mean", "start", "none"] | None = "mean",
        columns: list[str] | None = None,
    ) -> pd.DataFrame:
        """Zero the data in a DataFrame ("none"/None leaves it as-is)."""
        if columns is None:
            columns = ["east", "north", "up"]
        if zero_by is None or zero_by.lower() == "none":
            return df
        if zero_by.lower() == "mean":
            mean_val = df[columns].mean()
            df.loc[:, columns] -= mean_val
        elif zero_by.lower() == "start":
            start_val = df[columns].iloc[:10].mean()
            df.loc[:, columns] -= start_val
        else:
            msg = "zero_by must be 'mean', 'start', or 'none'"
            raise ValueError(msg)
        return df

    def coordinates(self, station_id: str) -> tuple[float, float, float]:
        """Get coordinates for a single station.

        Parameters
        ----------
        station_id : str
            The station identifier.

        Returns
        -------
        tuple[float, float, float]
            Longitude, latitude, and altitude in degrees and meters.

        """
        stations_df = self.stations()
        station_id = station_id.upper()
        if station_id not in stations_df["id"].values:
            closest_names = difflib.get_close_matches(
                station_id, stations_df["id"], n=5
            )
            msg = f"No station named {station_id} found. Closest: {closest_names}"
            raise ValueError(msg)
        row = stations_df[stations_df["id"] == station_id].iloc[0]
        return row["lon"], row["lat"], row["alt"]

    def read_station_llas(
        self,
        to_geodataframe: bool = True,
        filename=None,  # noqa: ARG002
    ) -> pd.DataFrame | gpd.GeoDataFrame:
        """Read station location information.

        .. deprecated::
            Use `stations()` instead.
        """
        warnings.warn(
            "read_station_llas is deprecated. Use stations() instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        result = self.stations()
        if not to_geodataframe:
            # Convert to regular DataFrame, drop geometry
            return pd.DataFrame(result.drop(columns="geometry"))
        return result

    def station_lonlat(self, station_id: str) -> tuple[float, float]:
        """Get longitude and latitude for a station.

        .. deprecated::
            Use `coordinates(station_id)[:2]` instead.
        """
        warnings.warn(
            "station_lonlat is deprecated. Use coordinates(station_id)[:2] instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        lon, lat, _ = self.coordinates(station_id)
        return lon, lat

__init__(cache_dir=None)

Initialize the GPS data source.

Parameters:

Name	Type	Description	Default
cache_dir	PathOrStr	Base directory to store cached data. Default is None, which uses utils.get_cache_dir(). Subclasses create directories under this base directory.	None

Source code in src/geepers/gps_sources/base.py

def __init__(self, cache_dir: PathOrStr | None = None):
    """Initialize the GPS data source.

    Parameters
    ----------
    cache_dir : PathOrStr, optional
        Base directory to store cached data.
        Default is None, which uses `utils.get_cache_dir()`.
        Subclasses create directories under this base directory.

    """
    if cache_dir is None:
        self._base_cache_dir = utils.get_cache_dir()
    else:
        self._base_cache_dir = Path(cache_dir)
    self._subdir = self.__class__.__name__.lower().replace("source", "")

coordinates(station_id)

Get coordinates for a single station.

Parameters:

Name	Type	Description	Default
station_id	str	The station identifier.	required

Returns:

Type	Description
tuple[float, float, float]	Longitude, latitude, and altitude in degrees and meters.

Source code in src/geepers/gps_sources/base.py

def coordinates(self, station_id: str) -> tuple[float, float, float]:
    """Get coordinates for a single station.

    Parameters
    ----------
    station_id : str
        The station identifier.

    Returns
    -------
    tuple[float, float, float]
        Longitude, latitude, and altitude in degrees and meters.

    """
    stations_df = self.stations()
    station_id = station_id.upper()
    if station_id not in stations_df["id"].values:
        closest_names = difflib.get_close_matches(
            station_id, stations_df["id"], n=5
        )
        msg = f"No station named {station_id} found. Closest: {closest_names}"
        raise ValueError(msg)
    row = stations_df[stations_df["id"] == station_id].iloc[0]
    return row["lon"], row["lat"], row["alt"]

read_station_llas(to_geodataframe=True, filename=None)

Read station location information.

.. deprecated:: Use stations() instead.

Source code in src/geepers/gps_sources/base.py

def read_station_llas(
    self,
    to_geodataframe: bool = True,
    filename=None,  # noqa: ARG002
) -> pd.DataFrame | gpd.GeoDataFrame:
    """Read station location information.

    .. deprecated::
        Use `stations()` instead.
    """
    warnings.warn(
        "read_station_llas is deprecated. Use stations() instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    result = self.stations()
    if not to_geodataframe:
        # Convert to regular DataFrame, drop geometry
        return pd.DataFrame(result.drop(columns="geometry"))
    return result

station_lonlat(station_id)

Get longitude and latitude for a station.

.. deprecated:: Use coordinates(station_id)[:2] instead.

Source code in src/geepers/gps_sources/base.py

def station_lonlat(self, station_id: str) -> tuple[float, float]:
    """Get longitude and latitude for a station.

    .. deprecated::
        Use `coordinates(station_id)[:2]` instead.
    """
    warnings.warn(
        "station_lonlat is deprecated. Use coordinates(station_id)[:2] instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    lon, lat, _ = self.coordinates(station_id)
    return lon, lat

stations(bbox=None, mask=None)

Get GPS stations, optionally filtered by spatial bounds.

Parameters:

Name	Type	Description	Default
bbox	tuple[float, float, float, float]	Bounding box as (west, south, east, north) in degrees.	None
mask	GeoSeries	Spatial mask to filter stations.	None

Returns:

Type	Description
GeoDataFrame	GeoDataFrame with station metadata including lon, lat, alt columns.

Source code in src/geepers/gps_sources/base.py

def stations(
    self,
    bbox: tuple[float, float, float, float] | None = None,
    mask: gpd.GeoSeries | None = None,
) -> gpd.GeoDataFrame:
    """Get GPS stations, optionally filtered by spatial bounds.

    Parameters
    ----------
    bbox : tuple[float, float, float, float], optional
        Bounding box as (west, south, east, north) in degrees.
    mask : gpd.GeoSeries, optional
        Spatial mask to filter stations.

    Returns
    -------
    gpd.GeoDataFrame
        GeoDataFrame with station metadata including lon, lat, alt columns.

    """
    # Read data from source
    gdf = self._read_station_data()

    # Apply spatial filters and validate
    gdf = self._apply_spatial_filters(gdf, bbox, mask)

    return gdf

timeseries(station_id, /, frame='ENU', start_date=None, end_date=None, zero_by='mean', download_if_missing=True) abstractmethod

Load GPS station time series data.

Parameters:

Name	Type	Description	Default
station_id	str	The station identifier.	required
frame	('ENU', 'XYZ')	Coordinate frame for the data. Default is "ENU".	"ENU"
start_date	str	Start date for data filtering (ISO format).	None
end_date	str	End date for data filtering (ISO format).	None
zero_by	Literal['mean', 'start']	How to zero the data. Either "mean" or "start".	'mean'
download_if_missing	bool	Whether to download data if not found locally.	True

Returns:

Type	Description
DataFrame	DataFrame validated against StationObservationSchema.

Source code in src/geepers/gps_sources/base.py

@abstractmethod
def timeseries(
    self,
    station_id: str,
    /,
    frame: Literal["ENU", "XYZ"] = "ENU",
    start_date: str | None = None,
    end_date: str | None = None,
    zero_by: Literal["mean", "start", "none"] = "mean",
    download_if_missing: bool = True,
) -> pd.DataFrame:
    """Load GPS station time series data.

    Parameters
    ----------
    station_id : str
        The station identifier.
    frame : {"ENU", "XYZ"}, optional
        Coordinate frame for the data. Default is "ENU".
    start_date : str, optional
        Start date for data filtering (ISO format).
    end_date : str, optional
        End date for data filtering (ISO format).
    zero_by : Literal["mean", "start"], optional
        How to zero the data. Either "mean" or "start".
    download_if_missing : bool, optional
        Whether to download data if not found locally.

    Returns
    -------
    pd.DataFrame
        DataFrame validated against StationObservationSchema.

    """

SideshowSource

Bases: BaseGpsSource

JPL Sideshow GPS data source.

Source code in src/geepers/gps_sources/sideshow.py

class SideshowSource(BaseGpsSource):
    """JPL Sideshow GPS data source."""

    _names: Final[list[str]] = [
        "date",
        "east",
        "north",
        "up",
        "sigma_east",
        "sigma_north",
        "sigma_up",
        "corr_en",
        "corr_eu",
        "corr_nu",
    ]

    def timeseries(
        self,
        station_id: str,
        /,
        frame: Literal["ENU", "XYZ"] = "ENU",
        start_date: str | None = None,
        end_date: str | None = None,
        zero_by: Literal["mean", "start", "none"] = "mean",
        download_if_missing: bool = True,  # noqa: ARG002
    ) -> pd.DataFrame:
        """Load GPS station time series data.

        Parameters
        ----------
        station_id : str
            The station identifier.
        frame : {"ENU", "XYZ"}, optional
            Coordinate frame for the data. Default is "ENU".
        start_date : str, optional
            Start date for data filtering (ISO format).
        end_date : str, optional
            End date for data filtering (ISO format).
        zero_by : str, optional
            How to zero the data. Either "mean" or "start".
        download_if_missing : bool, optional
            Whether to download data if not found locally.

        Returns
        -------
        pd.DataFrame
            DataFrame with ENU time series data validated against schema.

        """
        if frame == "XYZ":
            msg = "XYZ frame not supported for Sideshow data"
            raise ValueError(msg)

        # Copy so we never mutate the lru_cached DataFrame
        df = self._read_series(station_id).copy()
        # Replace decimal year with datetime
        df["date"] = decimal_years_to_datetimes(df["decimal_year"])
        df = df.drop(columns=["decimal_year"])
        # Move date to first column:
        df = df[["date", *df.columns[:-1].to_list()]]
        df = self._filter_by_date(df, start_date, end_date)
        df = self._zero_data(df, zero_by, columns=["east", "north", "up"])
        return StationObservationSchema.validate(df, lazy=True)

    @staticmethod
    @lru_cache(maxsize=128)
    def _read_series(station_id: str) -> pd.DataFrame:
        _raw_names = ["decimal_year"] + SideshowSource._names[1:]
        # https://sideshow.jpl.nasa.gov/post/tables/GNSS_Time_Series.pdf
        # Time Series and Residual Format
        # Column 1: Decimal_YR
        # Columns 2-4: East(m) North(m) Vert(m)
        # Columns 5-7: E_sig(m) N_sig(m) V_sig(m)
        # Columns 8-10: E_N_cor E_V_cor N_V_cor
        # Column 11: Time in Seconds past J2000
        # Columns 12-17: Time in YEAR MM DD HR MN SS
        series_url = GPS_BASE_URL.format(station=station_id)
        return pd.read_csv(
            series_url,
            sep=r"\s+",
            engine="c",
            header=None,
            names=_raw_names,
            usecols=list(range(len(_raw_names))),
        )

    @staticmethod
    @lru_cache(maxsize=1)
    def _fetch_station_data() -> gpd.GeoDataFrame:
        """Download and cache the JPL Sideshow site list."""
        import warnings

        with warnings.catch_warnings(category=UserWarning, action="ignore"):
            return np.loadtxt(SITE_LIST_URL, comments="<", skiprows=9, dtype=str)

    def _read_station_data(self) -> gpd.GeoDataFrame:
        lines = self._fetch_station_data()
        stations = {
            "id": lines[::2, 0],
            "lat": lines[::2, 2].astype(float),
            "lon": lines[::2, 3].astype(float),
            "alt": lines[::2, 4].astype(float) / 1000,  # JPL height is in millimeters
            # next three columns are sigmas for the site position
        }
        df = pd.DataFrame(stations)
        df.loc[:, "lon"] = df.lon - (np.round(df.lon / 360) * 360)

        return gpd.GeoDataFrame(
            df, geometry=gpd.points_from_xy(df.lon, df.lat), crs="EPSG:4326"
        )

timeseries(station_id, /, frame='ENU', start_date=None, end_date=None, zero_by='mean', download_if_missing=True)

Load GPS station time series data.

Parameters:

Name	Type	Description	Default
station_id	str	The station identifier.	required
frame	('ENU', 'XYZ')	Coordinate frame for the data. Default is "ENU".	"ENU"
start_date	str	Start date for data filtering (ISO format).	None
end_date	str	End date for data filtering (ISO format).	None
zero_by	str	How to zero the data. Either "mean" or "start".	'mean'
download_if_missing	bool	Whether to download data if not found locally.	True

Returns:

Type	Description
DataFrame	DataFrame with ENU time series data validated against schema.

Source code in src/geepers/gps_sources/sideshow.py

def timeseries(
    self,
    station_id: str,
    /,
    frame: Literal["ENU", "XYZ"] = "ENU",
    start_date: str | None = None,
    end_date: str | None = None,
    zero_by: Literal["mean", "start", "none"] = "mean",
    download_if_missing: bool = True,  # noqa: ARG002
) -> pd.DataFrame:
    """Load GPS station time series data.

    Parameters
    ----------
    station_id : str
        The station identifier.
    frame : {"ENU", "XYZ"}, optional
        Coordinate frame for the data. Default is "ENU".
    start_date : str, optional
        Start date for data filtering (ISO format).
    end_date : str, optional
        End date for data filtering (ISO format).
    zero_by : str, optional
        How to zero the data. Either "mean" or "start".
    download_if_missing : bool, optional
        Whether to download data if not found locally.

    Returns
    -------
    pd.DataFrame
        DataFrame with ENU time series data validated against schema.

    """
    if frame == "XYZ":
        msg = "XYZ frame not supported for Sideshow data"
        raise ValueError(msg)

    # Copy so we never mutate the lru_cached DataFrame
    df = self._read_series(station_id).copy()
    # Replace decimal year with datetime
    df["date"] = decimal_years_to_datetimes(df["decimal_year"])
    df = df.drop(columns=["decimal_year"])
    # Move date to first column:
    df = df[["date", *df.columns[:-1].to_list()]]
    df = self._filter_by_date(df, start_date, end_date)
    df = self._zero_data(df, zero_by, columns=["east", "north", "up"])
    return StationObservationSchema.validate(df, lazy=True)

UnrGridSource

Bases: BaseGpsSource

UNR Grid GPS data source for gridded time series data.

Source code in src/geepers/gps_sources/unr_grid.py

class UnrGridSource(BaseGpsSource):
    """UNR Grid GPS data source for gridded time series data."""

    def __init__(
        self,
        version: Literal["0.1", "0.3"] = DEFAULT_VERSION,
        gridded_type: Literal["constant", "variable"] = "variable",
        cache_dir: str | Path | None = None,
    ):
        """Initialize UNR Grid GPS data source.

        Parameters
        ----------
        version : Literal["0.1", "0.3"], optional
            Version of the UNR grid data to use. Default is "0.3".
        gridded_type : Literal["constant", "variable"], optional
            Whether to use the time-constant or time-variable gridded
            products. Only available starting with version "0.3";
            version "0.1" only has time-variable data. Default is "variable".
        cache_dir : str | Path, optional
            Directory to cache downloaded data files.

        """
        # Initialize BaseGpsSource
        super().__init__(cache_dir=cache_dir)

        # Store UNR version
        self.version = version
        self.gridded_type = gridded_type

    def timeseries(
        self,
        station_id: str,
        /,
        frame: Literal["ENU", "XYZ"] = "ENU",
        start_date: str | None = None,
        end_date: str | None = None,
        zero_by: Literal["mean", "start", "none"] = "mean",
        download_if_missing: bool = True,
        *,
        plate: Literal["NA", "PA", "IGS14", "IGS20"] = "IGS14",
    ) -> pd.DataFrame:
        """Load grid point time series data.

        Parameters
        ----------
        station_id : str
            The grid point identifier (6-digit string).
        frame : {"ENU", "XYZ"}, optional
            Coordinate frame for the data. Default is "ENU".
        start_date : str, optional
            Start date for data filtering (ISO format). Currently not implemented.
        end_date : str, optional
            End date for data filtering (ISO format). Currently not implemented.
        zero_by : Literal["mean", "start"], optional
            How to zero the data. Either "mean" or "start".
        download_if_missing : bool, optional
            Whether to download data if not found locally. Currently not implemented.
        plate : Literal["NA", "PA", "IGS14", "IGS20"], optional
            Plate for the data. Default is "IGS14".

        Returns
        -------
        pd.DataFrame
            DataFrame with ENU time series data validated against schema.

        Raises
        ------
        ValueError
            If XYZ frame is requested (not supported for grid data).

        """
        if frame == "XYZ":
            msg = "XYZ frame not supported for grid data"
            raise ValueError(msg)

        # TODO: how to handle fetching/saving, vs using pandas to read...
        if download_if_missing:
            local_file = self._download_file(
                station_id,
                plate=plate,
                version=self.version,
                gridded_type=self.gridded_type,
            )
            df = self.parse_data_file(local_file)
        else:
            filename = FILENAME_TEMPLATE.format(plate=plate, grid_id=int(station_id))
            gridded_dir = GRIDDED_TYPE_DIRS[
                self.gridded_type if self.version != "0.1" else "variable"
            ]
            uri = GRID_DATA_BASE_URL.format(
                version=self.version, gridded_dir=gridded_dir, filename=filename
            )
            df = self.parse_data_file(uri)

        df = self._filter_by_date(df, start_date, end_date)
        df = self._zero_data(df, zero_by, columns=["east", "north", "up"])
        return StationObservationSchema.validate(df, lazy=True)

    def _read_station_data(self) -> gpd.GeoDataFrame:
        """Read raw grid point data from the source.

        Returns
        -------
        gpd.GeoDataFrame
            GeoDataFrame with grid point metadata including lon, lat, alt columns.

        """
        df = self._read_grid_file(version=self.version)

        # Rename columns to match expected format
        df_out = df.reset_index()
        df_out = df_out.rename(
            columns={"grid_point": "id", "latitude": "lat", "longitude": "lon"}
        )
        df_out["alt"] = 0.0  # Grid points don't have altitude info

        # Convert to GeoDataFrame
        gdf = gpd.GeoDataFrame(
            df_out,
            geometry=gpd.points_from_xy(df_out.lon, df_out.lat),
            crs="EPSG:4326",
        )

        return gdf

    def stations(
        self,
        bbox: tuple[float, float, float, float] | None = None,
        mask: gpd.GeoSeries | None = None,
    ) -> gpd.GeoDataFrame:
        """Get grid points, optionally filtered by spatial bounds.

        Parameters
        ----------
        bbox : tuple[float, float, float, float], optional
            Bounding box as (west, south, east, north) in degrees.
        mask : gpd.GeoSeries, optional
            Spatial mask to filter grid points.

        Returns
        -------
        gpd.GeoDataFrame
            GeoDataFrame with grid point metadata including lon, lat, alt columns.

        """
        # Get data using base class method
        gdf = super().stations(bbox, mask)

        # Apply grid-specific schema validation
        GridCellSchema.validate(gdf, lazy=True)

        return gdf

    def _download_file(
        self,
        grid_id: str,
        plate: Literal["NA", "PA", "IGS14", "IGS20"] = "IGS14",
        output_dir: Path | None = None,
        session: requests.Session | None = None,
        version: Literal["0.1", "0.3"] = DEFAULT_VERSION,
        gridded_type: Literal["constant", "variable"] = "variable",
    ) -> Path:
        """Download ont .tenv8 data file.

        Parameters
        ----------
        grid_id: str
            Grid point ID to download.
        plate : Literal["NA", "PA", "IGS14", "IGS20], optional
            Plate for the data. Default is "IGS14".
        output_dir : Path | None, optional
            Directory to store downloaded data files.
            If None, the cache directory is used.
        session : requests.Session
            A shared requests.Session object.
            Can be used for retrying.
        version : Literal["0.1", "0.3"], optional
            Version of the UNR grid data to download.
        gridded_type : Literal["constant", "variable"], optional
            Whether to download the time-constant or time-variable
            gridded product. Version "0.1" only has time-variable data,
            so this is ignored when version is "0.1".

        Returns
        -------
        Path
            Paths to downloaded data files.

        """
        if output_dir is None:
            output_dir = self._cache_dir
        output_dir.mkdir(parents=True, exist_ok=True)

        if plate == "IGS14" and version == "0.3":
            # Warning: IGS14 plate is not available in UNR version 0.3.
            # Using IGS20 instead.
            plate = "IGS20"

        gridded_dir = GRIDDED_TYPE_DIRS[
            gridded_type if version != "0.1" else "variable"
        ]
        # Accept both int and zero-padded string ids ("000123" or 123)
        filename = FILENAME_TEMPLATE.format(plate=plate, grid_id=int(grid_id))
        url = GRID_DATA_BASE_URL.format(
            version=version, gridded_dir=gridded_dir, filename=filename
        )
        dest = output_dir / url.rsplit("/", 1)[-1]
        if not dest.exists():
            if session is None:
                resp = requests.get(url, timeout=REQUEST_TIMEOUT)
            else:
                resp = session.get(url, timeout=REQUEST_TIMEOUT)
            resp.raise_for_status()
            with dest.open("wb") as f:
                f.write(resp.content)
        return dest

    def download_data_files(
        self,
        grid_id_list: list[str] | None = None,
        plate: Literal["NA", "PA", "IGS14", "IGS20"] = "IGS14",
        max_workers: int = 8,
        output_dir: Path | None = None,
        version: Literal["0.1", "0.3"] = DEFAULT_VERSION,
        gridded_type: Literal["constant", "variable"] = "variable",
    ) -> list[Path]:
        """Download .tenv8 data files in parallel, showing progress.

        Parameters
        ----------
        grid_id_list : list[str], optional
            Specific grid point IDs to download.
            If None, all grid points are downloaded.
        plate : Literal["NA", "PA", "IGS14", "IGS20"], optional
            Plate for the data. Default is "IGS14".
        max_workers : int, optional
            Number of threads to use for downloading in parallel.
        output_dir : Path | None, optional
            Directory to store downloaded data files.
            If None, the cache directory is used.
        version : Literal["0.1", "0.3"], optional
            Version of the UNR grid data to download. Default is "0.3".
        gridded_type : Literal["constant", "variable"], optional
            Whether to download the time-constant or time-variable
            gridded product. Ignored when version is "0.1", since that
            version only has time-variable data.

        Returns
        -------
        list[Path]
            List of paths to downloaded data files.

        """
        if output_dir is None:
            output_dir = self._cache_dir
        output_dir.mkdir(parents=True, exist_ok=True)
        if grid_id_list is None:
            grid_id_list = self.stations().index.tolist()

        s = requests.Session()
        retries = Retry(total=5, backoff_factor=1, status_forcelist=[502, 503, 504])
        s.mount("https://", HTTPAdapter(max_retries=retries))

        return thread_map(
            self._download_file,
            grid_id_list,
            plate=plate,
            output_dir=output_dir,
            max_workers=max_workers,
            session=s,
            desc="Downloading data files",
            version=version,
            gridded_type=gridded_type,
        )

    @staticmethod
    @lru_cache(maxsize=128)
    def _read_data_file(uri: str | Path) -> pd.DataFrame:
        df = pd.read_csv(
            uri,
            sep=r"\s+",
            header=None,
            names=[
                "decimal_year",
                "east",
                "north",
                "up",
                "sigma_east",
                "sigma_north",
                "sigma_up",
                "rapid_flag",
            ],
        )
        return df

    def parse_data_file(self, uri: str | Path) -> pd.DataFrame:
        """Parse a .tenv8 time-series data file into a DataFrame.

        Parameters
        ----------
        uri : str | Path
            Path or URL to the .tenv8 file.

        Returns
        -------
        pd.DataFrame
            DataFrame with columns validated against GPSUncertaintySchema.

        """
        # Copy so we never mutate the lru_cached DataFrame
        df = self._read_data_file(uri).copy()

        # Convert decimal year to datetime (vectorized)
        df["date"] = decimal_years_to_datetimes(df["decimal_year"])

        # Add placeholder correlation values (not in .tenv8 format)
        df["corr_en"] = 0.0
        df["corr_eu"] = 0.0
        df["corr_nu"] = 0.0

        # Select relevant columns for validation
        df_out = df[
            [
                "date",
                "east",
                "north",
                "up",
                "sigma_east",
                "sigma_north",
                "sigma_up",
                "corr_en",
                "corr_eu",
                "corr_nu",
            ]
        ]
        # UNR Grid is in millimeters instead of meters (values and sigmas):
        sigma_cols = ["sigma_east", "sigma_north", "sigma_up"]
        df_out.loc[:, ["east", "north", "up", *sigma_cols]] /= 1000

        # Time-constant gridded products can report exactly-zero sigmas,
        # which the schema rejects; clamp to its minimum
        df_out.loc[:, sigma_cols] = df_out[sigma_cols].clip(lower=EPS)

        return StationObservationSchema.validate(df_out, lazy=True)

    @staticmethod
    @cache
    def _read_grid_file(version: str = DEFAULT_VERSION) -> pd.DataFrame:
        """Download and cache the UNR grid latitude/longitude lookup table."""
        if version not in VALID_VERSIONS:
            msg = (
                f"Unsupported version '{version}'. "
                f"Valid options are: {', '.join(VALID_VERSIONS)}."
            )
            raise ValueError(msg)

        url = LOOKUP_FILE_URL.format(version=version)
        df = pd.read_csv(
            url,
            sep=r"\s+",
            names=["grid_point", "longitude", "latitude"],
        )
        if version == "0.3":
            # Version 0.3 maps longitudes from 0 to 360
            # convert to -180 to 180, to be consistent with version 0.1
            df["longitude"] = ((df["longitude"] + 180) % 360) - 180
        return df.set_index("grid_point")

__init__(version=DEFAULT_VERSION, gridded_type='variable', cache_dir=None)

Initialize UNR Grid GPS data source.

Parameters:

Name	Type	Description	Default
version	Literal['0.1', '0.3']	Version of the UNR grid data to use. Default is "0.3".	DEFAULT_VERSION
gridded_type	Literal['constant', 'variable']	Whether to use the time-constant or time-variable gridded products. Only available starting with version "0.3"; version "0.1" only has time-variable data. Default is "variable".	'variable'
cache_dir	str | Path	Directory to cache downloaded data files.	None

Source code in src/geepers/gps_sources/unr_grid.py

def __init__(
    self,
    version: Literal["0.1", "0.3"] = DEFAULT_VERSION,
    gridded_type: Literal["constant", "variable"] = "variable",
    cache_dir: str | Path | None = None,
):
    """Initialize UNR Grid GPS data source.

    Parameters
    ----------
    version : Literal["0.1", "0.3"], optional
        Version of the UNR grid data to use. Default is "0.3".
    gridded_type : Literal["constant", "variable"], optional
        Whether to use the time-constant or time-variable gridded
        products. Only available starting with version "0.3";
        version "0.1" only has time-variable data. Default is "variable".
    cache_dir : str | Path, optional
        Directory to cache downloaded data files.

    """
    # Initialize BaseGpsSource
    super().__init__(cache_dir=cache_dir)

    # Store UNR version
    self.version = version
    self.gridded_type = gridded_type

download_data_files(grid_id_list=None, plate='IGS14', max_workers=8, output_dir=None, version=DEFAULT_VERSION, gridded_type='variable')

Download .tenv8 data files in parallel, showing progress.

Parameters:

Name	Type	Description	Default
grid_id_list	list[str]	Specific grid point IDs to download. If None, all grid points are downloaded.	None
plate	Literal['NA', 'PA', 'IGS14', 'IGS20']	Plate for the data. Default is "IGS14".	'IGS14'
max_workers	int	Number of threads to use for downloading in parallel.	8
output_dir	Path | None	Directory to store downloaded data files. If None, the cache directory is used.	None
version	Literal['0.1', '0.3']	Version of the UNR grid data to download. Default is "0.3".	DEFAULT_VERSION
gridded_type	Literal['constant', 'variable']	Whether to download the time-constant or time-variable gridded product. Ignored when version is "0.1", since that version only has time-variable data.	'variable'

Returns:

Type	Description
list[Path]	List of paths to downloaded data files.

Source code in src/geepers/gps_sources/unr_grid.py

def download_data_files(
    self,
    grid_id_list: list[str] | None = None,
    plate: Literal["NA", "PA", "IGS14", "IGS20"] = "IGS14",
    max_workers: int = 8,
    output_dir: Path | None = None,
    version: Literal["0.1", "0.3"] = DEFAULT_VERSION,
    gridded_type: Literal["constant", "variable"] = "variable",
) -> list[Path]:
    """Download .tenv8 data files in parallel, showing progress.

    Parameters
    ----------
    grid_id_list : list[str], optional
        Specific grid point IDs to download.
        If None, all grid points are downloaded.
    plate : Literal["NA", "PA", "IGS14", "IGS20"], optional
        Plate for the data. Default is "IGS14".
    max_workers : int, optional
        Number of threads to use for downloading in parallel.
    output_dir : Path | None, optional
        Directory to store downloaded data files.
        If None, the cache directory is used.
    version : Literal["0.1", "0.3"], optional
        Version of the UNR grid data to download. Default is "0.3".
    gridded_type : Literal["constant", "variable"], optional
        Whether to download the time-constant or time-variable
        gridded product. Ignored when version is "0.1", since that
        version only has time-variable data.

    Returns
    -------
    list[Path]
        List of paths to downloaded data files.

    """
    if output_dir is None:
        output_dir = self._cache_dir
    output_dir.mkdir(parents=True, exist_ok=True)
    if grid_id_list is None:
        grid_id_list = self.stations().index.tolist()

    s = requests.Session()
    retries = Retry(total=5, backoff_factor=1, status_forcelist=[502, 503, 504])
    s.mount("https://", HTTPAdapter(max_retries=retries))

    return thread_map(
        self._download_file,
        grid_id_list,
        plate=plate,
        output_dir=output_dir,
        max_workers=max_workers,
        session=s,
        desc="Downloading data files",
        version=version,
        gridded_type=gridded_type,
    )

parse_data_file(uri)

Parse a .tenv8 time-series data file into a DataFrame.

Parameters:

Name	Type	Description	Default
uri	str | Path	Path or URL to the .tenv8 file.	required

Returns:

Type	Description
DataFrame	DataFrame with columns validated against GPSUncertaintySchema.

Source code in src/geepers/gps_sources/unr_grid.py

def parse_data_file(self, uri: str | Path) -> pd.DataFrame:
    """Parse a .tenv8 time-series data file into a DataFrame.

    Parameters
    ----------
    uri : str | Path
        Path or URL to the .tenv8 file.

    Returns
    -------
    pd.DataFrame
        DataFrame with columns validated against GPSUncertaintySchema.

    """
    # Copy so we never mutate the lru_cached DataFrame
    df = self._read_data_file(uri).copy()

    # Convert decimal year to datetime (vectorized)
    df["date"] = decimal_years_to_datetimes(df["decimal_year"])

    # Add placeholder correlation values (not in .tenv8 format)
    df["corr_en"] = 0.0
    df["corr_eu"] = 0.0
    df["corr_nu"] = 0.0

    # Select relevant columns for validation
    df_out = df[
        [
            "date",
            "east",
            "north",
            "up",
            "sigma_east",
            "sigma_north",
            "sigma_up",
            "corr_en",
            "corr_eu",
            "corr_nu",
        ]
    ]
    # UNR Grid is in millimeters instead of meters (values and sigmas):
    sigma_cols = ["sigma_east", "sigma_north", "sigma_up"]
    df_out.loc[:, ["east", "north", "up", *sigma_cols]] /= 1000

    # Time-constant gridded products can report exactly-zero sigmas,
    # which the schema rejects; clamp to its minimum
    df_out.loc[:, sigma_cols] = df_out[sigma_cols].clip(lower=EPS)

    return StationObservationSchema.validate(df_out, lazy=True)

stations(bbox=None, mask=None)

Get grid points, optionally filtered by spatial bounds.

Parameters:

Name	Type	Description	Default
bbox	tuple[float, float, float, float]	Bounding box as (west, south, east, north) in degrees.	None
mask	GeoSeries	Spatial mask to filter grid points.	None

Returns:

Type	Description
GeoDataFrame	GeoDataFrame with grid point metadata including lon, lat, alt columns.

Source code in src/geepers/gps_sources/unr_grid.py

def stations(
    self,
    bbox: tuple[float, float, float, float] | None = None,
    mask: gpd.GeoSeries | None = None,
) -> gpd.GeoDataFrame:
    """Get grid points, optionally filtered by spatial bounds.

    Parameters
    ----------
    bbox : tuple[float, float, float, float], optional
        Bounding box as (west, south, east, north) in degrees.
    mask : gpd.GeoSeries, optional
        Spatial mask to filter grid points.

    Returns
    -------
    gpd.GeoDataFrame
        GeoDataFrame with grid point metadata including lon, lat, alt columns.

    """
    # Get data using base class method
    gdf = super().stations(bbox, mask)

    # Apply grid-specific schema validation
    GridCellSchema.validate(gdf, lazy=True)

    return gdf

timeseries(station_id, /, frame='ENU', start_date=None, end_date=None, zero_by='mean', download_if_missing=True, *, plate='IGS14')

Load grid point time series data.

Parameters:

Name	Type	Description	Default
station_id	str	The grid point identifier (6-digit string).	required
frame	('ENU', 'XYZ')	Coordinate frame for the data. Default is "ENU".	"ENU"
start_date	str	Start date for data filtering (ISO format). Currently not implemented.	None
end_date	str	End date for data filtering (ISO format). Currently not implemented.	None
zero_by	Literal['mean', 'start']	How to zero the data. Either "mean" or "start".	'mean'
download_if_missing	bool	Whether to download data if not found locally. Currently not implemented.	True
plate	Literal['NA', 'PA', 'IGS14', 'IGS20']	Plate for the data. Default is "IGS14".	'IGS14'

Returns:

Type	Description
DataFrame	DataFrame with ENU time series data validated against schema.

Raises:

Type	Description
ValueError	If XYZ frame is requested (not supported for grid data).

Source code in src/geepers/gps_sources/unr_grid.py

def timeseries(
    self,
    station_id: str,
    /,
    frame: Literal["ENU", "XYZ"] = "ENU",
    start_date: str | None = None,
    end_date: str | None = None,
    zero_by: Literal["mean", "start", "none"] = "mean",
    download_if_missing: bool = True,
    *,
    plate: Literal["NA", "PA", "IGS14", "IGS20"] = "IGS14",
) -> pd.DataFrame:
    """Load grid point time series data.

    Parameters
    ----------
    station_id : str
        The grid point identifier (6-digit string).
    frame : {"ENU", "XYZ"}, optional
        Coordinate frame for the data. Default is "ENU".
    start_date : str, optional
        Start date for data filtering (ISO format). Currently not implemented.
    end_date : str, optional
        End date for data filtering (ISO format). Currently not implemented.
    zero_by : Literal["mean", "start"], optional
        How to zero the data. Either "mean" or "start".
    download_if_missing : bool, optional
        Whether to download data if not found locally. Currently not implemented.
    plate : Literal["NA", "PA", "IGS14", "IGS20"], optional
        Plate for the data. Default is "IGS14".

    Returns
    -------
    pd.DataFrame
        DataFrame with ENU time series data validated against schema.

    Raises
    ------
    ValueError
        If XYZ frame is requested (not supported for grid data).

    """
    if frame == "XYZ":
        msg = "XYZ frame not supported for grid data"
        raise ValueError(msg)

    # TODO: how to handle fetching/saving, vs using pandas to read...
    if download_if_missing:
        local_file = self._download_file(
            station_id,
            plate=plate,
            version=self.version,
            gridded_type=self.gridded_type,
        )
        df = self.parse_data_file(local_file)
    else:
        filename = FILENAME_TEMPLATE.format(plate=plate, grid_id=int(station_id))
        gridded_dir = GRIDDED_TYPE_DIRS[
            self.gridded_type if self.version != "0.1" else "variable"
        ]
        uri = GRID_DATA_BASE_URL.format(
            version=self.version, gridded_dir=gridded_dir, filename=filename
        )
        df = self.parse_data_file(uri)

    df = self._filter_by_date(df, start_date, end_date)
    df = self._zero_data(df, zero_by, columns=["east", "north", "up"])
    return StationObservationSchema.validate(df, lazy=True)

UnrSource

Bases: BaseGpsSource

UNR GPS data source.

Source code in src/geepers/gps_sources/unr.py

class UnrSource(BaseGpsSource):
    """UNR GPS data source."""

    def timeseries(
        self,
        station_id: str,
        /,
        frame: Literal["ENU", "XYZ"] = "ENU",
        start_date: str | None = None,
        end_date: str | None = None,
        zero_by: Literal["mean", "start", "none"] = "mean",
        download_if_missing: bool = True,
        plate_fixed: bool = False,
        plate_name: str | None = None,
    ) -> pd.DataFrame:
        """Load GPS station time series data.

        Parameters
        ----------
        station_id : str
            The station identifier.
        frame : {"ENU", "XYZ"}, optional
            Coordinate frame for the data. Default is "ENU".
        start_date : str, optional
            Start date for data filtering (ISO format).
        end_date : str, optional
            End date for data filtering (ISO format).
        download_if_missing : bool, optional
            Whether to download data if not found locally.
        zero_by : Literal["mean", "start"], optional
            How to zero the data. Either "mean" or "start".
        plate_fixed : bool, optional
            Whether to use plate-fixed coordinates.
        plate_name : str, optional
            If `plate_fixed=True`, specify which plate to use.
            Some stations have multiple plates.
            If None, uses the first plate found.

        Returns
        -------
        pd.DataFrame
            DataFrame validated against StationObservationSchema.

        """
        if frame not in ["ENU", "XYZ"]:
            msg = f"Unsupported frame: {frame}. Use 'ENU' or 'XYZ'"
            raise ValueError(msg)

        station_id = validate_station_id(station_id.upper())

        plate = None
        if plate_fixed and frame == "ENU":
            plates = self._get_station_plates(station_id)
            if plate_name:
                if plate_name not in plates:
                    msg = (
                        f"Plate {plate_name} not found for {station_id}, which has"
                        f" plates {plates}"
                    )
                    raise ValueError(msg)
                plate = plate_name
            else:
                plate = plates[0]
            gps_data_file = self._cache_dir / plate / f"{station_id}.tenv3"
        else:
            if frame == "ENU":
                gps_data_file = self._cache_dir / f"{station_id}.tenv3"
            else:  # frame in ("XYZ")
                gps_data_file = self._cache_dir / f"{station_id}.txyz2"

        if not gps_data_file.exists():
            if download_if_missing:
                logger.info(f"Downloading {station_id} to {gps_data_file}")
                self.download_station_data(
                    station_id, frame=frame, plate_fixed=plate_fixed, plate=plate
                )
            else:
                msg = f"{gps_data_file} does not exist, download_if_missing = False"
                raise ValueError(msg)

        df = pd.read_csv(gps_data_file, sep=r"\s+", engine="c")
        df = self._clean_gps_df(
            df, start_date, end_date, coords="enu" if frame == "ENU" else "xyz"
        )

        if frame == "ENU" and zero_by:
            df = self._zero_data(df, zero_by, columns=["east", "north", "up"])

        if frame == "ENU":
            StationObservationSchema.validate(df, lazy=True)

        return df

    def _read_station_data(self) -> gpd.GeoDataFrame:
        """Read raw station data from the source.

        Returns
        -------
        gpd.GeoDataFrame
            GeoDataFrame with station metadata including lon, lat, alt columns.

        """
        today = datetime.date.today().strftime("%Y%m%d")
        lla_path = self._cache_dir / STATION_LLH_FILENAME.format(today=today)

        try:
            df = pd.read_csv(lla_path, sep=r"\s+", engine="c", header=None)
        except FileNotFoundError:
            logger.info(f"Downloading from {STATION_LLH_URL} to {lla_path}")
            self._download_station_locations(lla_path, STATION_LLH_URL)
            df = pd.read_csv(lla_path, sep=r"\s+", engine="c", header=None)

        processed_stations = self.get_global_station_list()
        df = df[df[0].isin(processed_stations)]
        df.columns = ["id", "lat", "lon", "alt"]
        df.loc[:, "lon"] = df.lon - (np.round(df.lon / 360) * 360)

        return gpd.GeoDataFrame(
            df, geometry=gpd.points_from_xy(df.lon, df.lat), crs="EPSG:4326"
        )

    def download_station_data(
        self,
        station_id: str,
        frame: Literal["ENU", "XYZ"] = "ENU",
        reference: Literal["IGS14", "IGS20"] = "IGS20",
        plate_fixed: bool = False,
        plate: str | None = None,
    ) -> None:
        """Download GPS station data from the Nevada Geodetic Laboratory.

        Parameters
        ----------
        station_id : str
            The station identifier.
        frame : {"ENU", "XYZ"}, optional
            The coordinate system of the data to download. Default is "ENU".
        reference : {"IGS14", "IGS20"}
            Geodetic reference of processed data.
        plate_fixed : bool, optional
            Whether to download plate-fixed data. Only applicable for "ENU" frame.
        plate : str, optional
            If using plate_fixed, specify which plate to use (in the case of a station
            on multiple plates).
            If None, uses the first plate from the UNR results.

        """
        station_id = validate_station_id(station_id.upper())

        if frame == "ENU":
            if plate_fixed:
                if plate is None:
                    plate = self._get_station_plates(station_id)[0]
                url = f"https://geodesy.unr.edu/gps_timeseries/tenv3/plates/{plate}/{station_id}.{plate}.tenv3"
                filename = self._cache_dir / plate / f"{station_id}.tenv3"
            else:
                url = GPS_BASE_URL.format(station=station_id, reference=reference)
                # Hack to get around bad url structure
                url = url.replace("gps_timeseries/IGS14", "gps_timeseries")
                filename = self._cache_dir / f"{station_id}.tenv3"
        elif frame == "XYZ":
            url = f"https://geodesy.unr.edu/gps_timeseries/txyz/{reference}/{station_id}.txyz2"
            filename = self._cache_dir / f"{station_id}.txyz2"
        else:
            msg = "frame must be 'ENU' or 'XYZ'"
            raise ValueError(msg)

        response = requests.get(url, timeout=REQUEST_TIMEOUT)
        response.raise_for_status()

        filename.parent.mkdir(parents=True, exist_ok=True)
        filename.write_text(response.text)
        logger.info(f"Saved {url} to {filename}")

    def _get_station_plates(self, station_id: str) -> list[str]:
        """Get the tectonic plate(s) for a given GPS station."""
        plates = self._read_station_plates_table().get(station_id)
        if plates is None:
            msg = f"Failed to find {station_id} in the UNR station plates table"
            raise ValueError(msg)
        return plates

    @staticmethod
    @cache
    def _read_station_plates_table() -> dict[str, list[str]]:
        """Download and parse the UNR station -> plates table (cached)."""
        # A text file that gives the plate associated with each station.
        # The directory also contains files for each frame ("plate_??.txt" where
        # ?? is the 2-character plate designation) that list the stations
        # associated with each plate.
        url = "https://geodesy.unr.edu/gps_timeseries/Plates/sta_frames.txt"
        response = requests.get(url, timeout=REQUEST_TIMEOUT)
        response.raise_for_status()
        table: dict[str, list[str]] = {}
        for line in response.text.splitlines():
            cur_id, *plates = line.split(" ")
            table[cur_id] = plates
        return table

    def _clean_gps_df(
        self,
        df: pd.DataFrame,
        start_date: str | None = None,
        end_date: str | None = None,
        coords: str = "enu",
    ) -> pd.DataFrame:
        """Clean and preprocess the GPS DataFrame."""
        df["date"] = pd.to_datetime(df["YYMMMDD"], format="%y%b%d")

        df = self._filter_by_date(df, start_date, end_date)

        if coords == "enu":
            df_integer = df[["_e0(m)", "____n0(m)", "u0(m)"]]
            df_out = df[
                [
                    "date",
                    "__east(m)",
                    "_north(m)",
                    "____up(m)",
                    "sig_e(m)",
                    "sig_n(m)",
                    "sig_u(m)",
                    "__corr_en",
                    "__corr_eu",
                    "__corr_nu",
                ]
            ]
            # Combine the integer e/n/u part with the fractional
            df_out.loc[:, ["__east(m)", "_north(m)", "____up(m)"]] += df_integer.values
        elif coords == "xyz":
            df_out = df[["date", "x", "y", "z"]]
        else:
            msg = "coords must be either 'enu' or 'xyz'"
            raise ValueError(msg)

        df_out = df_out.rename(columns=lambda s: s.replace("_", "").replace("(m)", ""))
        df_out = df_out.rename(
            columns={
                "sige": "sigma_east",
                "sign": "sigma_north",
                "sigu": "sigma_up",
                "corren": "corr_en",
                "correu": "corr_eu",
                "corrnu": "corr_nu",
            }
        )
        return df_out.reset_index(drop=True)

    def _download_station_locations(self, filename: PathOrStr, url: str) -> None:
        """Download the station location file from the Nevada Geodetic Laboratory."""
        resp = requests.get(url, timeout=REQUEST_TIMEOUT)
        resp.raise_for_status()

        with open(filename, "w") as f:
            f.write(resp.text)

    def steps(self, station_ids: list[str] | None = None) -> pd.DataFrame:
        """Fetch the UNR database of potential step epochs.

        Parses https://geodesy.unr.edu/NGLStationPages/steps.txt (format:
        https://geodesy.unr.edu/NGLStationPages/steps_readme.txt), which
        lists equipment changes (code 1) and earthquakes near the station
        (code 2).

        Parameters
        ----------
        station_ids : list of str, optional
            Only return steps for these stations.

        Returns
        -------
        pd.DataFrame
            Columns: ``id``, ``date`` (parsed datetime), ``code``,
            ``description``, plus for earthquake entries
            ``threshold_distance``, ``distance_from_eq`` and
            ``magnitude`` (NaN for equipment steps).

        """
        rows = self._read_steps_table()
        df = rows.copy()
        if station_ids is not None:
            wanted = {s.upper() for s in station_ids}
            df = df[df["id"].isin(wanted)].reset_index(drop=True)
        return df

    @staticmethod
    @cache
    def _read_steps_table() -> pd.DataFrame:
        """Download and parse the UNR steps file (cached)."""
        response = requests.get(STEPS_URL, timeout=REQUEST_TIMEOUT)
        response.raise_for_status()

        equipment, earthquakes = [], []
        for line in response.text.splitlines():
            parts = line.split()
            if len(parts) < 4:
                continue
            if len(parts) > 5:  # earthquake entries carry extra columns
                earthquakes.append(parts[:7])
            else:
                equipment.append(parts[:4])

        df_eq = pd.DataFrame(
            earthquakes,
            columns=[
                "id",
                "date",
                "code",
                "threshold_distance",
                "distance_from_eq",
                "magnitude",
                "description",
            ],
        )
        df_env = pd.DataFrame(equipment, columns=["id", "date", "code", "description"])
        df = pd.concat([df_env, df_eq], ignore_index=True)
        df["date"] = pd.to_datetime(df["date"], format="%y%b%d")
        df["code"] = df["code"].astype(int)
        for col in ("threshold_distance", "distance_from_eq", "magnitude"):
            df[col] = pd.to_numeric(df[col], errors="coerce")
        return df.sort_values(["id", "date"], ignore_index=True)

    def get_global_station_list(self) -> list[str]:
        """Get the list of "processed" stations from UNR.

        Source: https://geodesy.unr.edu/NGLStationPages/GlobalStationList

        Note that this may be smaller than the lat/lon/alt list at
        https://geodesy.unr.edu/NGLStationPages/llh.out.
        """
        return self._read_global_station_list().values.ravel().tolist()

    @staticmethod
    @cache
    def _read_global_station_list() -> pd.DataFrame:
        """Read the global station list from UNR."""
        return pd.read_html(
            "https://geodesy.unr.edu/NGLStationPages/GlobalStationList"
        )[0]

download_station_data(station_id, frame='ENU', reference='IGS20', plate_fixed=False, plate=None)

Download GPS station data from the Nevada Geodetic Laboratory.

Parameters:

Name	Type	Description	Default
station_id	str	The station identifier.	required
frame	('ENU', 'XYZ')	The coordinate system of the data to download. Default is "ENU".	"ENU"
reference	('IGS14', 'IGS20')	Geodetic reference of processed data.	"IGS14"
plate_fixed	bool	Whether to download plate-fixed data. Only applicable for "ENU" frame.	False
plate	str	If using plate_fixed, specify which plate to use (in the case of a station on multiple plates). If None, uses the first plate from the UNR results.	None

Source code in src/geepers/gps_sources/unr.py

def download_station_data(
    self,
    station_id: str,
    frame: Literal["ENU", "XYZ"] = "ENU",
    reference: Literal["IGS14", "IGS20"] = "IGS20",
    plate_fixed: bool = False,
    plate: str | None = None,
) -> None:
    """Download GPS station data from the Nevada Geodetic Laboratory.

    Parameters
    ----------
    station_id : str
        The station identifier.
    frame : {"ENU", "XYZ"}, optional
        The coordinate system of the data to download. Default is "ENU".
    reference : {"IGS14", "IGS20"}
        Geodetic reference of processed data.
    plate_fixed : bool, optional
        Whether to download plate-fixed data. Only applicable for "ENU" frame.
    plate : str, optional
        If using plate_fixed, specify which plate to use (in the case of a station
        on multiple plates).
        If None, uses the first plate from the UNR results.

    """
    station_id = validate_station_id(station_id.upper())

    if frame == "ENU":
        if plate_fixed:
            if plate is None:
                plate = self._get_station_plates(station_id)[0]
            url = f"https://geodesy.unr.edu/gps_timeseries/tenv3/plates/{plate}/{station_id}.{plate}.tenv3"
            filename = self._cache_dir / plate / f"{station_id}.tenv3"
        else:
            url = GPS_BASE_URL.format(station=station_id, reference=reference)
            # Hack to get around bad url structure
            url = url.replace("gps_timeseries/IGS14", "gps_timeseries")
            filename = self._cache_dir / f"{station_id}.tenv3"
    elif frame == "XYZ":
        url = f"https://geodesy.unr.edu/gps_timeseries/txyz/{reference}/{station_id}.txyz2"
        filename = self._cache_dir / f"{station_id}.txyz2"
    else:
        msg = "frame must be 'ENU' or 'XYZ'"
        raise ValueError(msg)

    response = requests.get(url, timeout=REQUEST_TIMEOUT)
    response.raise_for_status()

    filename.parent.mkdir(parents=True, exist_ok=True)
    filename.write_text(response.text)
    logger.info(f"Saved {url} to {filename}")

get_global_station_list()

Get the list of "processed" stations from UNR.

Source: https://geodesy.unr.edu/NGLStationPages/GlobalStationList

Note that this may be smaller than the lat/lon/alt list at https://geodesy.unr.edu/NGLStationPages/llh.out.

Source code in src/geepers/gps_sources/unr.py

def get_global_station_list(self) -> list[str]:
    """Get the list of "processed" stations from UNR.

    Source: https://geodesy.unr.edu/NGLStationPages/GlobalStationList

    Note that this may be smaller than the lat/lon/alt list at
    https://geodesy.unr.edu/NGLStationPages/llh.out.
    """
    return self._read_global_station_list().values.ravel().tolist()

steps(station_ids=None)

Fetch the UNR database of potential step epochs.

Parses https://geodesy.unr.edu/NGLStationPages/steps.txt (format: https://geodesy.unr.edu/NGLStationPages/steps_readme.txt), which lists equipment changes (code 1) and earthquakes near the station (code 2).

Parameters:

Name	Type	Description	Default
station_ids	list of str	Only return steps for these stations.	None

Returns:

Type	Description
DataFrame	Columns: id, date (parsed datetime), code, description, plus for earthquake entries threshold_distance, distance_from_eq and magnitude (NaN for equipment steps).

Source code in src/geepers/gps_sources/unr.py

def steps(self, station_ids: list[str] | None = None) -> pd.DataFrame:
    """Fetch the UNR database of potential step epochs.

    Parses https://geodesy.unr.edu/NGLStationPages/steps.txt (format:
    https://geodesy.unr.edu/NGLStationPages/steps_readme.txt), which
    lists equipment changes (code 1) and earthquakes near the station
    (code 2).

    Parameters
    ----------
    station_ids : list of str, optional
        Only return steps for these stations.

    Returns
    -------
    pd.DataFrame
        Columns: ``id``, ``date`` (parsed datetime), ``code``,
        ``description``, plus for earthquake entries
        ``threshold_distance``, ``distance_from_eq`` and
        ``magnitude`` (NaN for equipment steps).

    """
    rows = self._read_steps_table()
    df = rows.copy()
    if station_ids is not None:
        wanted = {s.upper() for s in station_ids}
        df = df[df["id"].isin(wanted)].reset_index(drop=True)
    return df

timeseries(station_id, /, frame='ENU', start_date=None, end_date=None, zero_by='mean', download_if_missing=True, plate_fixed=False, plate_name=None)

Load GPS station time series data.

Parameters:

Name	Type	Description	Default
station_id	str	The station identifier.	required
frame	('ENU', 'XYZ')	Coordinate frame for the data. Default is "ENU".	"ENU"
start_date	str	Start date for data filtering (ISO format).	None
end_date	str	End date for data filtering (ISO format).	None
download_if_missing	bool	Whether to download data if not found locally.	True
zero_by	Literal['mean', 'start']	How to zero the data. Either "mean" or "start".	'mean'
plate_fixed	bool	Whether to use plate-fixed coordinates.	False
plate_name	str	If plate_fixed=True, specify which plate to use. Some stations have multiple plates. If None, uses the first plate found.	None

Returns:

Type	Description
DataFrame	DataFrame validated against StationObservationSchema.

Source code in src/geepers/gps_sources/unr.py

def timeseries(
    self,
    station_id: str,
    /,
    frame: Literal["ENU", "XYZ"] = "ENU",
    start_date: str | None = None,
    end_date: str | None = None,
    zero_by: Literal["mean", "start", "none"] = "mean",
    download_if_missing: bool = True,
    plate_fixed: bool = False,
    plate_name: str | None = None,
) -> pd.DataFrame:
    """Load GPS station time series data.

    Parameters
    ----------
    station_id : str
        The station identifier.
    frame : {"ENU", "XYZ"}, optional
        Coordinate frame for the data. Default is "ENU".
    start_date : str, optional
        Start date for data filtering (ISO format).
    end_date : str, optional
        End date for data filtering (ISO format).
    download_if_missing : bool, optional
        Whether to download data if not found locally.
    zero_by : Literal["mean", "start"], optional
        How to zero the data. Either "mean" or "start".
    plate_fixed : bool, optional
        Whether to use plate-fixed coordinates.
    plate_name : str, optional
        If `plate_fixed=True`, specify which plate to use.
        Some stations have multiple plates.
        If None, uses the first plate found.

    Returns
    -------
    pd.DataFrame
        DataFrame validated against StationObservationSchema.

    """
    if frame not in ["ENU", "XYZ"]:
        msg = f"Unsupported frame: {frame}. Use 'ENU' or 'XYZ'"
        raise ValueError(msg)

    station_id = validate_station_id(station_id.upper())

    plate = None
    if plate_fixed and frame == "ENU":
        plates = self._get_station_plates(station_id)
        if plate_name:
            if plate_name not in plates:
                msg = (
                    f"Plate {plate_name} not found for {station_id}, which has"
                    f" plates {plates}"
                )
                raise ValueError(msg)
            plate = plate_name
        else:
            plate = plates[0]
        gps_data_file = self._cache_dir / plate / f"{station_id}.tenv3"
    else:
        if frame == "ENU":
            gps_data_file = self._cache_dir / f"{station_id}.tenv3"
        else:  # frame in ("XYZ")
            gps_data_file = self._cache_dir / f"{station_id}.txyz2"

    if not gps_data_file.exists():
        if download_if_missing:
            logger.info(f"Downloading {station_id} to {gps_data_file}")
            self.download_station_data(
                station_id, frame=frame, plate_fixed=plate_fixed, plate=plate
            )
        else:
            msg = f"{gps_data_file} does not exist, download_if_missing = False"
            raise ValueError(msg)

    df = pd.read_csv(gps_data_file, sep=r"\s+", engine="c")
    df = self._clean_gps_df(
        df, start_date, end_date, coords="enu" if frame == "ENU" else "xyz"
    )

    if frame == "ENU" and zero_by:
        df = self._zero_data(df, zero_by, columns=["east", "north", "up"])

    if frame == "ENU":
        StationObservationSchema.validate(df, lazy=True)

    return df