NEXRAD radar — API reference

NEXRAD Level-II radar data source subpackage — earthlens.radar. Background and usage are covered under the other pages in this section (Introduction, Usage); this page is the rendered API.

earthlens.radar

NEXRAD Level-II radar backend (real-time WSR-88D chunk feed).

Fetches WSR-88D Level-II radar volumes from the unsigned unidata-nexrad-level2-chunks AWS bucket and assembles each volume's ordered chunks into a single .ar2v file, returning a GeoDataFrame inventory of what was fetched. The feed is near-real-time (a rolling buffer of recent volumes), not a historical archive.

Public surface (re-exported from this package):

• :class:Radar — the backend; instantiate with a time window, a bbox, and a {station_id: [...]} mapping, then call :meth:Radar.download.
• :class:StationCatalog — loader for the bundled radar_data_catalog.yaml.
• :class:Station — one WSR-88D site row (name / lat / lon / state).
• :data:CATALOG_PATH — absolute path to the bundled station YAML.
• :data:BUCKET — the unsigned chunk bucket name.

The [radar] extra pulls boto3 (unsigned S3); it is imported lazily, so the package imports without the extra installed. Reading / gridding the assembled volumes (via pyart) is a downstream follow-on.

Catalog

Bases: AbstractCatalog

Catalog of NEXRAD WSR-88D sites.

Reads the bundled radar_data_catalog.yaml and exposes its stations: block as a typed dict[str, Station]. Instantiate with no arguments (Catalog()).

Examples:

• Look up a site and read its location:

  >>> from earthlens.radar import Catalog
  >>> ktlx = Catalog().get_station("KTLX")
  >>> (round(ktlx.latitude, 2), round(ktlx.longitude, 2))
  (35.33, -97.28)
  

Source code in libs/providers/atmosphere/src/earthlens/radar/catalog.py

class Catalog(AbstractCatalog):
    """Catalog of NEXRAD WSR-88D sites.

    Reads the bundled `radar_data_catalog.yaml` and exposes its `stations:` block
    as a typed `dict[str, Station]`. Instantiate with no arguments
    (`Catalog()`).

    Examples:
        - Look up a site and read its location:
            ```python
            >>> from earthlens.radar import Catalog
            >>> ktlx = Catalog().get_station("KTLX")
            >>> (round(ktlx.latitude, 2), round(ktlx.longitude, 2))
            (35.33, -97.28)

            ```
    """

    _catalog_kind: str = "NEXRAD station catalog"
    _entry_noun: str = "stations"

    datasets: dict[str, Station] = Field(default_factory=dict)

    @classmethod
    def _autoload(cls) -> dict[str, Any]:
        """Read the bundled station registry.

        Returns:
            dict[str, Any]: The `stations:` map keyed by site id.
        """
        return {"datasets": _load_stations(CATALOG_PATH)}

    def get_catalog(self) -> dict[str, Station]:
        """Return the structural per-site map (satisfies the base contract)."""
        return self.datasets

    def get_station(self, site_id: str) -> Station:
        """Resolve a site id to its :class:`Station` (did-you-mean on miss).

        Args:
            site_id: A four-letter WSR-88D id (e.g. `"KTLX"`).

        Returns:
            Station: The resolved site.

        Raises:
            ValueError: When `site_id` is unknown (with a did-you-mean
                hint from the base class).
        """
        return cast("Station", self.get_dataset(site_id))

    def in_bbox(
        self, west: float, south: float, east: float, north: float
    ) -> list[str]:
        """Return the site ids whose location falls inside a bbox.

        Args:
            west: West edge in degrees.
            south: South edge in degrees.
            east: East edge in degrees.
            north: North edge in degrees.

        Returns:
            list[str]: Matching site ids, sorted.

        Examples:
            - Find the catalogued sites over the south-central US:
                ```python
                >>> from earthlens.radar import Catalog
                >>> "KTLX" in Catalog().in_bbox(-100, 33, -95, 37)
                True

                ```
        """
        hits = [
            sid
            for sid, s in self.datasets.items()
            if west <= s.longitude <= east and south <= s.latitude <= north
        ]
        return sorted(hits)

get_catalog()

Return the structural per-site map (satisfies the base contract).

Source code in libs/providers/atmosphere/src/earthlens/radar/catalog.py

def get_catalog(self) -> dict[str, Station]:
    """Return the structural per-site map (satisfies the base contract)."""
    return self.datasets

get_station(site_id)

Resolve a site id to its :class:Station (did-you-mean on miss).

Parameters:

Name	Type	Description	Default
site_id	str	A four-letter WSR-88D id (e.g. "KTLX").	required

Returns:

Name	Type	Description
Station	Station	The resolved site.

Raises:

Type	Description
ValueError	When site_id is unknown (with a did-you-mean hint from the base class).

Source code in libs/providers/atmosphere/src/earthlens/radar/catalog.py

def get_station(self, site_id: str) -> Station:
    """Resolve a site id to its :class:`Station` (did-you-mean on miss).

    Args:
        site_id: A four-letter WSR-88D id (e.g. `"KTLX"`).

    Returns:
        Station: The resolved site.

    Raises:
        ValueError: When `site_id` is unknown (with a did-you-mean
            hint from the base class).
    """
    return cast("Station", self.get_dataset(site_id))

in_bbox(west, south, east, north)

Return the site ids whose location falls inside a bbox.

Parameters:

Name	Type	Description	Default
west	float	West edge in degrees.	required
south	float	South edge in degrees.	required
east	float	East edge in degrees.	required
north	float	North edge in degrees.	required

Returns:

Type	Description
list[str]	list[str]: Matching site ids, sorted.

Examples:

• Find the catalogued sites over the south-central US:

  >>> from earthlens.radar import Catalog
  >>> "KTLX" in Catalog().in_bbox(-100, 33, -95, 37)
  True
  

Source code in libs/providers/atmosphere/src/earthlens/radar/catalog.py

def in_bbox(
    self, west: float, south: float, east: float, north: float
) -> list[str]:
    """Return the site ids whose location falls inside a bbox.

    Args:
        west: West edge in degrees.
        south: South edge in degrees.
        east: East edge in degrees.
        north: North edge in degrees.

    Returns:
        list[str]: Matching site ids, sorted.

    Examples:
        - Find the catalogued sites over the south-central US:
            ```python
            >>> from earthlens.radar import Catalog
            >>> "KTLX" in Catalog().in_bbox(-100, 33, -95, 37)
            True

            ```
    """
    hits = [
        sid
        for sid, s in self.datasets.items()
        if west <= s.longitude <= east and south <= s.latitude <= north
    ]
    return sorted(hits)

Radar

Bases: AbstractDataSource

NEXRAD Level-II radar backend (real-time chunk feed).

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	"vector" — the result is a GeoDataFrame inventory of assembled volumes, not a gridded array, so the facade rejects aggregate= (raw radar volumes are not pyramids-reducible).

Source code in libs/providers/atmosphere/src/earthlens/radar/backend.py

class Radar(AbstractDataSource):
    """NEXRAD Level-II radar backend (real-time chunk feed).

    Attributes:
        OUTPUT_KIND: `"vector"` — the result is a `GeoDataFrame`
            inventory of assembled volumes, not a gridded array, so the
            facade rejects `aggregate=` (raw radar volumes are not
            pyramids-reducible).
    """

    OUTPUT_KIND: OutputKind = "vector"

    AGGREGATE_REFUSAL_REASON = (
        "raw Level-II volumes are not griddable by the pyramids reducer"
    )

    def __init__(
        self,
        start: str,
        end: str,
        variables: dict[str, list[str]],
        lat_lim: list[float],
        lon_lim: list[float],
        temporal_resolution: str = "raw",
        path: Path | str | None = None,
        fmt: str = "%Y-%m-%dT%H:%M:%S",
        *,
        region: str = "us-east-1",
        catalog: Catalog | None = None,
    ):
        """Initialise a radar backend instance.

        Args:
            start: Inclusive start of the scan-time window (parsed with
                `fmt`).
            end: Inclusive end of the scan-time window.
            variables: Mapping from WSR-88D site id to an (advisory)
                moment list, e.g. `{"KTLX": ["reflectivity"]}`.
            lat_lim: `[lat_min, lat_max]` in degrees.
            lon_lim: `[lon_min, lon_max]` in degrees.
            temporal_resolution: Advisory label (radar volumes are
                `"raw"`).
            path: Output directory for the assembled `.ar2v` files.
            fmt: `strptime` format for `start` / `end`. Defaults to an
                ISO datetime (`"%Y-%m-%dT%H:%M:%S"`) since the feed is
                sub-hourly real-time.
            region: AWS region of the chunk bucket.
            catalog: Optional pre-built :class:`Catalog` (tests
                inject a faked one).

        Raises:
            ValueError: When `variables` is empty.
        """
        if not variables:
            raise ValueError(
                "Radar requires a non-empty `variables` mapping of {station_id: [...]}."
            )
        self._region = region
        self._catalog = catalog if catalog is not None else Catalog()
        self._stations: list[tuple[str, Station | None]] = [
            (site_id, self._catalog.datasets.get(site_id)) for site_id in variables
        ]
        super().__init__(
            start=start,
            end=end,
            variables=variables,
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim,
            lon_lim=lon_lim,
            fmt=fmt,
            path=path,
        )

    def _check_input_dates(
        self, start: str, end: str, temporal_resolution: str, fmt: str
    ) -> TemporalExtent:
        """Parse the scan-time window into a :class:`TemporalExtent`.

        Args:
            start: Inclusive window start.
            end: Inclusive window end.
            temporal_resolution: Advisory cadence label.
            fmt: `strptime` format tried first for a string `start` /
                `end`; a non-matching string falls back to an ISO-8601
                parse, and a `datetime` / `date` ignores it.

        Returns:
            TemporalExtent: Frozen model with parsed bounds.

        Raises:
            ValueError: If `start` parses later than `end`.
        """
        self._end_is_date_only = end_is_date_only(end)
        return self._whole_window_extent(start, end, fmt=fmt, resolution="raw")

    def _window(self) -> tuple[dt.datetime, dt.datetime]:
        """Return the inclusive scan-time window.

        A date-only `end` covers its whole calendar day; an `end` that names a
        time of day means that instant and is returned unchanged.

        Returns:
            tuple[datetime.datetime, datetime.datetime]: The inclusive
                `(start, end)` scan-time bounds.
        """
        # A date-only end (midnight) would exclude the whole day's volumes; an
        # end naming a time means that instant and is left alone.
        end = expand_bare_date_end(self.time.end_date, date_only=self._end_is_date_only)
        return self.time.start_date, end

    def _search(self) -> list[RemoteProduct]:
        """List each station's volumes in the window; one product per volume.

        For every requested site, lists the volume prefixes under
        `{STATION}/`, then lists each volume's chunks, parses the
        volume scan start time, and keeps the volumes whose start time
        falls in the request window. The ordered chunk keys ride on the
        product metadata so :meth:`_fetch` needs no re-listing.

        Returns:
            list[RemoteProduct]: One product per in-window volume, each
                carrying `station`, `volume`, `start_time`, ordered
                `chunk_keys`, and the `station` row (for geometry).
        """
        client = _s3_client(self._region)
        start, end = self._window()
        products: list[RemoteProduct] = []
        for site_id, station in self._stations:
            volume_prefixes = self._list_prefixes(client, f"{site_id}/")
            for vp in volume_prefixes:
                # Read just the first chunk key (the `S` chunk carries the scan
                # start) to window-filter cheaply; only list a volume's full
                # chunk set once it is known to be in range (avoids the N+1
                # full-listing of every volume).
                first = self._first_key(client, vp)
                if first is None:
                    continue
                scan = _volume_start(first)
                if not (start <= scan <= end):
                    continue
                chunk_keys = sorted(self._list_keys(client, vp))
                if not chunk_keys:
                    continue
                volume = vp.rstrip("/").rsplit("/", 1)[-1]
                products.append(
                    RemoteProduct(
                        id=f"{site_id}.{volume}",
                        metadata={
                            "station_id": site_id,
                            "station": station,
                            "volume": volume,
                            "start_time": scan,
                            "chunk_keys": chunk_keys,
                        },
                    )
                )
        return products

    @staticmethod
    def _list_prefixes(client: Any, prefix: str) -> list[str]:
        """Return the immediate sub-prefixes under `prefix` (paginated)."""
        out: list[str] = []
        token: str | None = None
        while True:
            kwargs = {"Bucket": BUCKET, "Prefix": prefix, "Delimiter": "/"}
            if token:
                kwargs["ContinuationToken"] = token
            resp = client.list_objects_v2(**kwargs)
            out.extend(c["Prefix"] for c in resp.get("CommonPrefixes", []))
            token = resp.get("NextContinuationToken")
            if not resp.get("IsTruncated"):
                break
        return out

    @staticmethod
    def _first_key(client: Any, prefix: str) -> str | None:
        """Return the lexicographically first object key under `prefix`.

        Used to read a volume's `S` (start) chunk — which carries the
        scan-start timestamp — without listing the whole volume.

        Args:
            client: The S3 client.
            prefix: A volume prefix (`"KTLX/871/"`).

        Returns:
            str | None: The first key, or `None` if the prefix is empty.
        """
        resp = client.list_objects_v2(Bucket=BUCKET, Prefix=prefix, MaxKeys=1)
        contents = resp.get("Contents", [])
        return contents[0]["Key"] if contents else None

    @staticmethod
    def _list_keys(client: Any, prefix: str) -> list[str]:
        """Return all object keys under `prefix` (paginated)."""
        out: list[str] = []
        token: str | None = None
        while True:
            kwargs = {"Bucket": BUCKET, "Prefix": prefix}
            if token:
                kwargs["ContinuationToken"] = token
            resp = client.list_objects_v2(**kwargs)
            out.extend(o["Key"] for o in resp.get("Contents", []))
            token = resp.get("NextContinuationToken")
            if not resp.get("IsTruncated"):
                break
        return out

    def _fetch(self, products: list[RemoteProduct]) -> list[Path]:
        """Assemble each volume's chunks into one `.ar2v` file.

        Per volume: download the ordered chunks and concatenate them
        (atomically, via a `.part` rename) into a single Level-II file.
        A volume whose download fails is logged and skipped so one bad
        volume does not lose the others (mirrors the FDSN/NWP policy).

        Args:
            products: The products from :meth:`_search`.

        Returns:
            list[Path]: One assembled `.ar2v` path per successfully
                fetched volume, in product order.
        """
        return [path for _, path in self._fetch_pairs(products)]

    def _fetch_pairs(
        self, products: list[RemoteProduct]
    ) -> list[tuple[RemoteProduct, Path]]:
        """Assemble each volume, returning `(product, path)` pairs.

        Like :meth:`_fetch` but keeps each path paired with the product
        it came from, so the inventory needs no filename re-matching. A
        volume whose download fails is logged and skipped. The loop shows
        a `tqdm` bar unless `download(progress_bar=False)` disabled it.

        Args:
            products: The products from :meth:`_search`.

        Returns:
            list[tuple[RemoteProduct, Path]]: One pair per successfully
                assembled volume, in product order.
        """
        client = _s3_client(self._region)
        pairs, _failures = self._run_items(
            list(
                tqdm(
                    products,
                    disable=not getattr(self, "_show_progress", True),
                    desc="radar",
                )
            ),
            partial(self._assemble_pair, client),
            errors=self._errors,
            label="volume",
            describe=_describe_product,
        )
        return cast("list[tuple[RemoteProduct, Path]]", pairs)

    def _assemble_pair(
        self, client: Any, product: RemoteProduct
    ) -> tuple[RemoteProduct, Path]:
        """Assemble one volume, keeping it paired with its product.

        Args:
            client: The unsigned S3 client shared across the batch.
            product: The product to assemble.

        Returns:
            tuple[RemoteProduct, Path]: The product and the `.ar2v` written
                for it.
        """
        return product, self._assemble(client, product)

    def _assemble(self, client: Any, product: RemoteProduct) -> Path:
        """Download + concatenate one volume's chunks into a `.ar2v` file."""
        meta = product.metadata
        target = self.root_dir / (
            f"{meta['station_id']}_{meta['start_time']:%Y%m%d_%H%M%S}.ar2v"
        )
        tmp = target.with_name(target.name + ".part")
        try:
            with open(tmp, "wb") as handle:
                for key in meta["chunk_keys"]:
                    handle.write(
                        client.get_object(Bucket=BUCKET, Key=key)["Body"].read()
                    )
            tmp.replace(target)
        except BaseException:
            tmp.unlink(missing_ok=True)
            raise
        return target

    def download(
        self,
        progress_bar: bool = True,
        errors: str = "warn",
    ) -> gpd.GeoDataFrame:
        """Assemble the in-window volumes and return a `GeoDataFrame` inventory.

        Args:
            progress_bar: Unused (kept for interface parity).
            errors: Partial-failure policy across the in-window volumes —
                `"warn"` (default) logs each failed assembly and continues,
                `"raise"` propagates the first, `"ignore"` continues
                silently.

        Returns:
            geopandas.GeoDataFrame: One row per assembled volume —
                `station_id`, `volume`, `scan_time`, `n_chunks`, `path`,
                and a station-point `geometry` (`None` for sites absent
                from the catalog). Empty (with the right columns) when no
                volume falls in the window.

                Intentionally a `GeoDataFrame` inventory (one row per
                downloaded volume + its on-disk `path`), not the
                `FeatureCollection` the event/footprint `"vector"`
                backends (FDSN, FIRMS, GDACS) return — the rows index
                bulky files rather than describe point/polygon features.
                The base `download` contract lists radar as this
                documented exception.

        Raises:
            ValueError: If `errors` is not a recognised policy.
        """
        self._errors = self.check_errors_policy(errors)
        self._show_progress = progress_bar
        products = self._search()
        pairs = self._fetch_pairs(products)
        return self._inventory(pairs)

    @staticmethod
    def _inventory(pairs: list[tuple[RemoteProduct, Path]]):
        """Build the GeoDataFrame inventory from `(product, path)` pairs.

        Each path is already paired with the product it came from (no
        filename re-matching), so the metadata attaches unambiguously.
        """
        import geopandas as gpd
        from shapely.geometry import Point

        rows = []
        geoms = []
        for product, path in pairs:
            meta = product.metadata
            station: Station | None = meta.get("station")
            rows.append(
                {
                    "station_id": meta.get("station_id"),
                    "volume": meta.get("volume"),
                    "scan_time": meta.get("start_time"),
                    "n_chunks": len(meta.get("chunk_keys", [])),
                    "path": str(path),
                }
            )
            geoms.append(
                Point(station.longitude, station.latitude) if station else None
            )
        return gpd.GeoDataFrame(rows, geometry=geoms, crs="EPSG:4326")

__init__(start, end, variables, lat_lim, lon_lim, temporal_resolution='raw', path=None, fmt='%Y-%m-%dT%H:%M:%S', *, region='us-east-1', catalog=None)

Initialise a radar backend instance.

Parameters:

Name	Type	Description	Default
start	str	Inclusive start of the scan-time window (parsed with fmt).	required
end	str	Inclusive end of the scan-time window.	required
variables	dict[str, list[str]]	Mapping from WSR-88D site id to an (advisory) moment list, e.g. {"KTLX": ["reflectivity"]}.	required
lat_lim	list[float]	[lat_min, lat_max] in degrees.	required
lon_lim	list[float]	[lon_min, lon_max] in degrees.	required
temporal_resolution	str	Advisory label (radar volumes are "raw").	'raw'
path	Path | str | None	Output directory for the assembled .ar2v files.	None
fmt	str	strptime format for start / end. Defaults to an ISO datetime ("%Y-%m-%dT%H:%M:%S") since the feed is sub-hourly real-time.	'%Y-%m-%dT%H:%M:%S'
region	str	AWS region of the chunk bucket.	'us-east-1'
catalog	Catalog | None	Optional pre-built :class:Catalog (tests inject a faked one).	None

Raises:

Type	Description
ValueError	When variables is empty.

Source code in libs/providers/atmosphere/src/earthlens/radar/backend.py

def __init__(
    self,
    start: str,
    end: str,
    variables: dict[str, list[str]],
    lat_lim: list[float],
    lon_lim: list[float],
    temporal_resolution: str = "raw",
    path: Path | str | None = None,
    fmt: str = "%Y-%m-%dT%H:%M:%S",
    *,
    region: str = "us-east-1",
    catalog: Catalog | None = None,
):
    """Initialise a radar backend instance.

    Args:
        start: Inclusive start of the scan-time window (parsed with
            `fmt`).
        end: Inclusive end of the scan-time window.
        variables: Mapping from WSR-88D site id to an (advisory)
            moment list, e.g. `{"KTLX": ["reflectivity"]}`.
        lat_lim: `[lat_min, lat_max]` in degrees.
        lon_lim: `[lon_min, lon_max]` in degrees.
        temporal_resolution: Advisory label (radar volumes are
            `"raw"`).
        path: Output directory for the assembled `.ar2v` files.
        fmt: `strptime` format for `start` / `end`. Defaults to an
            ISO datetime (`"%Y-%m-%dT%H:%M:%S"`) since the feed is
            sub-hourly real-time.
        region: AWS region of the chunk bucket.
        catalog: Optional pre-built :class:`Catalog` (tests
            inject a faked one).

    Raises:
        ValueError: When `variables` is empty.
    """
    if not variables:
        raise ValueError(
            "Radar requires a non-empty `variables` mapping of {station_id: [...]}."
        )
    self._region = region
    self._catalog = catalog if catalog is not None else Catalog()
    self._stations: list[tuple[str, Station | None]] = [
        (site_id, self._catalog.datasets.get(site_id)) for site_id in variables
    ]
    super().__init__(
        start=start,
        end=end,
        variables=variables,
        temporal_resolution=temporal_resolution,
        lat_lim=lat_lim,
        lon_lim=lon_lim,
        fmt=fmt,
        path=path,
    )

download(progress_bar=True, errors='warn')

Assemble the in-window volumes and return a GeoDataFrame inventory.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Unused (kept for interface parity).	True
errors	str	Partial-failure policy across the in-window volumes — "warn" (default) logs each failed assembly and continues, "raise" propagates the first, "ignore" continues silently.	'warn'

Returns:

Type

Description

GeoDataFrame

geopandas.GeoDataFrame: One row per assembled volume — station_id, volume, scan_time, n_chunks, path, and a station-point geometry (None for sites absent from the catalog). Empty (with the right columns) when no volume falls in the window. Intentionally a GeoDataFrame inventory (one row per downloaded volume + its on-disk path), not the FeatureCollection the event/footprint "vector" backends (FDSN, FIRMS, GDACS) return — the rows index bulky files rather than describe point/polygon features. The base download contract lists radar as this documented exception.

Raises:

Type	Description
ValueError	If errors is not a recognised policy.

Source code in libs/providers/atmosphere/src/earthlens/radar/backend.py

def download(
    self,
    progress_bar: bool = True,
    errors: str = "warn",
) -> gpd.GeoDataFrame:
    """Assemble the in-window volumes and return a `GeoDataFrame` inventory.

    Args:
        progress_bar: Unused (kept for interface parity).
        errors: Partial-failure policy across the in-window volumes —
            `"warn"` (default) logs each failed assembly and continues,
            `"raise"` propagates the first, `"ignore"` continues
            silently.

    Returns:
        geopandas.GeoDataFrame: One row per assembled volume —
            `station_id`, `volume`, `scan_time`, `n_chunks`, `path`,
            and a station-point `geometry` (`None` for sites absent
            from the catalog). Empty (with the right columns) when no
            volume falls in the window.

            Intentionally a `GeoDataFrame` inventory (one row per
            downloaded volume + its on-disk `path`), not the
            `FeatureCollection` the event/footprint `"vector"`
            backends (FDSN, FIRMS, GDACS) return — the rows index
            bulky files rather than describe point/polygon features.
            The base `download` contract lists radar as this
            documented exception.

    Raises:
        ValueError: If `errors` is not a recognised policy.
    """
    self._errors = self.check_errors_policy(errors)
    self._show_progress = progress_bar
    products = self._search()
    pairs = self._fetch_pairs(products)
    return self._inventory(pairs)

Station

Bases: BaseModel

One WSR-88D radar site.

The site id (e.g. "KTLX") is the parent key in :attr:Catalog.datasets and is not stored on the row.

Attributes:

Name	Type	Description
name	str	Human-readable site name / location.
latitude	float	Site latitude in degrees (south negative).
longitude	float	Site longitude in degrees (west negative).
state	str	Two-letter US state / territory code.

Source code in libs/providers/atmosphere/src/earthlens/radar/catalog.py

class Station(BaseModel):
    """One WSR-88D radar site.

    The site id (e.g. `"KTLX"`) is the parent key in
    :attr:`Catalog.datasets` and is not stored on the row.

    Attributes:
        name: Human-readable site name / location.
        latitude: Site latitude in degrees (south negative).
        longitude: Site longitude in degrees (west negative).
        state: Two-letter US state / territory code.
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    name: str = ""
    latitude: float = Field(ge=-90.0, le=90.0)
    longitude: float = Field(ge=-180.0, le=180.0)
    state: str = ""

earthlens.radar.backend

Backend that assembles NEXRAD Level-II radar volumes from the chunk feed.

Radar(AbstractDataSource) fetches WSR-88D Level-II volumes from the unsigned unidata-nexrad-level2-chunks AWS bucket. That bucket is a near-real-time rolling buffer: each volume is delivered as ordered chunks — one S (start, carrying the AR2V… volume header), many I (intermediate), and a final E (end) — under {STATION}/{VOLUME}/{YYYYMMDD}-{HHMMSS}-{CHUNK:03d}-{TYPE}. Concatenating a volume's chunks in chunk order reconstructs a valid .ar2v Level-II file (verified: the assembled stream starts with AR2V0006).

The request is variables = {station_id: [...]} (e.g. {"KTLX": []}); the list value is advisory — a Level-II volume carries every moment, so the whole volume is fetched. start / end filter volumes by their scan start time. The result is a GeoDataFrame inventory of the assembled volumes (one row per volume: station, scan time, chunk count, local path, station-point geometry), so OUTPUT_KIND = "vector".

Real-time only. The chunks bucket holds roughly the last hour or two of volumes, not a historical archive, so a request for an old date returns nothing. (The archival noaa-nexrad-level2 bucket denies anonymous listing, so it is not used here.) Reading / gridding the assembled volumes (via pyart) is a downstream follow-on; this backend only fetches and inventories them.

Radar

Bases: AbstractDataSource

NEXRAD Level-II radar backend (real-time chunk feed).

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	"vector" — the result is a GeoDataFrame inventory of assembled volumes, not a gridded array, so the facade rejects aggregate= (raw radar volumes are not pyramids-reducible).

Source code in libs/providers/atmosphere/src/earthlens/radar/backend.py

class Radar(AbstractDataSource):
    """NEXRAD Level-II radar backend (real-time chunk feed).

    Attributes:
        OUTPUT_KIND: `"vector"` — the result is a `GeoDataFrame`
            inventory of assembled volumes, not a gridded array, so the
            facade rejects `aggregate=` (raw radar volumes are not
            pyramids-reducible).
    """

    OUTPUT_KIND: OutputKind = "vector"

    AGGREGATE_REFUSAL_REASON = (
        "raw Level-II volumes are not griddable by the pyramids reducer"
    )

    def __init__(
        self,
        start: str,
        end: str,
        variables: dict[str, list[str]],
        lat_lim: list[float],
        lon_lim: list[float],
        temporal_resolution: str = "raw",
        path: Path | str | None = None,
        fmt: str = "%Y-%m-%dT%H:%M:%S",
        *,
        region: str = "us-east-1",
        catalog: Catalog | None = None,
    ):
        """Initialise a radar backend instance.

        Args:
            start: Inclusive start of the scan-time window (parsed with
                `fmt`).
            end: Inclusive end of the scan-time window.
            variables: Mapping from WSR-88D site id to an (advisory)
                moment list, e.g. `{"KTLX": ["reflectivity"]}`.
            lat_lim: `[lat_min, lat_max]` in degrees.
            lon_lim: `[lon_min, lon_max]` in degrees.
            temporal_resolution: Advisory label (radar volumes are
                `"raw"`).
            path: Output directory for the assembled `.ar2v` files.
            fmt: `strptime` format for `start` / `end`. Defaults to an
                ISO datetime (`"%Y-%m-%dT%H:%M:%S"`) since the feed is
                sub-hourly real-time.
            region: AWS region of the chunk bucket.
            catalog: Optional pre-built :class:`Catalog` (tests
                inject a faked one).

        Raises:
            ValueError: When `variables` is empty.
        """
        if not variables:
            raise ValueError(
                "Radar requires a non-empty `variables` mapping of {station_id: [...]}."
            )
        self._region = region
        self._catalog = catalog if catalog is not None else Catalog()
        self._stations: list[tuple[str, Station | None]] = [
            (site_id, self._catalog.datasets.get(site_id)) for site_id in variables
        ]
        super().__init__(
            start=start,
            end=end,
            variables=variables,
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim,
            lon_lim=lon_lim,
            fmt=fmt,
            path=path,
        )

    def _check_input_dates(
        self, start: str, end: str, temporal_resolution: str, fmt: str
    ) -> TemporalExtent:
        """Parse the scan-time window into a :class:`TemporalExtent`.

        Args:
            start: Inclusive window start.
            end: Inclusive window end.
            temporal_resolution: Advisory cadence label.
            fmt: `strptime` format tried first for a string `start` /
                `end`; a non-matching string falls back to an ISO-8601
                parse, and a `datetime` / `date` ignores it.

        Returns:
            TemporalExtent: Frozen model with parsed bounds.

        Raises:
            ValueError: If `start` parses later than `end`.
        """
        self._end_is_date_only = end_is_date_only(end)
        return self._whole_window_extent(start, end, fmt=fmt, resolution="raw")

    def _window(self) -> tuple[dt.datetime, dt.datetime]:
        """Return the inclusive scan-time window.

        A date-only `end` covers its whole calendar day; an `end` that names a
        time of day means that instant and is returned unchanged.

        Returns:
            tuple[datetime.datetime, datetime.datetime]: The inclusive
                `(start, end)` scan-time bounds.
        """
        # A date-only end (midnight) would exclude the whole day's volumes; an
        # end naming a time means that instant and is left alone.
        end = expand_bare_date_end(self.time.end_date, date_only=self._end_is_date_only)
        return self.time.start_date, end

    def _search(self) -> list[RemoteProduct]:
        """List each station's volumes in the window; one product per volume.

        For every requested site, lists the volume prefixes under
        `{STATION}/`, then lists each volume's chunks, parses the
        volume scan start time, and keeps the volumes whose start time
        falls in the request window. The ordered chunk keys ride on the
        product metadata so :meth:`_fetch` needs no re-listing.

        Returns:
            list[RemoteProduct]: One product per in-window volume, each
                carrying `station`, `volume`, `start_time`, ordered
                `chunk_keys`, and the `station` row (for geometry).
        """
        client = _s3_client(self._region)
        start, end = self._window()
        products: list[RemoteProduct] = []
        for site_id, station in self._stations:
            volume_prefixes = self._list_prefixes(client, f"{site_id}/")
            for vp in volume_prefixes:
                # Read just the first chunk key (the `S` chunk carries the scan
                # start) to window-filter cheaply; only list a volume's full
                # chunk set once it is known to be in range (avoids the N+1
                # full-listing of every volume).
                first = self._first_key(client, vp)
                if first is None:
                    continue
                scan = _volume_start(first)
                if not (start <= scan <= end):
                    continue
                chunk_keys = sorted(self._list_keys(client, vp))
                if not chunk_keys:
                    continue
                volume = vp.rstrip("/").rsplit("/", 1)[-1]
                products.append(
                    RemoteProduct(
                        id=f"{site_id}.{volume}",
                        metadata={
                            "station_id": site_id,
                            "station": station,
                            "volume": volume,
                            "start_time": scan,
                            "chunk_keys": chunk_keys,
                        },
                    )
                )
        return products

    @staticmethod
    def _list_prefixes(client: Any, prefix: str) -> list[str]:
        """Return the immediate sub-prefixes under `prefix` (paginated)."""
        out: list[str] = []
        token: str | None = None
        while True:
            kwargs = {"Bucket": BUCKET, "Prefix": prefix, "Delimiter": "/"}
            if token:
                kwargs["ContinuationToken"] = token
            resp = client.list_objects_v2(**kwargs)
            out.extend(c["Prefix"] for c in resp.get("CommonPrefixes", []))
            token = resp.get("NextContinuationToken")
            if not resp.get("IsTruncated"):
                break
        return out

    @staticmethod
    def _first_key(client: Any, prefix: str) -> str | None:
        """Return the lexicographically first object key under `prefix`.

        Used to read a volume's `S` (start) chunk — which carries the
        scan-start timestamp — without listing the whole volume.

        Args:
            client: The S3 client.
            prefix: A volume prefix (`"KTLX/871/"`).

        Returns:
            str | None: The first key, or `None` if the prefix is empty.
        """
        resp = client.list_objects_v2(Bucket=BUCKET, Prefix=prefix, MaxKeys=1)
        contents = resp.get("Contents", [])
        return contents[0]["Key"] if contents else None

    @staticmethod
    def _list_keys(client: Any, prefix: str) -> list[str]:
        """Return all object keys under `prefix` (paginated)."""
        out: list[str] = []
        token: str | None = None
        while True:
            kwargs = {"Bucket": BUCKET, "Prefix": prefix}
            if token:
                kwargs["ContinuationToken"] = token
            resp = client.list_objects_v2(**kwargs)
            out.extend(o["Key"] for o in resp.get("Contents", []))
            token = resp.get("NextContinuationToken")
            if not resp.get("IsTruncated"):
                break
        return out

    def _fetch(self, products: list[RemoteProduct]) -> list[Path]:
        """Assemble each volume's chunks into one `.ar2v` file.

        Per volume: download the ordered chunks and concatenate them
        (atomically, via a `.part` rename) into a single Level-II file.
        A volume whose download fails is logged and skipped so one bad
        volume does not lose the others (mirrors the FDSN/NWP policy).

        Args:
            products: The products from :meth:`_search`.

        Returns:
            list[Path]: One assembled `.ar2v` path per successfully
                fetched volume, in product order.
        """
        return [path for _, path in self._fetch_pairs(products)]

    def _fetch_pairs(
        self, products: list[RemoteProduct]
    ) -> list[tuple[RemoteProduct, Path]]:
        """Assemble each volume, returning `(product, path)` pairs.

        Like :meth:`_fetch` but keeps each path paired with the product
        it came from, so the inventory needs no filename re-matching. A
        volume whose download fails is logged and skipped. The loop shows
        a `tqdm` bar unless `download(progress_bar=False)` disabled it.

        Args:
            products: The products from :meth:`_search`.

        Returns:
            list[tuple[RemoteProduct, Path]]: One pair per successfully
                assembled volume, in product order.
        """
        client = _s3_client(self._region)
        pairs, _failures = self._run_items(
            list(
                tqdm(
                    products,
                    disable=not getattr(self, "_show_progress", True),
                    desc="radar",
                )
            ),
            partial(self._assemble_pair, client),
            errors=self._errors,
            label="volume",
            describe=_describe_product,
        )
        return cast("list[tuple[RemoteProduct, Path]]", pairs)

    def _assemble_pair(
        self, client: Any, product: RemoteProduct
    ) -> tuple[RemoteProduct, Path]:
        """Assemble one volume, keeping it paired with its product.

        Args:
            client: The unsigned S3 client shared across the batch.
            product: The product to assemble.

        Returns:
            tuple[RemoteProduct, Path]: The product and the `.ar2v` written
                for it.
        """
        return product, self._assemble(client, product)

    def _assemble(self, client: Any, product: RemoteProduct) -> Path:
        """Download + concatenate one volume's chunks into a `.ar2v` file."""
        meta = product.metadata
        target = self.root_dir / (
            f"{meta['station_id']}_{meta['start_time']:%Y%m%d_%H%M%S}.ar2v"
        )
        tmp = target.with_name(target.name + ".part")
        try:
            with open(tmp, "wb") as handle:
                for key in meta["chunk_keys"]:
                    handle.write(
                        client.get_object(Bucket=BUCKET, Key=key)["Body"].read()
                    )
            tmp.replace(target)
        except BaseException:
            tmp.unlink(missing_ok=True)
            raise
        return target

    def download(
        self,
        progress_bar: bool = True,
        errors: str = "warn",
    ) -> gpd.GeoDataFrame:
        """Assemble the in-window volumes and return a `GeoDataFrame` inventory.

        Args:
            progress_bar: Unused (kept for interface parity).
            errors: Partial-failure policy across the in-window volumes —
                `"warn"` (default) logs each failed assembly and continues,
                `"raise"` propagates the first, `"ignore"` continues
                silently.

        Returns:
            geopandas.GeoDataFrame: One row per assembled volume —
                `station_id`, `volume`, `scan_time`, `n_chunks`, `path`,
                and a station-point `geometry` (`None` for sites absent
                from the catalog). Empty (with the right columns) when no
                volume falls in the window.

                Intentionally a `GeoDataFrame` inventory (one row per
                downloaded volume + its on-disk `path`), not the
                `FeatureCollection` the event/footprint `"vector"`
                backends (FDSN, FIRMS, GDACS) return — the rows index
                bulky files rather than describe point/polygon features.
                The base `download` contract lists radar as this
                documented exception.

        Raises:
            ValueError: If `errors` is not a recognised policy.
        """
        self._errors = self.check_errors_policy(errors)
        self._show_progress = progress_bar
        products = self._search()
        pairs = self._fetch_pairs(products)
        return self._inventory(pairs)

    @staticmethod
    def _inventory(pairs: list[tuple[RemoteProduct, Path]]):
        """Build the GeoDataFrame inventory from `(product, path)` pairs.

        Each path is already paired with the product it came from (no
        filename re-matching), so the metadata attaches unambiguously.
        """
        import geopandas as gpd
        from shapely.geometry import Point

        rows = []
        geoms = []
        for product, path in pairs:
            meta = product.metadata
            station: Station | None = meta.get("station")
            rows.append(
                {
                    "station_id": meta.get("station_id"),
                    "volume": meta.get("volume"),
                    "scan_time": meta.get("start_time"),
                    "n_chunks": len(meta.get("chunk_keys", [])),
                    "path": str(path),
                }
            )
            geoms.append(
                Point(station.longitude, station.latitude) if station else None
            )
        return gpd.GeoDataFrame(rows, geometry=geoms, crs="EPSG:4326")

__init__(start, end, variables, lat_lim, lon_lim, temporal_resolution='raw', path=None, fmt='%Y-%m-%dT%H:%M:%S', *, region='us-east-1', catalog=None)

Initialise a radar backend instance.

Parameters:

Name	Type	Description	Default
start	str	Inclusive start of the scan-time window (parsed with fmt).	required
end	str	Inclusive end of the scan-time window.	required
variables	dict[str, list[str]]	Mapping from WSR-88D site id to an (advisory) moment list, e.g. {"KTLX": ["reflectivity"]}.	required
lat_lim	list[float]	[lat_min, lat_max] in degrees.	required
lon_lim	list[float]	[lon_min, lon_max] in degrees.	required
temporal_resolution	str	Advisory label (radar volumes are "raw").	'raw'
path	Path | str | None	Output directory for the assembled .ar2v files.	None
fmt	str	strptime format for start / end. Defaults to an ISO datetime ("%Y-%m-%dT%H:%M:%S") since the feed is sub-hourly real-time.	'%Y-%m-%dT%H:%M:%S'
region	str	AWS region of the chunk bucket.	'us-east-1'
catalog	Catalog | None	Optional pre-built :class:Catalog (tests inject a faked one).	None

Raises:

Type	Description
ValueError	When variables is empty.

Source code in libs/providers/atmosphere/src/earthlens/radar/backend.py

def __init__(
    self,
    start: str,
    end: str,
    variables: dict[str, list[str]],
    lat_lim: list[float],
    lon_lim: list[float],
    temporal_resolution: str = "raw",
    path: Path | str | None = None,
    fmt: str = "%Y-%m-%dT%H:%M:%S",
    *,
    region: str = "us-east-1",
    catalog: Catalog | None = None,
):
    """Initialise a radar backend instance.

    Args:
        start: Inclusive start of the scan-time window (parsed with
            `fmt`).
        end: Inclusive end of the scan-time window.
        variables: Mapping from WSR-88D site id to an (advisory)
            moment list, e.g. `{"KTLX": ["reflectivity"]}`.
        lat_lim: `[lat_min, lat_max]` in degrees.
        lon_lim: `[lon_min, lon_max]` in degrees.
        temporal_resolution: Advisory label (radar volumes are
            `"raw"`).
        path: Output directory for the assembled `.ar2v` files.
        fmt: `strptime` format for `start` / `end`. Defaults to an
            ISO datetime (`"%Y-%m-%dT%H:%M:%S"`) since the feed is
            sub-hourly real-time.
        region: AWS region of the chunk bucket.
        catalog: Optional pre-built :class:`Catalog` (tests
            inject a faked one).

    Raises:
        ValueError: When `variables` is empty.
    """
    if not variables:
        raise ValueError(
            "Radar requires a non-empty `variables` mapping of {station_id: [...]}."
        )
    self._region = region
    self._catalog = catalog if catalog is not None else Catalog()
    self._stations: list[tuple[str, Station | None]] = [
        (site_id, self._catalog.datasets.get(site_id)) for site_id in variables
    ]
    super().__init__(
        start=start,
        end=end,
        variables=variables,
        temporal_resolution=temporal_resolution,
        lat_lim=lat_lim,
        lon_lim=lon_lim,
        fmt=fmt,
        path=path,
    )

download(progress_bar=True, errors='warn')

Assemble the in-window volumes and return a GeoDataFrame inventory.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Unused (kept for interface parity).	True
errors	str	Partial-failure policy across the in-window volumes — "warn" (default) logs each failed assembly and continues, "raise" propagates the first, "ignore" continues silently.	'warn'

Returns:

Type

Description

GeoDataFrame

geopandas.GeoDataFrame: One row per assembled volume — station_id, volume, scan_time, n_chunks, path, and a station-point geometry (None for sites absent from the catalog). Empty (with the right columns) when no volume falls in the window. Intentionally a GeoDataFrame inventory (one row per downloaded volume + its on-disk path), not the FeatureCollection the event/footprint "vector" backends (FDSN, FIRMS, GDACS) return — the rows index bulky files rather than describe point/polygon features. The base download contract lists radar as this documented exception.

Raises:

Type	Description
ValueError	If errors is not a recognised policy.

Source code in libs/providers/atmosphere/src/earthlens/radar/backend.py

def download(
    self,
    progress_bar: bool = True,
    errors: str = "warn",
) -> gpd.GeoDataFrame:
    """Assemble the in-window volumes and return a `GeoDataFrame` inventory.

    Args:
        progress_bar: Unused (kept for interface parity).
        errors: Partial-failure policy across the in-window volumes —
            `"warn"` (default) logs each failed assembly and continues,
            `"raise"` propagates the first, `"ignore"` continues
            silently.

    Returns:
        geopandas.GeoDataFrame: One row per assembled volume —
            `station_id`, `volume`, `scan_time`, `n_chunks`, `path`,
            and a station-point `geometry` (`None` for sites absent
            from the catalog). Empty (with the right columns) when no
            volume falls in the window.

            Intentionally a `GeoDataFrame` inventory (one row per
            downloaded volume + its on-disk `path`), not the
            `FeatureCollection` the event/footprint `"vector"`
            backends (FDSN, FIRMS, GDACS) return — the rows index
            bulky files rather than describe point/polygon features.
            The base `download` contract lists radar as this
            documented exception.

    Raises:
        ValueError: If `errors` is not a recognised policy.
    """
    self._errors = self.check_errors_policy(errors)
    self._show_progress = progress_bar
    products = self._search()
    pairs = self._fetch_pairs(products)
    return self._inventory(pairs)

earthlens.radar.catalog

Station registry for the NEXRAD radar backend.

Hosts :class:Catalog, the pydantic-backed reader for the bundled radar_data_catalog.yaml — a curated map of WSR-88D site ids ("KTLX") to name / latitude / longitude / state. The catalog gives each fetched volume a point geometry and lets a request select the radars inside a bounding box.

The catalog is informational: any valid four-letter site id can be fetched even if it is absent here (the volume just gets no geometry). The path to the bundled YAML lives at :data:CATALOG_PATH; monkey-patch it to redirect the loader in tests.

Catalog

Bases: AbstractCatalog

Catalog of NEXRAD WSR-88D sites.

Reads the bundled radar_data_catalog.yaml and exposes its stations: block as a typed dict[str, Station]. Instantiate with no arguments (Catalog()).

Examples:

• Look up a site and read its location:

  >>> from earthlens.radar import Catalog
  >>> ktlx = Catalog().get_station("KTLX")
  >>> (round(ktlx.latitude, 2), round(ktlx.longitude, 2))
  (35.33, -97.28)
  

Source code in libs/providers/atmosphere/src/earthlens/radar/catalog.py

class Catalog(AbstractCatalog):
    """Catalog of NEXRAD WSR-88D sites.

    Reads the bundled `radar_data_catalog.yaml` and exposes its `stations:` block
    as a typed `dict[str, Station]`. Instantiate with no arguments
    (`Catalog()`).

    Examples:
        - Look up a site and read its location:
            ```python
            >>> from earthlens.radar import Catalog
            >>> ktlx = Catalog().get_station("KTLX")
            >>> (round(ktlx.latitude, 2), round(ktlx.longitude, 2))
            (35.33, -97.28)

            ```
    """

    _catalog_kind: str = "NEXRAD station catalog"
    _entry_noun: str = "stations"

    datasets: dict[str, Station] = Field(default_factory=dict)

    @classmethod
    def _autoload(cls) -> dict[str, Any]:
        """Read the bundled station registry.

        Returns:
            dict[str, Any]: The `stations:` map keyed by site id.
        """
        return {"datasets": _load_stations(CATALOG_PATH)}

    def get_catalog(self) -> dict[str, Station]:
        """Return the structural per-site map (satisfies the base contract)."""
        return self.datasets

    def get_station(self, site_id: str) -> Station:
        """Resolve a site id to its :class:`Station` (did-you-mean on miss).

        Args:
            site_id: A four-letter WSR-88D id (e.g. `"KTLX"`).

        Returns:
            Station: The resolved site.

        Raises:
            ValueError: When `site_id` is unknown (with a did-you-mean
                hint from the base class).
        """
        return cast("Station", self.get_dataset(site_id))

    def in_bbox(
        self, west: float, south: float, east: float, north: float
    ) -> list[str]:
        """Return the site ids whose location falls inside a bbox.

        Args:
            west: West edge in degrees.
            south: South edge in degrees.
            east: East edge in degrees.
            north: North edge in degrees.

        Returns:
            list[str]: Matching site ids, sorted.

        Examples:
            - Find the catalogued sites over the south-central US:
                ```python
                >>> from earthlens.radar import Catalog
                >>> "KTLX" in Catalog().in_bbox(-100, 33, -95, 37)
                True

                ```
        """
        hits = [
            sid
            for sid, s in self.datasets.items()
            if west <= s.longitude <= east and south <= s.latitude <= north
        ]
        return sorted(hits)

get_catalog()

Return the structural per-site map (satisfies the base contract).

Source code in libs/providers/atmosphere/src/earthlens/radar/catalog.py

def get_catalog(self) -> dict[str, Station]:
    """Return the structural per-site map (satisfies the base contract)."""
    return self.datasets

get_station(site_id)

Resolve a site id to its :class:Station (did-you-mean on miss).

Parameters:

Name	Type	Description	Default
site_id	str	A four-letter WSR-88D id (e.g. "KTLX").	required

Returns:

Name	Type	Description
Station	Station	The resolved site.

Raises:

Type	Description
ValueError	When site_id is unknown (with a did-you-mean hint from the base class).

Source code in libs/providers/atmosphere/src/earthlens/radar/catalog.py

def get_station(self, site_id: str) -> Station:
    """Resolve a site id to its :class:`Station` (did-you-mean on miss).

    Args:
        site_id: A four-letter WSR-88D id (e.g. `"KTLX"`).

    Returns:
        Station: The resolved site.

    Raises:
        ValueError: When `site_id` is unknown (with a did-you-mean
            hint from the base class).
    """
    return cast("Station", self.get_dataset(site_id))

in_bbox(west, south, east, north)

Return the site ids whose location falls inside a bbox.

Parameters:

Name	Type	Description	Default
west	float	West edge in degrees.	required
south	float	South edge in degrees.	required
east	float	East edge in degrees.	required
north	float	North edge in degrees.	required

Returns:

Type	Description
list[str]	list[str]: Matching site ids, sorted.

Examples:

• Find the catalogued sites over the south-central US:

  >>> from earthlens.radar import Catalog
  >>> "KTLX" in Catalog().in_bbox(-100, 33, -95, 37)
  True
  

Source code in libs/providers/atmosphere/src/earthlens/radar/catalog.py

def in_bbox(
    self, west: float, south: float, east: float, north: float
) -> list[str]:
    """Return the site ids whose location falls inside a bbox.

    Args:
        west: West edge in degrees.
        south: South edge in degrees.
        east: East edge in degrees.
        north: North edge in degrees.

    Returns:
        list[str]: Matching site ids, sorted.

    Examples:
        - Find the catalogued sites over the south-central US:
            ```python
            >>> from earthlens.radar import Catalog
            >>> "KTLX" in Catalog().in_bbox(-100, 33, -95, 37)
            True

            ```
    """
    hits = [
        sid
        for sid, s in self.datasets.items()
        if west <= s.longitude <= east and south <= s.latitude <= north
    ]
    return sorted(hits)

Station

Bases: BaseModel

One WSR-88D radar site.

The site id (e.g. "KTLX") is the parent key in :attr:Catalog.datasets and is not stored on the row.

Attributes:

Name	Type	Description
name	str	Human-readable site name / location.
latitude	float	Site latitude in degrees (south negative).
longitude	float	Site longitude in degrees (west negative).
state	str	Two-letter US state / territory code.

Source code in libs/providers/atmosphere/src/earthlens/radar/catalog.py

class Station(BaseModel):
    """One WSR-88D radar site.

    The site id (e.g. `"KTLX"`) is the parent key in
    :attr:`Catalog.datasets` and is not stored on the row.

    Attributes:
        name: Human-readable site name / location.
        latitude: Site latitude in degrees (south negative).
        longitude: Site longitude in degrees (west negative).
        state: Two-letter US state / territory code.
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    name: str = ""
    latitude: float = Field(ge=-90.0, le=90.0)
    longitude: float = Field(ge=-180.0, le=180.0)
    state: str = ""

clear_catalog_cache()

Empty the module-level station-catalog parse cache.

Source code in libs/providers/atmosphere/src/earthlens/radar/catalog.py

def clear_catalog_cache() -> None:
    """Empty the module-level station-catalog parse cache."""
    _CATALOG_CACHE.clear()