FIRMS — API reference

NASA FIRMS active-fire data source subpackage — earthlens.firms. Background, usage, and credentials are covered under the other pages in this section; this page is the rendered API.

earthlens.firms

NASA FIRMS active-fire backend.

Thin wrapper over the NASA FIRMS (Fire Information for Resource Management System) area CSV API that returns near-real-time and archival active-fire detections from MODIS (C6.1) and VIIRS (S-NPP / NOAA-20 / NOAA-21) as a pyramids :class:~pyramids.feature.collection.FeatureCollection of fire-pixel points (CRS EPSG:4326).

This is a vector backend: the result is a table of geolocated fire detections, not a gridded array, so :data:FIRMS.OUTPUT_KIND is "vector" and the :class:earthlens.earthlens.EarthLens facade rejects an aggregate= argument for it.

FIRMS needs a free MAP_KEY (no SDK): the only dependencies are requests + pandas, both core, so there is no [firms] extra to install — the key lives in :class:FirmsAuth, not a dependency.

Sensor selection: for this backend variables is a list[str] of FIRMS sensor codes — variables=["VIIRS_SNPP_NRT"], variables=["MODIS_NRT", "VIIRS_SNPP_NRT"] — not data-variable names. This is an intentional, documented overload (the facade makes variables a required argument). The detection filters (min_confidence=, day_night=) arrive as explicit keyword arguments.

Public surface (re-exported from this package):

• :class:FIRMS — the backend; instantiate with a date range, a bbox, and variables=[sensor_code, ...], then call :meth:FIRMS.download.
• :class:Catalog — pydantic-backed loader for the bundled firms_data_catalog.yaml sensor dispatch table.
• :class:Sensor / :class:SensorColumn — one sensor's row and one of its CSV columns.
• :class:FirmsAuth / :class:FirmsCredentials — MAP_KEY resolution.
• :class:AuthenticationError — raised when no usable MAP_KEY resolves.
• :func:csv_to_fc / :func:empty_fc — the FIRMS CSV → FeatureCollection mapper and its empty-result counterpart.
• :data:CATALOG_PATH — path to the bundled sensor YAML; monkey-patchable in tests.

Examples:

• List the registered FIRMS sensor codes:

  >>> from earthlens.firms import Catalog
  >>> Catalog().codes()  # doctest: +NORMALIZE_WHITESPACE
  ['GOES_NRT', 'LANDSAT_NRT', 'MODIS_NRT', 'MODIS_SP',
   'VIIRS_NOAA20_NRT', 'VIIRS_NOAA20_SP', 'VIIRS_NOAA21_NRT',
   'VIIRS_SNPP_NRT', 'VIIRS_SNPP_SP']
  

AuthenticationError

Bases: AuthenticationError

Raised when no usable FIRMS MAP_KEY can be resolved.

Carries a message that names a fix: pass api_key= to EarthLens(...).authenticate(), set the FIRMS_MAP_KEY environment variable, or request a free key at firms.modaps.eosdis.nasa.gov/api/map_key/. A subclass of the cross-backend :class:earthlens.base.AuthenticationError so callers can catch every backend's auth failure with one except clause.

Source code in src/earthlens/firms/auth.py

class AuthenticationError(_BaseAuthenticationError):
    """Raised when no usable FIRMS `MAP_KEY` can be resolved.

    Carries a message that names a fix: pass `api_key=` to
    `EarthLens(...).authenticate()`, set the `FIRMS_MAP_KEY` environment
    variable, or request a free key at
    `firms.modaps.eosdis.nasa.gov/api/map_key/`. A subclass of the
    cross-backend :class:`earthlens.base.AuthenticationError` so callers
    can catch every backend's auth failure with one `except` clause.
    """

Catalog

Bases: AbstractCatalog

Sensor catalog for the NASA FIRMS backend.

Reads the bundled firms_data_catalog.yaml (shipped as package data) and exposes its sensors: block as a map of :class:Sensor rows, keyed by FIRMS source code under the inherited :attr:datasets field. Instantiate with no arguments (Catalog()); :func:model_post_init loads and validates the YAML in one pass. Resolve a sensor with :meth:get_sensor (a thin alias over :meth:~earthlens.base.AbstractCatalog.get_dataset) and a single column with :meth:get_column.

There is no available_* index — the listed sensors are the whole FIRMS universe (a deliberate deviation from the ECMWF/GEE catalogs, shared with GDACS/FDSN).

Attributes:

Name	Type	Description
datasets	dict[str, Sensor]	Map from the FIRMS source code to its :class:Sensor row.

Examples:

• List sensor codes and resolve one:

  >>> from earthlens.firms import Catalog
  >>> cat = Catalog()
  >>> cat.codes()  # doctest: +NORMALIZE_WHITESPACE
  ['GOES_NRT', 'LANDSAT_NRT', 'MODIS_NRT', 'MODIS_SP',
   'VIIRS_NOAA20_NRT', 'VIIRS_NOAA20_SP', 'VIIRS_NOAA21_NRT',
   'VIIRS_SNPP_NRT', 'VIIRS_SNPP_SP']
  >>> cat.get_sensor("MODIS_NRT").family
  'MODIS'
  >>> "MODIS_NRT" in cat
  True
  

• An unknown code raises with a did-you-mean hint:

  >>> from earthlens.firms import Catalog
  >>> Catalog().get_sensor("MODIS_NR")  # doctest: +ELLIPSIS
  Traceback (most recent call last):
      ...
  ValueError: 'MODIS_NR' is not in the FIRMS sensor catalog. Known sensors: [...]. Did you mean 'MODIS_NRT'?
  

Source code in src/earthlens/firms/catalog.py

class Catalog(AbstractCatalog):
    """Sensor catalog for the NASA FIRMS backend.

    Reads the bundled `firms_data_catalog.yaml` (shipped as package
    data) and exposes its `sensors:` block as a map of :class:`Sensor`
    rows, keyed by FIRMS source code under the inherited :attr:`datasets`
    field. Instantiate with no arguments (`Catalog()`);
    :func:`model_post_init` loads and validates the YAML in one pass.
    Resolve a sensor with :meth:`get_sensor` (a thin alias over
    :meth:`~earthlens.base.AbstractCatalog.get_dataset`) and a single
    column with :meth:`get_column`.

    There is no `available_*` index — the listed sensors are the whole
    FIRMS universe (a deliberate deviation from the ECMWF/GEE catalogs,
    shared with GDACS/FDSN).

    Attributes:
        datasets: Map from the FIRMS source code to its :class:`Sensor`
            row.

    Examples:
        - List sensor codes and resolve one:
            ```python
            >>> from earthlens.firms import Catalog
            >>> cat = Catalog()
            >>> cat.codes()  # doctest: +NORMALIZE_WHITESPACE
            ['GOES_NRT', 'LANDSAT_NRT', 'MODIS_NRT', 'MODIS_SP',
             'VIIRS_NOAA20_NRT', 'VIIRS_NOAA20_SP', 'VIIRS_NOAA21_NRT',
             'VIIRS_SNPP_NRT', 'VIIRS_SNPP_SP']
            >>> cat.get_sensor("MODIS_NRT").family
            'MODIS'
            >>> "MODIS_NRT" in cat
            True

            ```
        - An unknown code raises with a did-you-mean hint:
            ```python
            >>> from earthlens.firms import Catalog
            >>> Catalog().get_sensor("MODIS_NR")  # doctest: +ELLIPSIS
            Traceback (most recent call last):
                ...
            ValueError: 'MODIS_NR' is not in the FIRMS sensor catalog. Known sensors: [...]. Did you mean 'MODIS_NRT'?

            ```
    """

    _catalog_kind: str = "FIRMS sensor catalog"
    _entry_noun: str = "sensors"

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

    def model_post_init(self, __context: Any) -> None:
        """Auto-load the bundled catalog when no sensors were supplied.

        `Catalog()` with no args reads :data:`CATALOG_PATH`; passing
        `datasets=...` skips the disk read (used in tests).

        Raises:
            ValueError: Propagated from :meth:`load` when the YAML is
                missing, empty, or has a malformed sensor row.
        """
        if not self.datasets:
            loaded = Catalog.load()
            self.datasets = loaded.datasets
        super().model_post_init(__context)

    @classmethod
    def load(cls, catalog_path: Path | None = None) -> Catalog:
        """Read the FIRMS sensor catalog from disk.

        Args:
            catalog_path: Path to the catalog YAML. Defaults to the
                module-level :data:`CATALOG_PATH`.

        Returns:
            A fully-populated :class:`Catalog`.

        Raises:
            ValueError: If the file has no `sensors:` block, or a row
                fails :class:`Sensor` validation.
        """
        catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
        resolved = str(catalog_path.resolve())
        try:
            mtime = catalog_path.stat().st_mtime_ns
        except FileNotFoundError:
            mtime = 0
        key = (resolved, mtime)
        cached = _CATALOG_CACHE.get(key)
        if cached is not None:
            return cls(datasets=dict(cached))
        data = load_yaml_strict(catalog_path) or {}
        sensors_yaml = data.get("sensors") or {}
        if not sensors_yaml:
            raise ValueError(
                f"{catalog_path} is missing or has an empty 'sensors:' block. "
                "The FIRMS catalog must list at least one sensor."
            )
        sensors: dict[str, Sensor] = {}
        for code, body in sensors_yaml.items():
            try:
                sensors[code] = Sensor(**dict(body or {}))
            except ValidationError as exc:
                raise ValueError(
                    f"{catalog_path} sensor {code!r} failed validation:\n{exc}"
                ) from exc
        _CATALOG_CACHE[key] = sensors
        return cls(datasets=dict(sensors))

    def get_catalog(self) -> dict[str, Sensor]:
        """Return the sensor map (satisfies the abstract contract).

        Returns:
            dict[str, Sensor]: Same object as :attr:`datasets`.
        """
        return self.datasets

    def get_sensor(self, code: str) -> Sensor:
        """Return the :class:`Sensor` for `code`, with a did-you-mean hint.

        Thin alias over
        :meth:`~earthlens.base.AbstractCatalog.get_dataset`.

        Args:
            code: A FIRMS source code (`"VIIRS_SNPP_NRT"`, `"MODIS_NRT"`,
                …).

        Returns:
            Sensor: The matching sensor row.

        Raises:
            ValueError: If `code` is not a registered FIRMS sensor.
        """
        return self.get_dataset(code)

    def get_column(self, code: str, column: str) -> SensorColumn:
        """Return one column's metadata for a `(sensor, column)` pair.

        Args:
            code: A FIRMS source code as it appears in :attr:`datasets`.
            column: A CSV column name declared under that sensor.

        Returns:
            SensorColumn: The matching column metadata.

        Raises:
            ValueError: If `code` is not a registered sensor.
            KeyError: If `column` is not declared under that sensor.

        Examples:
            - Read a column's units:
                ```python
                >>> from earthlens.firms import Catalog
                >>> Catalog().get_column("MODIS_NRT", "confidence").units
                '%'

                ```
        """
        return self.get_sensor(code).columns[column]

    def get_variable(self, code: str, column: str) -> SensorColumn:
        """Leaf accessor for the shared two-arg get_variable contract.

        Alias of :meth:`get_column` so the FIRMS leaf is reachable under
        the same `get_variable(dataset_key, variable_name)` verb the
        other two-level catalogs use.

        Args:
            code: A FIRMS source code.
            column: A CSV column name declared under that sensor.

        Returns:
            SensorColumn: The matching column metadata.
        """
        return self.get_column(code, column)

    def codes(self) -> list[str]:
        """Return the registered FIRMS sensor codes, sorted.

        Returns:
            list[str]: The sensor codes (`["MODIS_NRT", "MODIS_SP", ...]`).
        """
        return sorted(self.datasets)

codes()

Return the registered FIRMS sensor codes, sorted.

Returns:

Type	Description
list[str]	list[str]: The sensor codes (["MODIS_NRT", "MODIS_SP", ...]).

Source code in src/earthlens/firms/catalog.py

def codes(self) -> list[str]:
    """Return the registered FIRMS sensor codes, sorted.

    Returns:
        list[str]: The sensor codes (`["MODIS_NRT", "MODIS_SP", ...]`).
    """
    return sorted(self.datasets)

get_catalog()

Return the sensor map (satisfies the abstract contract).

Returns:

Type	Description
dict[str, Sensor]	dict[str, Sensor]: Same object as :attr:datasets.

Source code in src/earthlens/firms/catalog.py

def get_catalog(self) -> dict[str, Sensor]:
    """Return the sensor map (satisfies the abstract contract).

    Returns:
        dict[str, Sensor]: Same object as :attr:`datasets`.
    """
    return self.datasets

get_column(code, column)

Return one column's metadata for a (sensor, column) pair.

Parameters:

Name	Type	Description	Default
code	str	A FIRMS source code as it appears in :attr:datasets.	required
column	str	A CSV column name declared under that sensor.	required

Returns:

Name	Type	Description
SensorColumn	SensorColumn	The matching column metadata.

Raises:

Type	Description
ValueError	If code is not a registered sensor.
KeyError	If column is not declared under that sensor.

Examples:

• Read a column's units:

  >>> from earthlens.firms import Catalog
  >>> Catalog().get_column("MODIS_NRT", "confidence").units
  '%'
  

Source code in src/earthlens/firms/catalog.py

def get_column(self, code: str, column: str) -> SensorColumn:
    """Return one column's metadata for a `(sensor, column)` pair.

    Args:
        code: A FIRMS source code as it appears in :attr:`datasets`.
        column: A CSV column name declared under that sensor.

    Returns:
        SensorColumn: The matching column metadata.

    Raises:
        ValueError: If `code` is not a registered sensor.
        KeyError: If `column` is not declared under that sensor.

    Examples:
        - Read a column's units:
            ```python
            >>> from earthlens.firms import Catalog
            >>> Catalog().get_column("MODIS_NRT", "confidence").units
            '%'

            ```
    """
    return self.get_sensor(code).columns[column]

get_sensor(code)

Return the :class:Sensor for code, with a did-you-mean hint.

Thin alias over :meth:~earthlens.base.AbstractCatalog.get_dataset.

Parameters:

Name	Type	Description	Default
code	str	A FIRMS source code ("VIIRS_SNPP_NRT", "MODIS_NRT", …).	required

Returns:

Name	Type	Description
Sensor	Sensor	The matching sensor row.

Raises:

Type	Description
ValueError	If code is not a registered FIRMS sensor.

Source code in src/earthlens/firms/catalog.py

def get_sensor(self, code: str) -> Sensor:
    """Return the :class:`Sensor` for `code`, with a did-you-mean hint.

    Thin alias over
    :meth:`~earthlens.base.AbstractCatalog.get_dataset`.

    Args:
        code: A FIRMS source code (`"VIIRS_SNPP_NRT"`, `"MODIS_NRT"`,
            …).

    Returns:
        Sensor: The matching sensor row.

    Raises:
        ValueError: If `code` is not a registered FIRMS sensor.
    """
    return self.get_dataset(code)

get_variable(code, column)

Leaf accessor for the shared two-arg get_variable contract.

Alias of :meth:get_column so the FIRMS leaf is reachable under the same get_variable(dataset_key, variable_name) verb the other two-level catalogs use.

Parameters:

Name	Type	Description	Default
code	str	A FIRMS source code.	required
column	str	A CSV column name declared under that sensor.	required

Returns:

Name	Type	Description
SensorColumn	SensorColumn	The matching column metadata.

Source code in src/earthlens/firms/catalog.py

def get_variable(self, code: str, column: str) -> SensorColumn:
    """Leaf accessor for the shared two-arg get_variable contract.

    Alias of :meth:`get_column` so the FIRMS leaf is reachable under
    the same `get_variable(dataset_key, variable_name)` verb the
    other two-level catalogs use.

    Args:
        code: A FIRMS source code.
        column: A CSV column name declared under that sensor.

    Returns:
        SensorColumn: The matching column metadata.
    """
    return self.get_column(code, column)

load(catalog_path=None) classmethod

Read the FIRMS sensor catalog from disk.

Parameters:

Name	Type	Description	Default
catalog_path	Path | None	Path to the catalog YAML. Defaults to the module-level :data:CATALOG_PATH.	None

Returns:

Type	Description
Catalog	A fully-populated :class:Catalog.

Raises:

Type	Description
ValueError	If the file has no sensors: block, or a row fails :class:Sensor validation.

Source code in src/earthlens/firms/catalog.py

@classmethod
def load(cls, catalog_path: Path | None = None) -> Catalog:
    """Read the FIRMS sensor catalog from disk.

    Args:
        catalog_path: Path to the catalog YAML. Defaults to the
            module-level :data:`CATALOG_PATH`.

    Returns:
        A fully-populated :class:`Catalog`.

    Raises:
        ValueError: If the file has no `sensors:` block, or a row
            fails :class:`Sensor` validation.
    """
    catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
    resolved = str(catalog_path.resolve())
    try:
        mtime = catalog_path.stat().st_mtime_ns
    except FileNotFoundError:
        mtime = 0
    key = (resolved, mtime)
    cached = _CATALOG_CACHE.get(key)
    if cached is not None:
        return cls(datasets=dict(cached))
    data = load_yaml_strict(catalog_path) or {}
    sensors_yaml = data.get("sensors") or {}
    if not sensors_yaml:
        raise ValueError(
            f"{catalog_path} is missing or has an empty 'sensors:' block. "
            "The FIRMS catalog must list at least one sensor."
        )
    sensors: dict[str, Sensor] = {}
    for code, body in sensors_yaml.items():
        try:
            sensors[code] = Sensor(**dict(body or {}))
        except ValidationError as exc:
            raise ValueError(
                f"{catalog_path} sensor {code!r} failed validation:\n{exc}"
            ) from exc
    _CATALOG_CACHE[key] = sensors
    return cls(datasets=dict(sensors))

model_post_init(__context)

Auto-load the bundled catalog when no sensors were supplied.

Catalog() with no args reads :data:CATALOG_PATH; passing datasets=... skips the disk read (used in tests).

Raises:

Type	Description
ValueError	Propagated from :meth:load when the YAML is missing, empty, or has a malformed sensor row.

Source code in src/earthlens/firms/catalog.py

def model_post_init(self, __context: Any) -> None:
    """Auto-load the bundled catalog when no sensors were supplied.

    `Catalog()` with no args reads :data:`CATALOG_PATH`; passing
    `datasets=...` skips the disk read (used in tests).

    Raises:
        ValueError: Propagated from :meth:`load` when the YAML is
            missing, empty, or has a malformed sensor row.
    """
    if not self.datasets:
        loaded = Catalog.load()
        self.datasets = loaded.datasets
    super().model_post_init(__context)

FIRMS

Bases: AbstractDataSource

NASA FIRMS active-fire backend (vector point-feature output).

Wraps the FIRMS area CSV API so a user can pull a space/time/sensor window of fire detections through the same download() shape every other earthlens backend uses. Windows longer than the FIRMS 5-day per-request cap are chunked, and each (sensor, ≤5-day chunk) is one CSV GET; the rows are mapped to a :class:~pyramids.feature.collection.FeatureCollection.

FIRMS needs a free MAP_KEY. Supply it to :meth:authenticate as api_key=, or set the FIRMS_MAP_KEY environment variable and let authenticate() / download() resolve it. Credentials are not a constructor argument — the constructor describes only what to fetch.

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	"vector" — the result is a table of detection features, so the facade rejects aggregate= with NotImplementedError.

Source code in src/earthlens/firms/backend.py

class FIRMS(AbstractDataSource):
    """NASA FIRMS active-fire backend (vector point-feature output).

    Wraps the FIRMS area CSV API so a user can pull a space/time/sensor
    window of fire detections through the same `download()` shape every
    other earthlens backend uses. Windows longer than the FIRMS 5-day
    per-request cap are chunked, and each `(sensor, ≤5-day chunk)` is
    one CSV GET; the rows are mapped to a
    :class:`~pyramids.feature.collection.FeatureCollection`.

    FIRMS needs a free `MAP_KEY`. Supply it to :meth:`authenticate` as
    `api_key=`, or set the `FIRMS_MAP_KEY` environment variable and let
    `authenticate()` / `download()` resolve it. Credentials are not a
    constructor argument — the constructor describes only what to fetch.

    Attributes:
        OUTPUT_KIND: `"vector"` — the result is a table of detection
            features, so the facade rejects `aggregate=` with
            `NotImplementedError`.
    """

    OUTPUT_KIND: OutputKind = "vector"

    def __init__(
        self,
        start: str,
        end: str,
        variables: list[str],
        lat_lim: list[float],
        lon_lim: list[float],
        temporal_resolution: str = "all",
        path: Path | str = "",
        fmt: str = "%Y-%m-%d",
        min_confidence: float | None = None,
        day_night: str | None = None,
        file_format: FileFormat = "gpkg",
        timeout: float = 60.0,
    ):
        """Initialise a FIRMS backend instance.

        Args:
            start: Inclusive start of the detection window, as a string
                parsed with `fmt`.
            end: Inclusive end of the detection window.
            variables: List of FIRMS sensor codes to query
                (`["VIIRS_SNPP_NRT"]`, `["MODIS_NRT", "VIIRS_SNPP_NRT"]`).
                For this backend `variables` names the *sensors*, not
                data variables (see the package docstring). An empty list
                defaults to `["VIIRS_SNPP_NRT"]`.
            lat_lim: `[lat_min, lat_max]` bounding-box latitudes in
                degrees, both in `[-90, 90]`.
            lon_lim: `[lon_min, lon_max]` bounding-box longitudes in
                degrees, both in `[-180, 180]`.
            temporal_resolution: FIRMS chunks by ≤5-day windows
                internally, not by a daily/monthly cadence, so this is
                the sentinel `"all"`, not a pandas frequency alias.
            path: Output directory for the written vector file. Created
                by the parent class if absent.
            fmt: `strptime` format for `start` / `end`.
            min_confidence: Optional 0-100 lower bound applied
                client-side on the normalised `confidence_pct` column
                (FIRMS has no server-side confidence filter). `None`
                keeps every detection.
            day_night: Optional `"D"` / `"N"` filter applied client-side
                on the `daynight` column. `None` keeps both.
            file_format: Output vector format — `"gpkg"` (default,
                GeoPackage) or `"geojson"`.
            timeout: Per-request timeout in seconds for each CSV GET.

        Raises:
            ValueError: If `file_format` is not `"gpkg"` / `"geojson"`.
            TypeError: If `variables` is a mapping rather than a list of
                sensor codes.
        """
        if file_format not in _DRIVERS:
            raise ValueError(
                f"file_format must be one of {sorted(_DRIVERS)}, got "
                f"{file_format!r}."
            )
        if isinstance(variables, dict):
            raise TypeError(
                "FIRMS `variables` must be a list of sensor codes (e.g. "
                "['VIIRS_SNPP_NRT', 'MODIS_NRT']), not a mapping. For this "
                "backend `variables` selects sensors, not data variables; the "
                "detection filters are the explicit min_confidence= / "
                "day_night= keyword arguments."
            )
        self._min_confidence = min_confidence
        self._day_night = day_night
        self._file_format: FileFormat = file_format
        self._timeout = timeout
        self._catalog = Catalog()
        # Reactive back-off knobs (G2); the sleep is an instance attr so
        # it can be swapped for a no-op in tests.
        self._sleep = time.sleep
        self._max_retries = 5
        self._backoff_factor = 1.0
        super().__init__(
            start=start,
            end=end,
            variables=list(variables) or list(_DEFAULT_SENSORS),
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim,
            lon_lim=lon_lim,
            fmt=fmt,
            path=path,
        )

    def _initialize(self) -> FirmsAuth:
        """Build the (unconfigured) :class:`FirmsAuth` holder.

        No credentials are resolved here — the constructor describes only
        what to fetch. The `MAP_KEY` is resolved later by
        :meth:`authenticate` (explicitly via `api_key=`, or from the
        `FIRMS_MAP_KEY` environment variable), which `download()` also
        triggers lazily if it has not run.

        Returns:
            FirmsAuth: An unconfigured auth; `is_authenticated()` is
                `False` until :meth:`authenticate` resolves a key.
        """
        return FirmsAuth(FirmsCredentials(api_key=None))

    def authenticate(self, api_key: str | None = None) -> FIRMS:
        """Resolve the FIRMS `MAP_KEY` and arm the backend for download.

        The explicit, fail-fast credential step. Pass `api_key=` to use a
        key directly; omit it (or pass `None`) to read the `FIRMS_MAP_KEY`
        environment variable. Either way the resolved key is held for the
        subsequent :meth:`download`. Calling it again with a different
        `api_key` re-arms with the new key. `download()` calls this with
        no argument on your behalf if you never do, so an explicit call is
        only needed to pass a key directly or to validate up front.

        Args:
            api_key: The FIRMS `MAP_KEY` to use. When `None`, the
                `FIRMS_MAP_KEY` environment variable is read instead.

        Returns:
            The backend instance, so it chains
            `EarthLens(...).authenticate(api_key=...).download()`.

        Raises:
            AuthenticationError: If `api_key` is `None` and no
                `FIRMS_MAP_KEY` environment variable is set.

        Examples:
            - Arm the backend with an explicit key and read it back:
                ```python
                >>> import tempfile
                >>> from earthlens.firms import FIRMS
                >>> backend = FIRMS(
                ...     start="2024-08-01", end="2024-08-01",
                ...     variables=["VIIRS_SNPP_NRT"],
                ...     lat_lim=[33.0, 35.0], lon_lim=[-119.0, -117.0],
                ...     path=tempfile.mkdtemp(),
                ... )
                >>> backend.authenticate(api_key="demo-key").client.api_key
                'demo-key'

                ```
            - A fresh backend is unauthenticated until the key resolves:
                ```python
                >>> import tempfile
                >>> from earthlens.firms import FIRMS
                >>> backend = FIRMS(
                ...     start="2024-08-01", end="2024-08-01",
                ...     variables=["VIIRS_SNPP_NRT"],
                ...     lat_lim=[33.0, 35.0], lon_lim=[-119.0, -117.0],
                ...     path=tempfile.mkdtemp(),
                ... )
                >>> backend.client.is_authenticated()
                False
                >>> backend.authenticate(api_key="abc123").client.is_authenticated()
                True

                ```
        """
        auth = FirmsAuth(FirmsCredentials(api_key=api_key))
        auth.configure()
        self.client = auth
        return self

    def _create_grid(self, lat_lim: list, lon_lim: list) -> SpatialExtent:
        """Wrap the WGS84 bbox into a :class:`SpatialExtent` (no snapping).

        FIRMS clips server-side to the bbox path segment, so the box
        passes through unchanged.

        Args:
            lat_lim: `[lat_min, lat_max]` in degrees.
            lon_lim: `[lon_min, lon_max]` in degrees.

        Returns:
            SpatialExtent: Validated, frozen bbox.
        """
        return SpatialExtent.from_pairs(lat_lim=lat_lim, lon_lim=lon_lim)

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

        FIRMS chunks the window into ≤5-day requests internally (see
        :meth:`_search`), so the resolution is kept as the sentinel
        `"all"` (not a real pandas frequency alias) and `dates`
        collapses to the two endpoints.

        Args:
            start: Inclusive start date string.
            end: Inclusive end date string.
            temporal_resolution: Recorded as the resolution label;
                FIRMS always chunks the full window.
            fmt: `strptime` format applied to `start` and `end`.

        Returns:
            TemporalExtent: Frozen model with the parsed endpoints.

        Raises:
            ValueError: If `start` parses to a date later than `end`.
        """
        start_dt = dt.datetime.strptime(start, fmt)
        end_dt = dt.datetime.strptime(end, fmt)
        return TemporalExtent(
            start_date=start_dt,
            end_date=end_dt,
            resolution="all",
            dates=pd.DatetimeIndex([start_dt, end_dt]),
        )

    def _search(self) -> list[RemoteProduct]:
        """List one :class:`RemoteProduct` per `(sensor, ≤5-day chunk)`.

        Validates each code in `self.vars` against the bundled catalog
        (raising with a did-you-mean hint on an unknown sensor), warns
        when the requested window falls outside an `*_NRT` sensor's
        coverage (naming the `*_SP` archive variant — it does *not*
        auto-swap), and walks the `[start, end]` window in ≤5-day
        chunks. No network call is made here.

        Returns:
            list[RemoteProduct]: One product per `(sensor, chunk)`, whose
                `metadata` carries `sensor`, `family`, `start_date`, and
                `day_range`. The product `id` is `f"{sensor}:{start}"`.

        Raises:
            ValueError: If a code in `self.vars` is not a registered
                FIRMS sensor.
        """
        start_date = self.time.start_date.date()
        end_date = self.time.end_date.date()
        windows = chunk_windows(start_date, end_date)
        total_gets = len(self.vars) * len(windows)
        logger.info(
            f"FIRMS request: {len(self.vars)} sensor(s) x {len(windows)} chunk(s) "
            f"= {total_gets} CSV GET(s)"
        )
        if total_gets > FANOUT_WARN_THRESHOLD:
            logger.warning(
                f"FIRMS request fans out to {total_gets} CSV GET(s) (one "
                f"transaction each); FIRMS allows ~5000 per rolling 10 minutes. "
                "The per-request back-off will pace this, but consider narrowing "
                "the window or sensor list for a large pull."
            )
        products: list[RemoteProduct] = []
        non_percent: list[str] = []
        for code in self.vars:
            sensor = self._catalog.get_sensor(code)
            self._warn_if_out_of_coverage(sensor, start_date, end_date)
            if (
                self._min_confidence is not None
                and sensor.family not in events.PERCENT_CONFIDENCE_FAMILIES
            ):
                non_percent.append(code)
            for chunk_start, day_range in windows:
                products.append(
                    RemoteProduct(
                        id=f"{code}:{chunk_start.isoformat()}",
                        metadata={
                            "sensor": code,
                            "family": sensor.family,
                            "start_date": chunk_start,
                            "day_range": day_range,
                        },
                    )
                )
        if non_percent:
            logger.warning(
                f"min_confidence={self._min_confidence} is not applied to "
                f"{non_percent}: their confidence is a provider-scale (non "
                "0-100) value, so thresholding would drop every detection; "
                "those sensors' detections are kept unfiltered."
            )
        return products

    def _warn_if_out_of_coverage(
        self, sensor, start_date: dt.date, end_date: dt.date
    ) -> None:
        """Warn (do not auto-swap) when the window is outside coverage.

        An `*_NRT` sensor holds only roughly the last
        :data:`NRT_RETENTION_DAYS` days; a request for older data returns
        a silently empty CSV rather than an error. This logs a loud
        warning naming the `*_SP` archive variant when that variant
        exists. A request that predates the sensor's mission start is
        warned the same way.

        Args:
            sensor: The resolved :class:`~earthlens.firms.Sensor`.
            start_date: Requested inclusive start.
            end_date: Requested inclusive end.
        """
        # `temporal.start` / `temporal.end` are typed `datetime.date | None`
        # on the catalog model, so no datetime-narrowing is needed here.
        mission_start = sensor.temporal.start
        if mission_start is not None and start_date < mission_start:
            logger.warning(
                f"{sensor.code} coverage begins {mission_start}; the requested "
                f"window starts {start_date} and may return no detections."
            )
        coverage_end = sensor.temporal.end
        if coverage_end is not None and end_date > coverage_end:
            logger.warning(
                f"{sensor.code} coverage ends {coverage_end}; the requested "
                f"window ends {end_date} and may return no detections past the "
                "coverage end."
            )
        # The NRT-retention heuristic below is advisory (retention drifts
        # per sensor); it only applies to NRT sensors.
        if sensor.temporal.quality != "NRT":
            return
        cutoff = dt.date.today() - dt.timedelta(days=NRT_RETENTION_DAYS)
        if end_date < cutoff:
            sp_variant = sensor.code.replace("_NRT", "_SP")
            hint = (
                f" for archive data use {sp_variant}"
                if sp_variant in self._catalog
                else ""
            )
            logger.warning(
                f"{sensor.code} is near-real-time and covers only the last "
                f"~{NRT_RETENTION_DAYS} days; the requested window ending "
                f"{end_date} is older and will likely be empty{hint}."
            )

    def _fetch(self, products: list[RemoteProduct]) -> list[FeatureCollection]:
        """Fetch each product's CSV and map it to a FeatureCollection.

        Widens the inherited `-> list[Path]` contract: a vector backend
        returns in-memory :class:`FeatureCollection`s, not file paths.
        Each product is one CSV GET issued through the quota back-off
        (`G2`); the response body is classified before parsing (`G6`) so
        a FIRMS error-as-HTTP-200 text body never reaches
        `pandas.read_csv`.

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            list[FeatureCollection]: One collection per product, in the
                same order.
        """
        return [self._fetch_one(product) for product in products]

    def _fetch_one(self, product: RemoteProduct) -> FeatureCollection:
        """Fetch and map one `(sensor, chunk)` product.

        Args:
            product: One :class:`RemoteProduct` from :meth:`_search`.

        Returns:
            FeatureCollection: The chunk's detections (schema-only empty
                when the CSV had no rows).

        Raises:
            AuthenticationError: If the body is a bad-key message (`G6`).
            RuntimeError: If the body is a non-CSV error, or a quota body
                survives the back-off retries.
            requests.HTTPError: On a non-quota HTTP error status.
        """
        url = self._build_url(product)
        response = firms_get(
            url,
            timeout=self._timeout,
            get=requests.get,
            sleep=self._sleep,
            max_retries=self._max_retries,
            backoff_factor=self._backoff_factor,
        )
        status = getattr(response, "status_code", 200)
        if status >= 400:
            # Do NOT call response.raise_for_status(): its message embeds
            # the request URL, which carries the MAP_KEY as a path segment
            # and would leak the secret into logs/tracebacks. Raise a
            # redacted HTTPError instead.
            raise requests.HTTPError(
                f"FIRMS area request for sensor {product.metadata['sensor']} "
                f"failed with HTTP {status} (URL omitted to avoid leaking the "
                "MAP_KEY)."
            )
        text = response.text
        kind = classify_body(text)
        if kind == "auth":
            raise AuthenticationError(f"FIRMS rejected the MAP_KEY: {_truncate(text)}")
        if kind == "quota":
            raise RuntimeError(
                "FIRMS transaction quota exhausted after back-off retries: "
                f"{_truncate(text)}"
            )
        if kind == "error":
            raise RuntimeError(
                f"FIRMS returned a non-CSV error body: {_truncate(text)}"
            )
        frame = pd.read_csv(StringIO(text))
        return events.csv_to_fc(
            frame,
            sensor=product.metadata["sensor"],
            family=product.metadata["family"],
            min_confidence=self._min_confidence,
            day_night=self._day_night,
        )

    def _build_url(self, product: RemoteProduct) -> str:
        """Compose the FIRMS area-CSV URL for one product.

        The bbox path segment is `W,S,E,N` (FIRMS area order). The
        `MAP_KEY` is read from the configured :class:`FirmsAuth`.

        Args:
            product: One :class:`RemoteProduct` from :meth:`_search`.

        Returns:
            str: The fully-formed request URL.
        """
        bbox = (
            f"{self.space.west},{self.space.south},{self.space.east},{self.space.north}"
        )
        return AREA_URL_TEMPLATE.format(
            map_key=self.client.api_key,
            sensor=product.metadata["sensor"],
            bbox=bbox,
            day_range=product.metadata["day_range"],
            start_date=product.metadata["start_date"].isoformat(),
        )

    def _api(self) -> list[FeatureCollection]:
        """Compose `_search` and `_fetch` into the canonical C3 shape."""
        return self._api_via_search_fetch()

    def _api_via_search_fetch_with_progress(
        self, progress_bar: bool
    ) -> list[FeatureCollection]:
        """C3 composition with a per-chunk progress bar.

        Mirrors the CMEMS / OpenAQ progress-aware composition: run the
        cheap :meth:`_search`, then map :meth:`_fetch_one` over the
        products wrapped in a `tqdm` bar (disabled when `progress_bar`
        is `False`). Short-circuits on an empty search.

        Args:
            progress_bar: Show the per-chunk `tqdm` bar when `True`.

        Returns:
            list[FeatureCollection]: One collection per product, or `[]`
                when nothing matched.
        """
        return self._search_fetch_each(
            progress_bar=progress_bar, desc="FIRMS chunks", unit="chunk"
        )

    def download(
        self,
        progress_bar: bool = True,
        aggregate: AggregationConfig | None = None,
    ) -> FeatureCollection:
        """Query FIRMS and return the matched detections.

        Runs the cheap :meth:`_search` (sensor validation + chunk
        planning) then the throttled :meth:`_fetch` (one CSV GET per
        chunk), concatenates the per-chunk collections into one
        FeatureCollection, writes it to one vector file under `path`, and
        returns it. An empty result returns — and writes nothing for — a
        schema-correct empty FeatureCollection.

        Args:
            progress_bar: Show a per-chunk progress bar. Defaults to
                `True`.
            aggregate: Must be `None`. Detections are vector, not
                gridded, so there is no meaningful aggregation. The
                facade already rejects a non-`None` `aggregate=` for a
                `vector` backend; this is the belt-and-suspenders guard
                for direct backend callers.

        Returns:
            FeatureCollection: The matched detections, CRS `EPSG:4326`.
                Empty (schema-only) when nothing matched.

        Raises:
            NotImplementedError: If `aggregate` is not `None`.
        """
        if aggregate is not None:
            raise NotImplementedError(
                "FIRMS.download(aggregate=...) is not supported: fire "
                "detections are vector point features, not gridded rasters, so "
                "there is no meaningful gridded reduction. Call download() "
                "without aggregate= and post-process the returned "
                "FeatureCollection (a GeoDataFrame) directly."
            )

        # Resolve the MAP_KEY from FIRMS_MAP_KEY if authenticate() was not
        # called explicitly, so EarthLens(...).download() still works when
        # the key lives in the environment.
        if not self.client.is_authenticated():
            self.authenticate()

        collections = self._api_via_search_fetch_with_progress(progress_bar)
        collection = events.concat(collections)

        if len(collection):
            out_path = self._write(collection)
            logger.info(
                f"FIRMS download summary: {len(collection)} detection(s) "
                f"written to {out_path}"
            )
        else:
            logger.warning(
                "FIRMS download summary: no detections matched the request, "
                "nothing written"
            )
        return collection

    def _write(self, collection: FeatureCollection) -> Path:
        """Write the detections to one vector file under `root_dir`.

        The filename embeds the sensor list and the query's date window
        so successive downloads into the same `path` yield distinct
        files. Two downloads of the same request overwrite, the intended
        idempotent behaviour.

        Args:
            collection: The detections to write.

        Returns:
            Path: Absolute path of the file written.
        """
        driver, ext = _DRIVERS[self._file_format]
        sensors = "-".join(self.vars)
        stem = (
            f"firms_{sensors}_{self.time.start_date:%Y%m%d}"
            f"_{self.time.end_date:%Y%m%d}"
        )
        out_path = self.root_dir / f"{stem}.{ext}"
        collection.to_file(str(out_path), driver=driver)
        return out_path

__init__(start, end, variables, lat_lim, lon_lim, temporal_resolution='all', path='', fmt='%Y-%m-%d', min_confidence=None, day_night=None, file_format='gpkg', timeout=60.0)

Initialise a FIRMS backend instance.

Parameters:

Name	Type	Description	Default
start	str	Inclusive start of the detection window, as a string parsed with fmt.	required
end	str	Inclusive end of the detection window.	required
variables	list[str]	List of FIRMS sensor codes to query (["VIIRS_SNPP_NRT"], ["MODIS_NRT", "VIIRS_SNPP_NRT"]). For this backend variables names the sensors, not data variables (see the package docstring). An empty list defaults to ["VIIRS_SNPP_NRT"].	required
lat_lim	list[float]	[lat_min, lat_max] bounding-box latitudes in degrees, both in [-90, 90].	required
lon_lim	list[float]	[lon_min, lon_max] bounding-box longitudes in degrees, both in [-180, 180].	required
temporal_resolution	str	FIRMS chunks by ≤5-day windows internally, not by a daily/monthly cadence, so this is the sentinel "all", not a pandas frequency alias.	'all'
path	Path | str	Output directory for the written vector file. Created by the parent class if absent.	''
fmt	str	strptime format for start / end.	'%Y-%m-%d'
min_confidence	float | None	Optional 0-100 lower bound applied client-side on the normalised confidence_pct column (FIRMS has no server-side confidence filter). None keeps every detection.	None
day_night	str | None	Optional "D" / "N" filter applied client-side on the daynight column. None keeps both.	None
file_format	FileFormat	Output vector format — "gpkg" (default, GeoPackage) or "geojson".	'gpkg'
timeout	float	Per-request timeout in seconds for each CSV GET.	60.0

Raises:

Type	Description
ValueError	If file_format is not "gpkg" / "geojson".
TypeError	If variables is a mapping rather than a list of sensor codes.

Source code in src/earthlens/firms/backend.py

def __init__(
    self,
    start: str,
    end: str,
    variables: list[str],
    lat_lim: list[float],
    lon_lim: list[float],
    temporal_resolution: str = "all",
    path: Path | str = "",
    fmt: str = "%Y-%m-%d",
    min_confidence: float | None = None,
    day_night: str | None = None,
    file_format: FileFormat = "gpkg",
    timeout: float = 60.0,
):
    """Initialise a FIRMS backend instance.

    Args:
        start: Inclusive start of the detection window, as a string
            parsed with `fmt`.
        end: Inclusive end of the detection window.
        variables: List of FIRMS sensor codes to query
            (`["VIIRS_SNPP_NRT"]`, `["MODIS_NRT", "VIIRS_SNPP_NRT"]`).
            For this backend `variables` names the *sensors*, not
            data variables (see the package docstring). An empty list
            defaults to `["VIIRS_SNPP_NRT"]`.
        lat_lim: `[lat_min, lat_max]` bounding-box latitudes in
            degrees, both in `[-90, 90]`.
        lon_lim: `[lon_min, lon_max]` bounding-box longitudes in
            degrees, both in `[-180, 180]`.
        temporal_resolution: FIRMS chunks by ≤5-day windows
            internally, not by a daily/monthly cadence, so this is
            the sentinel `"all"`, not a pandas frequency alias.
        path: Output directory for the written vector file. Created
            by the parent class if absent.
        fmt: `strptime` format for `start` / `end`.
        min_confidence: Optional 0-100 lower bound applied
            client-side on the normalised `confidence_pct` column
            (FIRMS has no server-side confidence filter). `None`
            keeps every detection.
        day_night: Optional `"D"` / `"N"` filter applied client-side
            on the `daynight` column. `None` keeps both.
        file_format: Output vector format — `"gpkg"` (default,
            GeoPackage) or `"geojson"`.
        timeout: Per-request timeout in seconds for each CSV GET.

    Raises:
        ValueError: If `file_format` is not `"gpkg"` / `"geojson"`.
        TypeError: If `variables` is a mapping rather than a list of
            sensor codes.
    """
    if file_format not in _DRIVERS:
        raise ValueError(
            f"file_format must be one of {sorted(_DRIVERS)}, got "
            f"{file_format!r}."
        )
    if isinstance(variables, dict):
        raise TypeError(
            "FIRMS `variables` must be a list of sensor codes (e.g. "
            "['VIIRS_SNPP_NRT', 'MODIS_NRT']), not a mapping. For this "
            "backend `variables` selects sensors, not data variables; the "
            "detection filters are the explicit min_confidence= / "
            "day_night= keyword arguments."
        )
    self._min_confidence = min_confidence
    self._day_night = day_night
    self._file_format: FileFormat = file_format
    self._timeout = timeout
    self._catalog = Catalog()
    # Reactive back-off knobs (G2); the sleep is an instance attr so
    # it can be swapped for a no-op in tests.
    self._sleep = time.sleep
    self._max_retries = 5
    self._backoff_factor = 1.0
    super().__init__(
        start=start,
        end=end,
        variables=list(variables) or list(_DEFAULT_SENSORS),
        temporal_resolution=temporal_resolution,
        lat_lim=lat_lim,
        lon_lim=lon_lim,
        fmt=fmt,
        path=path,
    )

authenticate(api_key=None)

Resolve the FIRMS MAP_KEY and arm the backend for download.

The explicit, fail-fast credential step. Pass api_key= to use a key directly; omit it (or pass None) to read the FIRMS_MAP_KEY environment variable. Either way the resolved key is held for the subsequent :meth:download. Calling it again with a different api_key re-arms with the new key. download() calls this with no argument on your behalf if you never do, so an explicit call is only needed to pass a key directly or to validate up front.

Parameters:

Name	Type	Description	Default
api_key	str | None	The FIRMS MAP_KEY to use. When None, the FIRMS_MAP_KEY environment variable is read instead.	None

Returns:

Type	Description
FIRMS	The backend instance, so it chains
FIRMS	EarthLens(...).authenticate(api_key=...).download().

Raises:

Type	Description
AuthenticationError	If api_key is None and no FIRMS_MAP_KEY environment variable is set.

Examples:

• Arm the backend with an explicit key and read it back:

  >>> import tempfile
  >>> from earthlens.firms import FIRMS
  >>> backend = FIRMS(
  ...     start="2024-08-01", end="2024-08-01",
  ...     variables=["VIIRS_SNPP_NRT"],
  ...     lat_lim=[33.0, 35.0], lon_lim=[-119.0, -117.0],
  ...     path=tempfile.mkdtemp(),
  ... )
  >>> backend.authenticate(api_key="demo-key").client.api_key
  'demo-key'
  

• A fresh backend is unauthenticated until the key resolves:

  >>> import tempfile
  >>> from earthlens.firms import FIRMS
  >>> backend = FIRMS(
  ...     start="2024-08-01", end="2024-08-01",
  ...     variables=["VIIRS_SNPP_NRT"],
  ...     lat_lim=[33.0, 35.0], lon_lim=[-119.0, -117.0],
  ...     path=tempfile.mkdtemp(),
  ... )
  >>> backend.client.is_authenticated()
  False
  >>> backend.authenticate(api_key="abc123").client.is_authenticated()
  True
  

Source code in src/earthlens/firms/backend.py

def authenticate(self, api_key: str | None = None) -> FIRMS:
    """Resolve the FIRMS `MAP_KEY` and arm the backend for download.

    The explicit, fail-fast credential step. Pass `api_key=` to use a
    key directly; omit it (or pass `None`) to read the `FIRMS_MAP_KEY`
    environment variable. Either way the resolved key is held for the
    subsequent :meth:`download`. Calling it again with a different
    `api_key` re-arms with the new key. `download()` calls this with
    no argument on your behalf if you never do, so an explicit call is
    only needed to pass a key directly or to validate up front.

    Args:
        api_key: The FIRMS `MAP_KEY` to use. When `None`, the
            `FIRMS_MAP_KEY` environment variable is read instead.

    Returns:
        The backend instance, so it chains
        `EarthLens(...).authenticate(api_key=...).download()`.

    Raises:
        AuthenticationError: If `api_key` is `None` and no
            `FIRMS_MAP_KEY` environment variable is set.

    Examples:
        - Arm the backend with an explicit key and read it back:
            ```python
            >>> import tempfile
            >>> from earthlens.firms import FIRMS
            >>> backend = FIRMS(
            ...     start="2024-08-01", end="2024-08-01",
            ...     variables=["VIIRS_SNPP_NRT"],
            ...     lat_lim=[33.0, 35.0], lon_lim=[-119.0, -117.0],
            ...     path=tempfile.mkdtemp(),
            ... )
            >>> backend.authenticate(api_key="demo-key").client.api_key
            'demo-key'

            ```
        - A fresh backend is unauthenticated until the key resolves:
            ```python
            >>> import tempfile
            >>> from earthlens.firms import FIRMS
            >>> backend = FIRMS(
            ...     start="2024-08-01", end="2024-08-01",
            ...     variables=["VIIRS_SNPP_NRT"],
            ...     lat_lim=[33.0, 35.0], lon_lim=[-119.0, -117.0],
            ...     path=tempfile.mkdtemp(),
            ... )
            >>> backend.client.is_authenticated()
            False
            >>> backend.authenticate(api_key="abc123").client.is_authenticated()
            True

            ```
    """
    auth = FirmsAuth(FirmsCredentials(api_key=api_key))
    auth.configure()
    self.client = auth
    return self

download(progress_bar=True, aggregate=None)

Query FIRMS and return the matched detections.

Runs the cheap :meth:_search (sensor validation + chunk planning) then the throttled :meth:_fetch (one CSV GET per chunk), concatenates the per-chunk collections into one FeatureCollection, writes it to one vector file under path, and returns it. An empty result returns — and writes nothing for — a schema-correct empty FeatureCollection.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Show a per-chunk progress bar. Defaults to True.	True
aggregate	AggregationConfig | None	Must be None. Detections are vector, not gridded, so there is no meaningful aggregation. The facade already rejects a non-None aggregate= for a vector backend; this is the belt-and-suspenders guard for direct backend callers.	None

Returns:

Name	Type	Description
FeatureCollection	FeatureCollection	The matched detections, CRS EPSG:4326. Empty (schema-only) when nothing matched.

Raises:

Type	Description
NotImplementedError	If aggregate is not None.

Source code in src/earthlens/firms/backend.py

def download(
    self,
    progress_bar: bool = True,
    aggregate: AggregationConfig | None = None,
) -> FeatureCollection:
    """Query FIRMS and return the matched detections.

    Runs the cheap :meth:`_search` (sensor validation + chunk
    planning) then the throttled :meth:`_fetch` (one CSV GET per
    chunk), concatenates the per-chunk collections into one
    FeatureCollection, writes it to one vector file under `path`, and
    returns it. An empty result returns — and writes nothing for — a
    schema-correct empty FeatureCollection.

    Args:
        progress_bar: Show a per-chunk progress bar. Defaults to
            `True`.
        aggregate: Must be `None`. Detections are vector, not
            gridded, so there is no meaningful aggregation. The
            facade already rejects a non-`None` `aggregate=` for a
            `vector` backend; this is the belt-and-suspenders guard
            for direct backend callers.

    Returns:
        FeatureCollection: The matched detections, CRS `EPSG:4326`.
            Empty (schema-only) when nothing matched.

    Raises:
        NotImplementedError: If `aggregate` is not `None`.
    """
    if aggregate is not None:
        raise NotImplementedError(
            "FIRMS.download(aggregate=...) is not supported: fire "
            "detections are vector point features, not gridded rasters, so "
            "there is no meaningful gridded reduction. Call download() "
            "without aggregate= and post-process the returned "
            "FeatureCollection (a GeoDataFrame) directly."
        )

    # Resolve the MAP_KEY from FIRMS_MAP_KEY if authenticate() was not
    # called explicitly, so EarthLens(...).download() still works when
    # the key lives in the environment.
    if not self.client.is_authenticated():
        self.authenticate()

    collections = self._api_via_search_fetch_with_progress(progress_bar)
    collection = events.concat(collections)

    if len(collection):
        out_path = self._write(collection)
        logger.info(
            f"FIRMS download summary: {len(collection)} detection(s) "
            f"written to {out_path}"
        )
    else:
        logger.warning(
            "FIRMS download summary: no detections matched the request, "
            "nothing written"
        )
    return collection

FirmsAuth

Bases: AbstractAuth[FirmsCredentials]

Resolve and hold the FIRMS MAP_KEY.

Implements the :class:earthlens.base.AbstractAuth contract for a single-secret backend. Construction does not touch the environment; :meth:configure performs the resolution and is idempotent. After a successful configure(), the key is available via the :attr:api_key property for the backend to drop into the request URL.

The class is a context manager (inherited from :class:AbstractAuth): with FirmsAuth(creds) as auth: ... calls configure() on enter and the default no-op close() on exit — there is no per-instance resource to release.

Attributes:

Name	Type	Description
_creds		The :class:FirmsCredentials passed at construction.

Examples:

• Resolve an explicit key:

  >>> from earthlens.firms import FirmsAuth, FirmsCredentials
  >>> auth = FirmsAuth(FirmsCredentials(api_key="k"))
  >>> auth.is_authenticated()
  False
  >>> auth.configure()
  >>> auth.is_authenticated()
  True
  >>> auth.api_key
  'k'
  

Source code in src/earthlens/firms/auth.py

class FirmsAuth(AbstractAuth[FirmsCredentials]):
    """Resolve and hold the FIRMS `MAP_KEY`.

    Implements the :class:`earthlens.base.AbstractAuth` contract for a
    single-secret backend. Construction does not touch the environment;
    :meth:`configure` performs the resolution and is idempotent. After a
    successful `configure()`, the key is available via the
    :attr:`api_key` property for the backend to drop into the request
    URL.

    The class is a context manager (inherited from
    :class:`AbstractAuth`): `with FirmsAuth(creds) as auth: ...` calls
    `configure()` on enter and the default no-op `close()` on exit —
    there is no per-instance resource to release.

    Attributes:
        _creds: The :class:`FirmsCredentials` passed at construction.

    Examples:
        - Resolve an explicit key:
            ```python
            >>> from earthlens.firms import FirmsAuth, FirmsCredentials
            >>> auth = FirmsAuth(FirmsCredentials(api_key="k"))
            >>> auth.is_authenticated()
            False
            >>> auth.configure()
            >>> auth.is_authenticated()
            True
            >>> auth.api_key
            'k'

            ```
    """

    def __init__(self, credentials: FirmsCredentials) -> None:
        """Store credentials; does not resolve the key yet.

        Args:
            credentials: The :class:`FirmsCredentials` value object
                carrying the optional `MAP_KEY`.
        """
        super().__init__(credentials)
        self._configured = False
        self._key: str | None = None

    def configure(self) -> None:
        """Resolve the `MAP_KEY` so subsequent requests can authenticate.

        Idempotent — short-circuits when :meth:`is_authenticated`
        already returns `True`. On the first call, resolves the key in
        this order: the explicit `api_key` on the credentials, then the
        `FIRMS_MAP_KEY` environment variable.

        Raises:
            AuthenticationError: When neither source supplies a key. The
                message names the `api_key=` argument, the
                `FIRMS_MAP_KEY` env var, and the free-registration URL —
                it never blocks on an interactive prompt.
        """
        if self.is_authenticated():
            return
        key = (
            self._creds.api_key.get_secret_value()
            if self._creds.api_key is not None
            else os.environ.get("FIRMS_MAP_KEY")
        )
        if not key:
            raise AuthenticationError(
                "no FIRMS MAP_KEY available: pass api_key= to "
                "EarthLens(...).authenticate() or set the FIRMS_MAP_KEY "
                f"environment variable. Request a free key at {_MAP_KEY_URL}."
            )
        self._key = key
        self._configured = True

    def is_authenticated(self) -> bool:
        """Return `True` once :meth:`configure` has resolved a key.

        Cheap predicate — does not call the network. A return of `True`
        means a usable key is held by this instance.

        Returns:
            bool: `True` after a successful :meth:`configure`, `False`
                before.
        """
        return self._configured

    @property
    def api_key(self) -> str:
        """The resolved `MAP_KEY`; valid only after :meth:`configure`.

        Returns:
            str: The FIRMS `MAP_KEY` string.

        Raises:
            AuthenticationError: When read before :meth:`configure` has
                resolved a key.
        """
        if self._key is None:
            raise AuthenticationError(
                "FirmsAuth.configure() has not run yet; no MAP_KEY resolved."
            )
        return self._key

api_key property

The resolved MAP_KEY; valid only after :meth:configure.

Returns:

Name	Type	Description
str	str	The FIRMS MAP_KEY string.

Raises:

Type	Description
AuthenticationError	When read before :meth:configure has resolved a key.

__init__(credentials)

Store credentials; does not resolve the key yet.

Parameters:

Name	Type	Description	Default
credentials	FirmsCredentials	The :class:FirmsCredentials value object carrying the optional MAP_KEY.	required

Source code in src/earthlens/firms/auth.py

def __init__(self, credentials: FirmsCredentials) -> None:
    """Store credentials; does not resolve the key yet.

    Args:
        credentials: The :class:`FirmsCredentials` value object
            carrying the optional `MAP_KEY`.
    """
    super().__init__(credentials)
    self._configured = False
    self._key: str | None = None

configure()

Resolve the MAP_KEY so subsequent requests can authenticate.

Idempotent — short-circuits when :meth:is_authenticated already returns True. On the first call, resolves the key in this order: the explicit api_key on the credentials, then the FIRMS_MAP_KEY environment variable.

Raises:

Type	Description
AuthenticationError	When neither source supplies a key. The message names the api_key= argument, the FIRMS_MAP_KEY env var, and the free-registration URL — it never blocks on an interactive prompt.

Source code in src/earthlens/firms/auth.py

def configure(self) -> None:
    """Resolve the `MAP_KEY` so subsequent requests can authenticate.

    Idempotent — short-circuits when :meth:`is_authenticated`
    already returns `True`. On the first call, resolves the key in
    this order: the explicit `api_key` on the credentials, then the
    `FIRMS_MAP_KEY` environment variable.

    Raises:
        AuthenticationError: When neither source supplies a key. The
            message names the `api_key=` argument, the
            `FIRMS_MAP_KEY` env var, and the free-registration URL —
            it never blocks on an interactive prompt.
    """
    if self.is_authenticated():
        return
    key = (
        self._creds.api_key.get_secret_value()
        if self._creds.api_key is not None
        else os.environ.get("FIRMS_MAP_KEY")
    )
    if not key:
        raise AuthenticationError(
            "no FIRMS MAP_KEY available: pass api_key= to "
            "EarthLens(...).authenticate() or set the FIRMS_MAP_KEY "
            f"environment variable. Request a free key at {_MAP_KEY_URL}."
        )
    self._key = key
    self._configured = True

is_authenticated()

Return True once :meth:configure has resolved a key.

Cheap predicate — does not call the network. A return of True means a usable key is held by this instance.

Returns:

Name	Type	Description
bool	bool	True after a successful :meth:configure, False before.

Source code in src/earthlens/firms/auth.py

def is_authenticated(self) -> bool:
    """Return `True` once :meth:`configure` has resolved a key.

    Cheap predicate — does not call the network. A return of `True`
    means a usable key is held by this instance.

    Returns:
        bool: `True` after a successful :meth:`configure`, `False`
            before.
    """
    return self._configured

FirmsCredentials

Bases: BaseModel

Frozen value object holding the FIRMS MAP_KEY.

The key is optional at construction time: None means "resolve from the FIRMS_MAP_KEY environment variable at :meth:FirmsAuth.configure time". The real "is there a usable key?" gate is :meth:FirmsAuth.configure, not this model.

Attributes:

Name	Type	Description
api_key	SecretStr | None	The FIRMS MAP_KEY, stored as a :class:pydantic.SecretStr so it is never echoed by repr(creds) or in logs. None defers resolution to the environment variable.

Examples:

• Build from an explicit key; the secret is hidden in repr:

  >>> from earthlens.firms import FirmsCredentials
  >>> creds = FirmsCredentials(api_key="topsecret")
  >>> creds.api_key.get_secret_value()
  'topsecret'
  >>> "topsecret" in repr(creds)
  False
  

• The key is optional — rely on the environment instead:

  >>> from earthlens.firms import FirmsCredentials
  >>> FirmsCredentials().api_key is None
  True
  

Source code in src/earthlens/firms/auth.py

class FirmsCredentials(BaseModel):
    """Frozen value object holding the FIRMS `MAP_KEY`.

    The key is optional at construction time: `None` means "resolve from
    the `FIRMS_MAP_KEY` environment variable at
    :meth:`FirmsAuth.configure` time". The real "is there a usable key?"
    gate is :meth:`FirmsAuth.configure`, not this model.

    Attributes:
        api_key: The FIRMS `MAP_KEY`, stored as a
            :class:`pydantic.SecretStr` so it is never echoed by
            `repr(creds)` or in logs. `None` defers resolution to the
            environment variable.

    Examples:
        - Build from an explicit key; the secret is hidden in `repr`:
            ```python
            >>> from earthlens.firms import FirmsCredentials
            >>> creds = FirmsCredentials(api_key="topsecret")
            >>> creds.api_key.get_secret_value()
            'topsecret'
            >>> "topsecret" in repr(creds)
            False

            ```
        - The key is optional — rely on the environment instead:
            ```python
            >>> from earthlens.firms import FirmsCredentials
            >>> FirmsCredentials().api_key is None
            True

            ```
    """

    model_config = ConfigDict(frozen=True)

    api_key: SecretStr | None = None

Sensor

Bases: BaseModel

One FIRMS sensor's dispatch row (the "dataset" analog).

The FIRMS source code is the parent key in :attr:Catalog.datasets and is repeated here as :attr:code so a :class:Sensor carries its own identity when passed around outside the catalog.

Attributes:

Name	Type	Description
code	str	FIRMS source code ("VIIRS_SNPP_NRT", "MODIS_NRT", …) — the value passed in variables=[...] and used as the URL source path segment.
name	str	Human-readable sensor name used in logs and docs.
family	Literal['MODIS', 'VIIRS', 'GOES', 'LANDSAT']	"MODIS", "VIIRS", "GOES", or "LANDSAT" — selects the confidence / brightness schema handling in :mod:earthlens.firms.events. MODIS and GOES report numeric confidence; VIIRS reports the categorical token l/n/h and LANDSAT reports l/m/h. Brightness comes from brightness (MODIS), bright_ti4 (VIIRS / GOES), or is absent (LANDSAT carries no brightness or FRP column).
resolution_m	int	Nominal nadir pixel size in metres (375 for VIIRS, 1000 for MODIS).
temporal	Temporal	The sensor's coverage window and quality tier.
columns	dict[str, SensorColumn]	Per-column metadata keyed by CSV column name.

Examples:

• Inspect a sensor's resolution and a column:

  >>> from earthlens.firms import Catalog
  >>> sensor = Catalog().get_sensor("VIIRS_SNPP_NRT")
  >>> sensor.resolution_m
  375
  >>> sensor.columns["frp"].units
  'MW'
  

Source code in src/earthlens/firms/catalog.py

class Sensor(BaseModel):
    """One FIRMS sensor's dispatch row (the "dataset" analog).

    The FIRMS source code is the parent key in :attr:`Catalog.datasets`
    and is repeated here as :attr:`code` so a :class:`Sensor` carries its
    own identity when passed around outside the catalog.

    Attributes:
        code: FIRMS source code (`"VIIRS_SNPP_NRT"`, `"MODIS_NRT"`, …) —
            the value passed in `variables=[...]` and used as the URL
            `source` path segment.
        name: Human-readable sensor name used in logs and docs.
        family: `"MODIS"`, `"VIIRS"`, `"GOES"`, or `"LANDSAT"` — selects
            the confidence / brightness schema handling in
            :mod:`earthlens.firms.events`. MODIS and GOES report numeric
            confidence; VIIRS reports the categorical token `l`/`n`/`h`
            and LANDSAT reports `l`/`m`/`h`. Brightness comes from
            `brightness` (MODIS), `bright_ti4` (VIIRS / GOES), or is
            absent (LANDSAT carries no brightness or FRP column).
        resolution_m: Nominal nadir pixel size in metres (375 for VIIRS,
            1000 for MODIS).
        temporal: The sensor's coverage window and quality tier.
        columns: Per-column metadata keyed by CSV column name.

    Examples:
        - Inspect a sensor's resolution and a column:
            ```python
            >>> from earthlens.firms import Catalog
            >>> sensor = Catalog().get_sensor("VIIRS_SNPP_NRT")
            >>> sensor.resolution_m
            375
            >>> sensor.columns["frp"].units
            'MW'

            ```
    """

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

    code: str
    name: str = ""
    family: Literal["MODIS", "VIIRS", "GOES", "LANDSAT"]
    resolution_m: int
    temporal: Temporal = Field(default_factory=Temporal)
    columns: dict[str, SensorColumn] = Field(default_factory=dict)

SensorColumn

Bases: BaseModel

One FIRMS CSV column's metadata (the "variable" analog).

A frozen value object describing a single column a sensor emits in its area-CSV response. Mirrors the ECMWF / GEE per-variable row, but minimal: FIRMS CSV columns carry no request-shaping parameters, only descriptive metadata.

Attributes:

Name	Type	Description
units	str	Physical unit of the column ("K", "MW", "%", or "1" for the dimensionless VIIRS confidence token).
long_name	str	Human-readable description used in docs and logs.

Examples:

• Build a column row directly:

  >>> from earthlens.firms import SensorColumn
  >>> col = SensorColumn(units="MW", long_name="Fire radiative power")
  >>> col.units
  'MW'
  

Source code in src/earthlens/firms/catalog.py

class SensorColumn(BaseModel):
    """One FIRMS CSV column's metadata (the "variable" analog).

    A frozen value object describing a single column a sensor emits in
    its area-CSV response. Mirrors the ECMWF / GEE per-variable row, but
    minimal: FIRMS CSV columns carry no request-shaping parameters, only
    descriptive metadata.

    Attributes:
        units: Physical unit of the column (`"K"`, `"MW"`, `"%"`, or
            `"1"` for the dimensionless VIIRS confidence token).
        long_name: Human-readable description used in docs and logs.

    Examples:
        - Build a column row directly:
            ```python
            >>> from earthlens.firms import SensorColumn
            >>> col = SensorColumn(units="MW", long_name="Fire radiative power")
            >>> col.units
            'MW'

            ```
    """

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

    units: str = ""
    long_name: str = ""

csv_to_fc(df, sensor, family, min_confidence=None, day_night=None)

Normalise one sensor's FIRMS CSV frame into a FeatureCollection.

One row per detection, columns per :data:ATTRIBUTE_COLUMNS plus a geometry column of Point(longitude, latitude). The MODIS/VIIRS confidence and brightness schemas are unified (G4), acq_date + integer-HHMM acq_time are combined into a tz-aware UTC acq_datetime, and the optional min_confidence / day_night filters are applied client-side (FIRMS offers no server-side equivalent). An empty input frame returns an empty FeatureCollection with the same columns/dtypes (see :func:empty_fc).

Parameters:

Name	Type	Description	Default
df	DataFrame	The decoded FIRMS area CSV for one sensor/chunk.	required
sensor	str	The FIRMS sensor code; recorded in the sensor column.	required
family	str	"MODIS" or "VIIRS" — selects the confidence and brightness source columns.	required
min_confidence	float | None	Optional 0-100 lower bound on the normalised confidence_pct; rows below it are dropped. Applied only to families whose confidence is a true 0-100 percent (MODIS / VIIRS / LANDSAT); for GOES (a provider-scale numeric confidence) the filter is skipped with a warning rather than silently dropping every row. None keeps all.	None
day_night	str | None	Optional "D" / "N" filter on the daynight column. None keeps both.	None

Returns:

Name	Type	Description
FeatureCollection	FeatureCollection	One feature per surviving detection, CRS EPSG:4326. Empty (schema-only) when the input is empty or the filters drop everything.

Examples:

• Map a one-row VIIRS frame; the l token becomes 25 %:

  >>> import pandas as pd
  >>> from earthlens.firms.events import csv_to_fc
  >>> df = pd.DataFrame(
  ...     {
  ...         "latitude": [34.0],
  ...         "longitude": [-118.0],
  ...         "acq_date": ["2024-08-01"],
  ...         "acq_time": [1325],
  ...         "satellite": ["N"],
  ...         "confidence": ["l"],
  ...         "bright_ti4": [320.0],
  ...         "frp": [12.5],
  ...         "daynight": ["D"],
  ...     }
  ... )
  >>> fc = csv_to_fc(df, "VIIRS_SNPP_NRT", "VIIRS")
  >>> float(fc["confidence_pct"].iloc[0])
  25.0
  >>> fc["acq_datetime"].iloc[0].strftime("%Y-%m-%d %H:%M")
  '2024-08-01 13:25'
  >>> fc.crs.to_epsg()
  4326
  

Source code in src/earthlens/firms/events.py

def csv_to_fc(
    df: pd.DataFrame,
    sensor: str,
    family: str,
    min_confidence: float | None = None,
    day_night: str | None = None,
) -> FeatureCollection:
    """Normalise one sensor's FIRMS CSV frame into a `FeatureCollection`.

    One row per detection, columns per :data:`ATTRIBUTE_COLUMNS` plus a
    `geometry` column of `Point(longitude, latitude)`. The MODIS/VIIRS
    confidence and brightness schemas are unified (`G4`), `acq_date` +
    integer-HHMM `acq_time` are combined into a tz-aware UTC
    `acq_datetime`, and the optional `min_confidence` / `day_night`
    filters are applied client-side (FIRMS offers no server-side
    equivalent). An empty input frame returns an empty FeatureCollection
    with the same columns/dtypes (see :func:`empty_fc`).

    Args:
        df: The decoded FIRMS area CSV for one sensor/chunk.
        sensor: The FIRMS sensor code; recorded in the `sensor` column.
        family: `"MODIS"` or `"VIIRS"` — selects the confidence and
            brightness source columns.
        min_confidence: Optional 0-100 lower bound on the normalised
            `confidence_pct`; rows below it are dropped. Applied only to
            families whose confidence is a true 0-100 percent
            (MODIS / VIIRS / LANDSAT); for GOES (a provider-scale numeric
            confidence) the filter is skipped with a warning rather than
            silently dropping every row. `None` keeps all.
        day_night: Optional `"D"` / `"N"` filter on the `daynight`
            column. `None` keeps both.

    Returns:
        FeatureCollection: One feature per surviving detection, CRS
            `EPSG:4326`. Empty (schema-only) when the input is empty or
            the filters drop everything.

    Examples:
        - Map a one-row VIIRS frame; the `l` token becomes 25 %:
            ```python
            >>> import pandas as pd
            >>> from earthlens.firms.events import csv_to_fc
            >>> df = pd.DataFrame(
            ...     {
            ...         "latitude": [34.0],
            ...         "longitude": [-118.0],
            ...         "acq_date": ["2024-08-01"],
            ...         "acq_time": [1325],
            ...         "satellite": ["N"],
            ...         "confidence": ["l"],
            ...         "bright_ti4": [320.0],
            ...         "frp": [12.5],
            ...         "daynight": ["D"],
            ...     }
            ... )
            >>> fc = csv_to_fc(df, "VIIRS_SNPP_NRT", "VIIRS")
            >>> float(fc["confidence_pct"].iloc[0])
            25.0
            >>> fc["acq_datetime"].iloc[0].strftime("%Y-%m-%d %H:%M")
            '2024-08-01 13:25'
            >>> fc.crs.to_epsg()
            4326

            ```
    """
    if df is None or df.empty:
        result = empty_fc()
    else:
        frame = pd.DataFrame(index=df.index)
        frame["latitude"] = pd.to_numeric(df.get("latitude"), errors="coerce")
        frame["longitude"] = pd.to_numeric(df.get("longitude"), errors="coerce")
        frame["acq_datetime"] = _acq_datetime(df)
        frame["sensor"] = sensor
        frame["satellite"] = _as_string(df.get("satellite"))
        raw_confidence = df.get("confidence")
        frame["confidence"] = _as_string(raw_confidence)
        frame["confidence_pct"] = _confidence_pct(raw_confidence, family)
        frame["brightness_k"] = _brightness(df, family)
        frame["frp"] = _numeric(df, "frp")
        frame["daynight"] = _as_string(df.get("daynight"))

        # `min_confidence` applies only to families whose confidence_pct is a
        # true 0-100 percent; non-percent families (GOES) intentionally skip
        # the filter (thresholding a ~0-1 provider scale would drop every
        # row). The backend warns about that once per download, not here per
        # chunk.
        if min_confidence is not None and family in PERCENT_CONFIDENCE_FAMILIES:
            frame = frame[frame["confidence_pct"] >= min_confidence]
        if day_night is not None:
            frame = frame[frame["daynight"] == day_night]

        if frame.empty:
            result = empty_fc()
        else:
            frame = frame.reset_index(drop=True)
            for column, dtype in ATTRIBUTE_COLUMNS.items():
                if column == "acq_datetime":
                    frame[column] = frame[column].astype(dtype)
                elif dtype == "string":
                    frame[column] = frame[column].astype("string")
            geometry = gpd.points_from_xy(frame["longitude"], frame["latitude"])
            gdf = gpd.GeoDataFrame(
                frame[list(ATTRIBUTE_COLUMNS)],
                geometry=gpd.GeoSeries(geometry, crs=DETECTION_CRS),
                crs=DETECTION_CRS,
            )
            result = FeatureCollection(gdf)
    return result

empty_fc()

Return an empty FeatureCollection with the canonical schema.

Used for an empty CSV, an out-of-coverage window, or a request whose filters dropped every row, so callers always get the same columns and dtypes back regardless of hit count.

Returns:

Name	Type	Description
FeatureCollection	FeatureCollection	Zero rows, the :data:ATTRIBUTE_COLUMNS columns with their declared dtypes, an empty geometry column, CRS EPSG:4326.

Examples:

• The schema is present even with no rows:

  >>> from earthlens.firms.events import empty_fc, ATTRIBUTE_COLUMNS
  >>> fc = empty_fc()
  >>> len(fc)
  0
  >>> set(ATTRIBUTE_COLUMNS).issubset(fc.columns)
  True
  >>> fc.crs.to_epsg()
  4326
  

Source code in src/earthlens/firms/events.py

def empty_fc() -> FeatureCollection:
    """Return an empty `FeatureCollection` with the canonical schema.

    Used for an empty CSV, an out-of-coverage window, or a request whose
    filters dropped every row, so callers always get the same columns
    and dtypes back regardless of hit count.

    Returns:
        FeatureCollection: Zero rows, the :data:`ATTRIBUTE_COLUMNS`
            columns with their declared dtypes, an empty `geometry`
            column, CRS `EPSG:4326`.

    Examples:
        - The schema is present even with no rows:
            ```python
            >>> from earthlens.firms.events import empty_fc, ATTRIBUTE_COLUMNS
            >>> fc = empty_fc()
            >>> len(fc)
            0
            >>> set(ATTRIBUTE_COLUMNS).issubset(fc.columns)
            True
            >>> fc.crs.to_epsg()
            4326

            ```
    """
    frame = pd.DataFrame(
        {
            column: pd.Series([], dtype=dtype)
            for column, dtype in ATTRIBUTE_COLUMNS.items()
        }
    )
    gdf = gpd.GeoDataFrame(
        frame, geometry=gpd.GeoSeries([], crs=DETECTION_CRS), crs=DETECTION_CRS
    )
    return FeatureCollection(gdf)

earthlens.firms.backend

Backend that queries the NASA FIRMS active-fire CSV API over HTTPS.

FIRMS(AbstractDataSource) fetches active-fire detections — one point per fire pixel, with brightness, confidence, and fire-radiative-power — from the NASA FIRMS (Fire Information for Resource Management System) area CSV endpoint, across MODIS (C6.1) and VIIRS (S-NPP / NOAA-20 / NOAA-21) sensors. The rows for a [start, end] window over a bbox come back as CSV, which :mod:earthlens.firms.events maps to a pyramids :class:~pyramids.feature.collection.FeatureCollection of fire-pixel points.

This is a vector backend: the on-the-wire result is a table of geolocated detections, not a gridded array, so OUTPUT_KIND = "vector" and the :class:earthlens.earthlens.EarthLens facade rejects an aggregate= argument (there is no meaningful gridded reduction of a detection table). download() returns the in-memory FeatureCollection and, as a side effect, writes it to one vector file under path.

FIRMS needs a free MAP_KEY — pass it to :meth:FIRMS.authenticate as api_key=, or set FIRMS_MAP_KEY and let authenticate() / download() read it from the environment. It is not a constructor argument: the constructor describes only what to fetch. There is no SDK and no [firms] extra — the only dependencies are requests + pandas, both core. Sensor selection follows the vector-backend reading of variables (see the package docstring): variables is a list[str] of FIRMS sensor codes (["VIIRS_SNPP_NRT"], ["MODIS_NRT", "VIIRS_SNPP_NRT"]); the detection filters ride as explicit min_confidence= / day_night= keyword arguments. The temporal window is chunked internally into ≤5-day requests (the FIRMS per-request cap), so temporal_resolution carries the sentinel "all".

FIRMS

Bases: AbstractDataSource

NASA FIRMS active-fire backend (vector point-feature output).

Wraps the FIRMS area CSV API so a user can pull a space/time/sensor window of fire detections through the same download() shape every other earthlens backend uses. Windows longer than the FIRMS 5-day per-request cap are chunked, and each (sensor, ≤5-day chunk) is one CSV GET; the rows are mapped to a :class:~pyramids.feature.collection.FeatureCollection.

FIRMS needs a free MAP_KEY. Supply it to :meth:authenticate as api_key=, or set the FIRMS_MAP_KEY environment variable and let authenticate() / download() resolve it. Credentials are not a constructor argument — the constructor describes only what to fetch.

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	"vector" — the result is a table of detection features, so the facade rejects aggregate= with NotImplementedError.

Source code in src/earthlens/firms/backend.py

class FIRMS(AbstractDataSource):
    """NASA FIRMS active-fire backend (vector point-feature output).

    Wraps the FIRMS area CSV API so a user can pull a space/time/sensor
    window of fire detections through the same `download()` shape every
    other earthlens backend uses. Windows longer than the FIRMS 5-day
    per-request cap are chunked, and each `(sensor, ≤5-day chunk)` is
    one CSV GET; the rows are mapped to a
    :class:`~pyramids.feature.collection.FeatureCollection`.

    FIRMS needs a free `MAP_KEY`. Supply it to :meth:`authenticate` as
    `api_key=`, or set the `FIRMS_MAP_KEY` environment variable and let
    `authenticate()` / `download()` resolve it. Credentials are not a
    constructor argument — the constructor describes only what to fetch.

    Attributes:
        OUTPUT_KIND: `"vector"` — the result is a table of detection
            features, so the facade rejects `aggregate=` with
            `NotImplementedError`.
    """

    OUTPUT_KIND: OutputKind = "vector"

    def __init__(
        self,
        start: str,
        end: str,
        variables: list[str],
        lat_lim: list[float],
        lon_lim: list[float],
        temporal_resolution: str = "all",
        path: Path | str = "",
        fmt: str = "%Y-%m-%d",
        min_confidence: float | None = None,
        day_night: str | None = None,
        file_format: FileFormat = "gpkg",
        timeout: float = 60.0,
    ):
        """Initialise a FIRMS backend instance.

        Args:
            start: Inclusive start of the detection window, as a string
                parsed with `fmt`.
            end: Inclusive end of the detection window.
            variables: List of FIRMS sensor codes to query
                (`["VIIRS_SNPP_NRT"]`, `["MODIS_NRT", "VIIRS_SNPP_NRT"]`).
                For this backend `variables` names the *sensors*, not
                data variables (see the package docstring). An empty list
                defaults to `["VIIRS_SNPP_NRT"]`.
            lat_lim: `[lat_min, lat_max]` bounding-box latitudes in
                degrees, both in `[-90, 90]`.
            lon_lim: `[lon_min, lon_max]` bounding-box longitudes in
                degrees, both in `[-180, 180]`.
            temporal_resolution: FIRMS chunks by ≤5-day windows
                internally, not by a daily/monthly cadence, so this is
                the sentinel `"all"`, not a pandas frequency alias.
            path: Output directory for the written vector file. Created
                by the parent class if absent.
            fmt: `strptime` format for `start` / `end`.
            min_confidence: Optional 0-100 lower bound applied
                client-side on the normalised `confidence_pct` column
                (FIRMS has no server-side confidence filter). `None`
                keeps every detection.
            day_night: Optional `"D"` / `"N"` filter applied client-side
                on the `daynight` column. `None` keeps both.
            file_format: Output vector format — `"gpkg"` (default,
                GeoPackage) or `"geojson"`.
            timeout: Per-request timeout in seconds for each CSV GET.

        Raises:
            ValueError: If `file_format` is not `"gpkg"` / `"geojson"`.
            TypeError: If `variables` is a mapping rather than a list of
                sensor codes.
        """
        if file_format not in _DRIVERS:
            raise ValueError(
                f"file_format must be one of {sorted(_DRIVERS)}, got "
                f"{file_format!r}."
            )
        if isinstance(variables, dict):
            raise TypeError(
                "FIRMS `variables` must be a list of sensor codes (e.g. "
                "['VIIRS_SNPP_NRT', 'MODIS_NRT']), not a mapping. For this "
                "backend `variables` selects sensors, not data variables; the "
                "detection filters are the explicit min_confidence= / "
                "day_night= keyword arguments."
            )
        self._min_confidence = min_confidence
        self._day_night = day_night
        self._file_format: FileFormat = file_format
        self._timeout = timeout
        self._catalog = Catalog()
        # Reactive back-off knobs (G2); the sleep is an instance attr so
        # it can be swapped for a no-op in tests.
        self._sleep = time.sleep
        self._max_retries = 5
        self._backoff_factor = 1.0
        super().__init__(
            start=start,
            end=end,
            variables=list(variables) or list(_DEFAULT_SENSORS),
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim,
            lon_lim=lon_lim,
            fmt=fmt,
            path=path,
        )

    def _initialize(self) -> FirmsAuth:
        """Build the (unconfigured) :class:`FirmsAuth` holder.

        No credentials are resolved here — the constructor describes only
        what to fetch. The `MAP_KEY` is resolved later by
        :meth:`authenticate` (explicitly via `api_key=`, or from the
        `FIRMS_MAP_KEY` environment variable), which `download()` also
        triggers lazily if it has not run.

        Returns:
            FirmsAuth: An unconfigured auth; `is_authenticated()` is
                `False` until :meth:`authenticate` resolves a key.
        """
        return FirmsAuth(FirmsCredentials(api_key=None))

    def authenticate(self, api_key: str | None = None) -> FIRMS:
        """Resolve the FIRMS `MAP_KEY` and arm the backend for download.

        The explicit, fail-fast credential step. Pass `api_key=` to use a
        key directly; omit it (or pass `None`) to read the `FIRMS_MAP_KEY`
        environment variable. Either way the resolved key is held for the
        subsequent :meth:`download`. Calling it again with a different
        `api_key` re-arms with the new key. `download()` calls this with
        no argument on your behalf if you never do, so an explicit call is
        only needed to pass a key directly or to validate up front.

        Args:
            api_key: The FIRMS `MAP_KEY` to use. When `None`, the
                `FIRMS_MAP_KEY` environment variable is read instead.

        Returns:
            The backend instance, so it chains
            `EarthLens(...).authenticate(api_key=...).download()`.

        Raises:
            AuthenticationError: If `api_key` is `None` and no
                `FIRMS_MAP_KEY` environment variable is set.

        Examples:
            - Arm the backend with an explicit key and read it back:
                ```python
                >>> import tempfile
                >>> from earthlens.firms import FIRMS
                >>> backend = FIRMS(
                ...     start="2024-08-01", end="2024-08-01",
                ...     variables=["VIIRS_SNPP_NRT"],
                ...     lat_lim=[33.0, 35.0], lon_lim=[-119.0, -117.0],
                ...     path=tempfile.mkdtemp(),
                ... )
                >>> backend.authenticate(api_key="demo-key").client.api_key
                'demo-key'

                ```
            - A fresh backend is unauthenticated until the key resolves:
                ```python
                >>> import tempfile
                >>> from earthlens.firms import FIRMS
                >>> backend = FIRMS(
                ...     start="2024-08-01", end="2024-08-01",
                ...     variables=["VIIRS_SNPP_NRT"],
                ...     lat_lim=[33.0, 35.0], lon_lim=[-119.0, -117.0],
                ...     path=tempfile.mkdtemp(),
                ... )
                >>> backend.client.is_authenticated()
                False
                >>> backend.authenticate(api_key="abc123").client.is_authenticated()
                True

                ```
        """
        auth = FirmsAuth(FirmsCredentials(api_key=api_key))
        auth.configure()
        self.client = auth
        return self

    def _create_grid(self, lat_lim: list, lon_lim: list) -> SpatialExtent:
        """Wrap the WGS84 bbox into a :class:`SpatialExtent` (no snapping).

        FIRMS clips server-side to the bbox path segment, so the box
        passes through unchanged.

        Args:
            lat_lim: `[lat_min, lat_max]` in degrees.
            lon_lim: `[lon_min, lon_max]` in degrees.

        Returns:
            SpatialExtent: Validated, frozen bbox.
        """
        return SpatialExtent.from_pairs(lat_lim=lat_lim, lon_lim=lon_lim)

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

        FIRMS chunks the window into ≤5-day requests internally (see
        :meth:`_search`), so the resolution is kept as the sentinel
        `"all"` (not a real pandas frequency alias) and `dates`
        collapses to the two endpoints.

        Args:
            start: Inclusive start date string.
            end: Inclusive end date string.
            temporal_resolution: Recorded as the resolution label;
                FIRMS always chunks the full window.
            fmt: `strptime` format applied to `start` and `end`.

        Returns:
            TemporalExtent: Frozen model with the parsed endpoints.

        Raises:
            ValueError: If `start` parses to a date later than `end`.
        """
        start_dt = dt.datetime.strptime(start, fmt)
        end_dt = dt.datetime.strptime(end, fmt)
        return TemporalExtent(
            start_date=start_dt,
            end_date=end_dt,
            resolution="all",
            dates=pd.DatetimeIndex([start_dt, end_dt]),
        )

    def _search(self) -> list[RemoteProduct]:
        """List one :class:`RemoteProduct` per `(sensor, ≤5-day chunk)`.

        Validates each code in `self.vars` against the bundled catalog
        (raising with a did-you-mean hint on an unknown sensor), warns
        when the requested window falls outside an `*_NRT` sensor's
        coverage (naming the `*_SP` archive variant — it does *not*
        auto-swap), and walks the `[start, end]` window in ≤5-day
        chunks. No network call is made here.

        Returns:
            list[RemoteProduct]: One product per `(sensor, chunk)`, whose
                `metadata` carries `sensor`, `family`, `start_date`, and
                `day_range`. The product `id` is `f"{sensor}:{start}"`.

        Raises:
            ValueError: If a code in `self.vars` is not a registered
                FIRMS sensor.
        """
        start_date = self.time.start_date.date()
        end_date = self.time.end_date.date()
        windows = chunk_windows(start_date, end_date)
        total_gets = len(self.vars) * len(windows)
        logger.info(
            f"FIRMS request: {len(self.vars)} sensor(s) x {len(windows)} chunk(s) "
            f"= {total_gets} CSV GET(s)"
        )
        if total_gets > FANOUT_WARN_THRESHOLD:
            logger.warning(
                f"FIRMS request fans out to {total_gets} CSV GET(s) (one "
                f"transaction each); FIRMS allows ~5000 per rolling 10 minutes. "
                "The per-request back-off will pace this, but consider narrowing "
                "the window or sensor list for a large pull."
            )
        products: list[RemoteProduct] = []
        non_percent: list[str] = []
        for code in self.vars:
            sensor = self._catalog.get_sensor(code)
            self._warn_if_out_of_coverage(sensor, start_date, end_date)
            if (
                self._min_confidence is not None
                and sensor.family not in events.PERCENT_CONFIDENCE_FAMILIES
            ):
                non_percent.append(code)
            for chunk_start, day_range in windows:
                products.append(
                    RemoteProduct(
                        id=f"{code}:{chunk_start.isoformat()}",
                        metadata={
                            "sensor": code,
                            "family": sensor.family,
                            "start_date": chunk_start,
                            "day_range": day_range,
                        },
                    )
                )
        if non_percent:
            logger.warning(
                f"min_confidence={self._min_confidence} is not applied to "
                f"{non_percent}: their confidence is a provider-scale (non "
                "0-100) value, so thresholding would drop every detection; "
                "those sensors' detections are kept unfiltered."
            )
        return products

    def _warn_if_out_of_coverage(
        self, sensor, start_date: dt.date, end_date: dt.date
    ) -> None:
        """Warn (do not auto-swap) when the window is outside coverage.

        An `*_NRT` sensor holds only roughly the last
        :data:`NRT_RETENTION_DAYS` days; a request for older data returns
        a silently empty CSV rather than an error. This logs a loud
        warning naming the `*_SP` archive variant when that variant
        exists. A request that predates the sensor's mission start is
        warned the same way.

        Args:
            sensor: The resolved :class:`~earthlens.firms.Sensor`.
            start_date: Requested inclusive start.
            end_date: Requested inclusive end.
        """
        # `temporal.start` / `temporal.end` are typed `datetime.date | None`
        # on the catalog model, so no datetime-narrowing is needed here.
        mission_start = sensor.temporal.start
        if mission_start is not None and start_date < mission_start:
            logger.warning(
                f"{sensor.code} coverage begins {mission_start}; the requested "
                f"window starts {start_date} and may return no detections."
            )
        coverage_end = sensor.temporal.end
        if coverage_end is not None and end_date > coverage_end:
            logger.warning(
                f"{sensor.code} coverage ends {coverage_end}; the requested "
                f"window ends {end_date} and may return no detections past the "
                "coverage end."
            )
        # The NRT-retention heuristic below is advisory (retention drifts
        # per sensor); it only applies to NRT sensors.
        if sensor.temporal.quality != "NRT":
            return
        cutoff = dt.date.today() - dt.timedelta(days=NRT_RETENTION_DAYS)
        if end_date < cutoff:
            sp_variant = sensor.code.replace("_NRT", "_SP")
            hint = (
                f" for archive data use {sp_variant}"
                if sp_variant in self._catalog
                else ""
            )
            logger.warning(
                f"{sensor.code} is near-real-time and covers only the last "
                f"~{NRT_RETENTION_DAYS} days; the requested window ending "
                f"{end_date} is older and will likely be empty{hint}."
            )

    def _fetch(self, products: list[RemoteProduct]) -> list[FeatureCollection]:
        """Fetch each product's CSV and map it to a FeatureCollection.

        Widens the inherited `-> list[Path]` contract: a vector backend
        returns in-memory :class:`FeatureCollection`s, not file paths.
        Each product is one CSV GET issued through the quota back-off
        (`G2`); the response body is classified before parsing (`G6`) so
        a FIRMS error-as-HTTP-200 text body never reaches
        `pandas.read_csv`.

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            list[FeatureCollection]: One collection per product, in the
                same order.
        """
        return [self._fetch_one(product) for product in products]

    def _fetch_one(self, product: RemoteProduct) -> FeatureCollection:
        """Fetch and map one `(sensor, chunk)` product.

        Args:
            product: One :class:`RemoteProduct` from :meth:`_search`.

        Returns:
            FeatureCollection: The chunk's detections (schema-only empty
                when the CSV had no rows).

        Raises:
            AuthenticationError: If the body is a bad-key message (`G6`).
            RuntimeError: If the body is a non-CSV error, or a quota body
                survives the back-off retries.
            requests.HTTPError: On a non-quota HTTP error status.
        """
        url = self._build_url(product)
        response = firms_get(
            url,
            timeout=self._timeout,
            get=requests.get,
            sleep=self._sleep,
            max_retries=self._max_retries,
            backoff_factor=self._backoff_factor,
        )
        status = getattr(response, "status_code", 200)
        if status >= 400:
            # Do NOT call response.raise_for_status(): its message embeds
            # the request URL, which carries the MAP_KEY as a path segment
            # and would leak the secret into logs/tracebacks. Raise a
            # redacted HTTPError instead.
            raise requests.HTTPError(
                f"FIRMS area request for sensor {product.metadata['sensor']} "
                f"failed with HTTP {status} (URL omitted to avoid leaking the "
                "MAP_KEY)."
            )
        text = response.text
        kind = classify_body(text)
        if kind == "auth":
            raise AuthenticationError(f"FIRMS rejected the MAP_KEY: {_truncate(text)}")
        if kind == "quota":
            raise RuntimeError(
                "FIRMS transaction quota exhausted after back-off retries: "
                f"{_truncate(text)}"
            )
        if kind == "error":
            raise RuntimeError(
                f"FIRMS returned a non-CSV error body: {_truncate(text)}"
            )
        frame = pd.read_csv(StringIO(text))
        return events.csv_to_fc(
            frame,
            sensor=product.metadata["sensor"],
            family=product.metadata["family"],
            min_confidence=self._min_confidence,
            day_night=self._day_night,
        )

    def _build_url(self, product: RemoteProduct) -> str:
        """Compose the FIRMS area-CSV URL for one product.

        The bbox path segment is `W,S,E,N` (FIRMS area order). The
        `MAP_KEY` is read from the configured :class:`FirmsAuth`.

        Args:
            product: One :class:`RemoteProduct` from :meth:`_search`.

        Returns:
            str: The fully-formed request URL.
        """
        bbox = (
            f"{self.space.west},{self.space.south},{self.space.east},{self.space.north}"
        )
        return AREA_URL_TEMPLATE.format(
            map_key=self.client.api_key,
            sensor=product.metadata["sensor"],
            bbox=bbox,
            day_range=product.metadata["day_range"],
            start_date=product.metadata["start_date"].isoformat(),
        )

    def _api(self) -> list[FeatureCollection]:
        """Compose `_search` and `_fetch` into the canonical C3 shape."""
        return self._api_via_search_fetch()

    def _api_via_search_fetch_with_progress(
        self, progress_bar: bool
    ) -> list[FeatureCollection]:
        """C3 composition with a per-chunk progress bar.

        Mirrors the CMEMS / OpenAQ progress-aware composition: run the
        cheap :meth:`_search`, then map :meth:`_fetch_one` over the
        products wrapped in a `tqdm` bar (disabled when `progress_bar`
        is `False`). Short-circuits on an empty search.

        Args:
            progress_bar: Show the per-chunk `tqdm` bar when `True`.

        Returns:
            list[FeatureCollection]: One collection per product, or `[]`
                when nothing matched.
        """
        return self._search_fetch_each(
            progress_bar=progress_bar, desc="FIRMS chunks", unit="chunk"
        )

    def download(
        self,
        progress_bar: bool = True,
        aggregate: AggregationConfig | None = None,
    ) -> FeatureCollection:
        """Query FIRMS and return the matched detections.

        Runs the cheap :meth:`_search` (sensor validation + chunk
        planning) then the throttled :meth:`_fetch` (one CSV GET per
        chunk), concatenates the per-chunk collections into one
        FeatureCollection, writes it to one vector file under `path`, and
        returns it. An empty result returns — and writes nothing for — a
        schema-correct empty FeatureCollection.

        Args:
            progress_bar: Show a per-chunk progress bar. Defaults to
                `True`.
            aggregate: Must be `None`. Detections are vector, not
                gridded, so there is no meaningful aggregation. The
                facade already rejects a non-`None` `aggregate=` for a
                `vector` backend; this is the belt-and-suspenders guard
                for direct backend callers.

        Returns:
            FeatureCollection: The matched detections, CRS `EPSG:4326`.
                Empty (schema-only) when nothing matched.

        Raises:
            NotImplementedError: If `aggregate` is not `None`.
        """
        if aggregate is not None:
            raise NotImplementedError(
                "FIRMS.download(aggregate=...) is not supported: fire "
                "detections are vector point features, not gridded rasters, so "
                "there is no meaningful gridded reduction. Call download() "
                "without aggregate= and post-process the returned "
                "FeatureCollection (a GeoDataFrame) directly."
            )

        # Resolve the MAP_KEY from FIRMS_MAP_KEY if authenticate() was not
        # called explicitly, so EarthLens(...).download() still works when
        # the key lives in the environment.
        if not self.client.is_authenticated():
            self.authenticate()

        collections = self._api_via_search_fetch_with_progress(progress_bar)
        collection = events.concat(collections)

        if len(collection):
            out_path = self._write(collection)
            logger.info(
                f"FIRMS download summary: {len(collection)} detection(s) "
                f"written to {out_path}"
            )
        else:
            logger.warning(
                "FIRMS download summary: no detections matched the request, "
                "nothing written"
            )
        return collection

    def _write(self, collection: FeatureCollection) -> Path:
        """Write the detections to one vector file under `root_dir`.

        The filename embeds the sensor list and the query's date window
        so successive downloads into the same `path` yield distinct
        files. Two downloads of the same request overwrite, the intended
        idempotent behaviour.

        Args:
            collection: The detections to write.

        Returns:
            Path: Absolute path of the file written.
        """
        driver, ext = _DRIVERS[self._file_format]
        sensors = "-".join(self.vars)
        stem = (
            f"firms_{sensors}_{self.time.start_date:%Y%m%d}"
            f"_{self.time.end_date:%Y%m%d}"
        )
        out_path = self.root_dir / f"{stem}.{ext}"
        collection.to_file(str(out_path), driver=driver)
        return out_path

__init__(start, end, variables, lat_lim, lon_lim, temporal_resolution='all', path='', fmt='%Y-%m-%d', min_confidence=None, day_night=None, file_format='gpkg', timeout=60.0)

Initialise a FIRMS backend instance.

Parameters:

Name	Type	Description	Default
start	str	Inclusive start of the detection window, as a string parsed with fmt.	required
end	str	Inclusive end of the detection window.	required
variables	list[str]	List of FIRMS sensor codes to query (["VIIRS_SNPP_NRT"], ["MODIS_NRT", "VIIRS_SNPP_NRT"]). For this backend variables names the sensors, not data variables (see the package docstring). An empty list defaults to ["VIIRS_SNPP_NRT"].	required
lat_lim	list[float]	[lat_min, lat_max] bounding-box latitudes in degrees, both in [-90, 90].	required
lon_lim	list[float]	[lon_min, lon_max] bounding-box longitudes in degrees, both in [-180, 180].	required
temporal_resolution	str	FIRMS chunks by ≤5-day windows internally, not by a daily/monthly cadence, so this is the sentinel "all", not a pandas frequency alias.	'all'
path	Path | str	Output directory for the written vector file. Created by the parent class if absent.	''
fmt	str	strptime format for start / end.	'%Y-%m-%d'
min_confidence	float | None	Optional 0-100 lower bound applied client-side on the normalised confidence_pct column (FIRMS has no server-side confidence filter). None keeps every detection.	None
day_night	str | None	Optional "D" / "N" filter applied client-side on the daynight column. None keeps both.	None
file_format	FileFormat	Output vector format — "gpkg" (default, GeoPackage) or "geojson".	'gpkg'
timeout	float	Per-request timeout in seconds for each CSV GET.	60.0

Raises:

Type	Description
ValueError	If file_format is not "gpkg" / "geojson".
TypeError	If variables is a mapping rather than a list of sensor codes.

Source code in src/earthlens/firms/backend.py

def __init__(
    self,
    start: str,
    end: str,
    variables: list[str],
    lat_lim: list[float],
    lon_lim: list[float],
    temporal_resolution: str = "all",
    path: Path | str = "",
    fmt: str = "%Y-%m-%d",
    min_confidence: float | None = None,
    day_night: str | None = None,
    file_format: FileFormat = "gpkg",
    timeout: float = 60.0,
):
    """Initialise a FIRMS backend instance.

    Args:
        start: Inclusive start of the detection window, as a string
            parsed with `fmt`.
        end: Inclusive end of the detection window.
        variables: List of FIRMS sensor codes to query
            (`["VIIRS_SNPP_NRT"]`, `["MODIS_NRT", "VIIRS_SNPP_NRT"]`).
            For this backend `variables` names the *sensors*, not
            data variables (see the package docstring). An empty list
            defaults to `["VIIRS_SNPP_NRT"]`.
        lat_lim: `[lat_min, lat_max]` bounding-box latitudes in
            degrees, both in `[-90, 90]`.
        lon_lim: `[lon_min, lon_max]` bounding-box longitudes in
            degrees, both in `[-180, 180]`.
        temporal_resolution: FIRMS chunks by ≤5-day windows
            internally, not by a daily/monthly cadence, so this is
            the sentinel `"all"`, not a pandas frequency alias.
        path: Output directory for the written vector file. Created
            by the parent class if absent.
        fmt: `strptime` format for `start` / `end`.
        min_confidence: Optional 0-100 lower bound applied
            client-side on the normalised `confidence_pct` column
            (FIRMS has no server-side confidence filter). `None`
            keeps every detection.
        day_night: Optional `"D"` / `"N"` filter applied client-side
            on the `daynight` column. `None` keeps both.
        file_format: Output vector format — `"gpkg"` (default,
            GeoPackage) or `"geojson"`.
        timeout: Per-request timeout in seconds for each CSV GET.

    Raises:
        ValueError: If `file_format` is not `"gpkg"` / `"geojson"`.
        TypeError: If `variables` is a mapping rather than a list of
            sensor codes.
    """
    if file_format not in _DRIVERS:
        raise ValueError(
            f"file_format must be one of {sorted(_DRIVERS)}, got "
            f"{file_format!r}."
        )
    if isinstance(variables, dict):
        raise TypeError(
            "FIRMS `variables` must be a list of sensor codes (e.g. "
            "['VIIRS_SNPP_NRT', 'MODIS_NRT']), not a mapping. For this "
            "backend `variables` selects sensors, not data variables; the "
            "detection filters are the explicit min_confidence= / "
            "day_night= keyword arguments."
        )
    self._min_confidence = min_confidence
    self._day_night = day_night
    self._file_format: FileFormat = file_format
    self._timeout = timeout
    self._catalog = Catalog()
    # Reactive back-off knobs (G2); the sleep is an instance attr so
    # it can be swapped for a no-op in tests.
    self._sleep = time.sleep
    self._max_retries = 5
    self._backoff_factor = 1.0
    super().__init__(
        start=start,
        end=end,
        variables=list(variables) or list(_DEFAULT_SENSORS),
        temporal_resolution=temporal_resolution,
        lat_lim=lat_lim,
        lon_lim=lon_lim,
        fmt=fmt,
        path=path,
    )

authenticate(api_key=None)

Resolve the FIRMS MAP_KEY and arm the backend for download.

The explicit, fail-fast credential step. Pass api_key= to use a key directly; omit it (or pass None) to read the FIRMS_MAP_KEY environment variable. Either way the resolved key is held for the subsequent :meth:download. Calling it again with a different api_key re-arms with the new key. download() calls this with no argument on your behalf if you never do, so an explicit call is only needed to pass a key directly or to validate up front.

Parameters:

Name	Type	Description	Default
api_key	str | None	The FIRMS MAP_KEY to use. When None, the FIRMS_MAP_KEY environment variable is read instead.	None

Returns:

Type	Description
FIRMS	The backend instance, so it chains
FIRMS	EarthLens(...).authenticate(api_key=...).download().

Raises:

Type	Description
AuthenticationError	If api_key is None and no FIRMS_MAP_KEY environment variable is set.

Examples:

• Arm the backend with an explicit key and read it back:

  >>> import tempfile
  >>> from earthlens.firms import FIRMS
  >>> backend = FIRMS(
  ...     start="2024-08-01", end="2024-08-01",
  ...     variables=["VIIRS_SNPP_NRT"],
  ...     lat_lim=[33.0, 35.0], lon_lim=[-119.0, -117.0],
  ...     path=tempfile.mkdtemp(),
  ... )
  >>> backend.authenticate(api_key="demo-key").client.api_key
  'demo-key'
  

• A fresh backend is unauthenticated until the key resolves:

  >>> import tempfile
  >>> from earthlens.firms import FIRMS
  >>> backend = FIRMS(
  ...     start="2024-08-01", end="2024-08-01",
  ...     variables=["VIIRS_SNPP_NRT"],
  ...     lat_lim=[33.0, 35.0], lon_lim=[-119.0, -117.0],
  ...     path=tempfile.mkdtemp(),
  ... )
  >>> backend.client.is_authenticated()
  False
  >>> backend.authenticate(api_key="abc123").client.is_authenticated()
  True
  

Source code in src/earthlens/firms/backend.py

def authenticate(self, api_key: str | None = None) -> FIRMS:
    """Resolve the FIRMS `MAP_KEY` and arm the backend for download.

    The explicit, fail-fast credential step. Pass `api_key=` to use a
    key directly; omit it (or pass `None`) to read the `FIRMS_MAP_KEY`
    environment variable. Either way the resolved key is held for the
    subsequent :meth:`download`. Calling it again with a different
    `api_key` re-arms with the new key. `download()` calls this with
    no argument on your behalf if you never do, so an explicit call is
    only needed to pass a key directly or to validate up front.

    Args:
        api_key: The FIRMS `MAP_KEY` to use. When `None`, the
            `FIRMS_MAP_KEY` environment variable is read instead.

    Returns:
        The backend instance, so it chains
        `EarthLens(...).authenticate(api_key=...).download()`.

    Raises:
        AuthenticationError: If `api_key` is `None` and no
            `FIRMS_MAP_KEY` environment variable is set.

    Examples:
        - Arm the backend with an explicit key and read it back:
            ```python
            >>> import tempfile
            >>> from earthlens.firms import FIRMS
            >>> backend = FIRMS(
            ...     start="2024-08-01", end="2024-08-01",
            ...     variables=["VIIRS_SNPP_NRT"],
            ...     lat_lim=[33.0, 35.0], lon_lim=[-119.0, -117.0],
            ...     path=tempfile.mkdtemp(),
            ... )
            >>> backend.authenticate(api_key="demo-key").client.api_key
            'demo-key'

            ```
        - A fresh backend is unauthenticated until the key resolves:
            ```python
            >>> import tempfile
            >>> from earthlens.firms import FIRMS
            >>> backend = FIRMS(
            ...     start="2024-08-01", end="2024-08-01",
            ...     variables=["VIIRS_SNPP_NRT"],
            ...     lat_lim=[33.0, 35.0], lon_lim=[-119.0, -117.0],
            ...     path=tempfile.mkdtemp(),
            ... )
            >>> backend.client.is_authenticated()
            False
            >>> backend.authenticate(api_key="abc123").client.is_authenticated()
            True

            ```
    """
    auth = FirmsAuth(FirmsCredentials(api_key=api_key))
    auth.configure()
    self.client = auth
    return self

download(progress_bar=True, aggregate=None)

Query FIRMS and return the matched detections.

Runs the cheap :meth:_search (sensor validation + chunk planning) then the throttled :meth:_fetch (one CSV GET per chunk), concatenates the per-chunk collections into one FeatureCollection, writes it to one vector file under path, and returns it. An empty result returns — and writes nothing for — a schema-correct empty FeatureCollection.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Show a per-chunk progress bar. Defaults to True.	True
aggregate	AggregationConfig | None	Must be None. Detections are vector, not gridded, so there is no meaningful aggregation. The facade already rejects a non-None aggregate= for a vector backend; this is the belt-and-suspenders guard for direct backend callers.	None

Returns:

Name	Type	Description
FeatureCollection	FeatureCollection	The matched detections, CRS EPSG:4326. Empty (schema-only) when nothing matched.

Raises:

Type	Description
NotImplementedError	If aggregate is not None.

Source code in src/earthlens/firms/backend.py

def download(
    self,
    progress_bar: bool = True,
    aggregate: AggregationConfig | None = None,
) -> FeatureCollection:
    """Query FIRMS and return the matched detections.

    Runs the cheap :meth:`_search` (sensor validation + chunk
    planning) then the throttled :meth:`_fetch` (one CSV GET per
    chunk), concatenates the per-chunk collections into one
    FeatureCollection, writes it to one vector file under `path`, and
    returns it. An empty result returns — and writes nothing for — a
    schema-correct empty FeatureCollection.

    Args:
        progress_bar: Show a per-chunk progress bar. Defaults to
            `True`.
        aggregate: Must be `None`. Detections are vector, not
            gridded, so there is no meaningful aggregation. The
            facade already rejects a non-`None` `aggregate=` for a
            `vector` backend; this is the belt-and-suspenders guard
            for direct backend callers.

    Returns:
        FeatureCollection: The matched detections, CRS `EPSG:4326`.
            Empty (schema-only) when nothing matched.

    Raises:
        NotImplementedError: If `aggregate` is not `None`.
    """
    if aggregate is not None:
        raise NotImplementedError(
            "FIRMS.download(aggregate=...) is not supported: fire "
            "detections are vector point features, not gridded rasters, so "
            "there is no meaningful gridded reduction. Call download() "
            "without aggregate= and post-process the returned "
            "FeatureCollection (a GeoDataFrame) directly."
        )

    # Resolve the MAP_KEY from FIRMS_MAP_KEY if authenticate() was not
    # called explicitly, so EarthLens(...).download() still works when
    # the key lives in the environment.
    if not self.client.is_authenticated():
        self.authenticate()

    collections = self._api_via_search_fetch_with_progress(progress_bar)
    collection = events.concat(collections)

    if len(collection):
        out_path = self._write(collection)
        logger.info(
            f"FIRMS download summary: {len(collection)} detection(s) "
            f"written to {out_path}"
        )
    else:
        logger.warning(
            "FIRMS download summary: no detections matched the request, "
            "nothing written"
        )
    return collection

earthlens.firms.events

Map a FIRMS area-CSV response into a pyramids FeatureCollection.

This module is the only place in the FIRMS backend that touches a GIS vector container, so per the pyramids policy it keeps all geometry/CRS handling inside pyramids primitives: earthlens assembles the normalised attribute rows, builds a Point geometry column from longitude / latitude, and hands the whole thing to :class:pyramids.feature.collection.FeatureCollection (a geopandas.GeoDataFrame subclass) tagged EPSG:4326.

The canonical detection schema lives here as :data:ATTRIBUTE_COLUMNS (attribute columns + dtypes) plus the geometry column. Both the populated path (:func:csv_to_fc) and the empty path (:func:empty_fc) produce a FeatureCollection with exactly these columns and dtypes, so a downstream to_file never chokes on a schema mismatch between a hit and a miss.

Two FIRMS data-shape wrinkles are absorbed here:

• Confidence differs by sensor family (G4). MODIS reports a numeric 0-100 confidence; VIIRS reports a categorical l/n/h token. The mapper keeps the raw value in confidence and derives a uniform confidence_pct float — VIIRS l/n/h map to 25/60/90, MODIS passes through. The raw confidence is always rendered as a string so the column dtype stays stable across families (categorical l/n/h and numeric 85 coexist); numeric consumers should read the float confidence_pct rather than re-parsing confidence. A single brightness_k column is filled from brightness (MODIS) or bright_ti4 (VIIRS), whichever the sensor provides.
• acq_time is an unpadded integer HHMM (e.g. 5 = 00:05, 1325 = 13:25), so it is split into hours/minutes and added to acq_date rather than string-concatenated.

Columns are read defensively (.get / column-presence checks) so a sensor missing a column degrades to NaN/None rather than raising.

concat(collections)

Concatenate per-chunk collections into one, schema-stable.

Parameters:

Name	Type	Description	Default
collections	list[FeatureCollection]	The per-(sensor, chunk) collections to merge.	required

Returns:

Name	Type	Description
FeatureCollection	FeatureCollection	Their row-wise union, CRS EPSG:4326, or a schema-only empty collection when every input was empty.

Examples:

• Concatenating only empty collections returns an empty one:

  >>> from earthlens.firms.events import concat, empty_fc
  >>> len(concat([empty_fc(), empty_fc()]))
  0
  

Source code in src/earthlens/firms/events.py

def concat(collections: list[FeatureCollection]) -> FeatureCollection:
    """Concatenate per-chunk collections into one, schema-stable.

    Args:
        collections: The per-`(sensor, chunk)` collections to merge.

    Returns:
        FeatureCollection: Their row-wise union, CRS `EPSG:4326`, or a
            schema-only empty collection when every input was empty.

    Examples:
        - Concatenating only empty collections returns an empty one:
            ```python
            >>> from earthlens.firms.events import concat, empty_fc
            >>> len(concat([empty_fc(), empty_fc()]))
            0

            ```
    """
    non_empty = [fc for fc in collections if len(fc)]
    if not non_empty:
        result = empty_fc()
    else:
        merged = pd.concat(non_empty, ignore_index=True)
        gdf = gpd.GeoDataFrame(merged, geometry="geometry", crs=DETECTION_CRS)
        result = FeatureCollection(gdf)
    return result

csv_to_fc(df, sensor, family, min_confidence=None, day_night=None)

Normalise one sensor's FIRMS CSV frame into a FeatureCollection.

One row per detection, columns per :data:ATTRIBUTE_COLUMNS plus a geometry column of Point(longitude, latitude). The MODIS/VIIRS confidence and brightness schemas are unified (G4), acq_date + integer-HHMM acq_time are combined into a tz-aware UTC acq_datetime, and the optional min_confidence / day_night filters are applied client-side (FIRMS offers no server-side equivalent). An empty input frame returns an empty FeatureCollection with the same columns/dtypes (see :func:empty_fc).

Parameters:

Name	Type	Description	Default
df	DataFrame	The decoded FIRMS area CSV for one sensor/chunk.	required
sensor	str	The FIRMS sensor code; recorded in the sensor column.	required
family	str	"MODIS" or "VIIRS" — selects the confidence and brightness source columns.	required
min_confidence	float | None	Optional 0-100 lower bound on the normalised confidence_pct; rows below it are dropped. Applied only to families whose confidence is a true 0-100 percent (MODIS / VIIRS / LANDSAT); for GOES (a provider-scale numeric confidence) the filter is skipped with a warning rather than silently dropping every row. None keeps all.	None
day_night	str | None	Optional "D" / "N" filter on the daynight column. None keeps both.	None

Returns:

Name	Type	Description
FeatureCollection	FeatureCollection	One feature per surviving detection, CRS EPSG:4326. Empty (schema-only) when the input is empty or the filters drop everything.

Examples:

• Map a one-row VIIRS frame; the l token becomes 25 %:

  >>> import pandas as pd
  >>> from earthlens.firms.events import csv_to_fc
  >>> df = pd.DataFrame(
  ...     {
  ...         "latitude": [34.0],
  ...         "longitude": [-118.0],
  ...         "acq_date": ["2024-08-01"],
  ...         "acq_time": [1325],
  ...         "satellite": ["N"],
  ...         "confidence": ["l"],
  ...         "bright_ti4": [320.0],
  ...         "frp": [12.5],
  ...         "daynight": ["D"],
  ...     }
  ... )
  >>> fc = csv_to_fc(df, "VIIRS_SNPP_NRT", "VIIRS")
  >>> float(fc["confidence_pct"].iloc[0])
  25.0
  >>> fc["acq_datetime"].iloc[0].strftime("%Y-%m-%d %H:%M")
  '2024-08-01 13:25'
  >>> fc.crs.to_epsg()
  4326
  

Source code in src/earthlens/firms/events.py

def csv_to_fc(
    df: pd.DataFrame,
    sensor: str,
    family: str,
    min_confidence: float | None = None,
    day_night: str | None = None,
) -> FeatureCollection:
    """Normalise one sensor's FIRMS CSV frame into a `FeatureCollection`.

    One row per detection, columns per :data:`ATTRIBUTE_COLUMNS` plus a
    `geometry` column of `Point(longitude, latitude)`. The MODIS/VIIRS
    confidence and brightness schemas are unified (`G4`), `acq_date` +
    integer-HHMM `acq_time` are combined into a tz-aware UTC
    `acq_datetime`, and the optional `min_confidence` / `day_night`
    filters are applied client-side (FIRMS offers no server-side
    equivalent). An empty input frame returns an empty FeatureCollection
    with the same columns/dtypes (see :func:`empty_fc`).

    Args:
        df: The decoded FIRMS area CSV for one sensor/chunk.
        sensor: The FIRMS sensor code; recorded in the `sensor` column.
        family: `"MODIS"` or `"VIIRS"` — selects the confidence and
            brightness source columns.
        min_confidence: Optional 0-100 lower bound on the normalised
            `confidence_pct`; rows below it are dropped. Applied only to
            families whose confidence is a true 0-100 percent
            (MODIS / VIIRS / LANDSAT); for GOES (a provider-scale numeric
            confidence) the filter is skipped with a warning rather than
            silently dropping every row. `None` keeps all.
        day_night: Optional `"D"` / `"N"` filter on the `daynight`
            column. `None` keeps both.

    Returns:
        FeatureCollection: One feature per surviving detection, CRS
            `EPSG:4326`. Empty (schema-only) when the input is empty or
            the filters drop everything.

    Examples:
        - Map a one-row VIIRS frame; the `l` token becomes 25 %:
            ```python
            >>> import pandas as pd
            >>> from earthlens.firms.events import csv_to_fc
            >>> df = pd.DataFrame(
            ...     {
            ...         "latitude": [34.0],
            ...         "longitude": [-118.0],
            ...         "acq_date": ["2024-08-01"],
            ...         "acq_time": [1325],
            ...         "satellite": ["N"],
            ...         "confidence": ["l"],
            ...         "bright_ti4": [320.0],
            ...         "frp": [12.5],
            ...         "daynight": ["D"],
            ...     }
            ... )
            >>> fc = csv_to_fc(df, "VIIRS_SNPP_NRT", "VIIRS")
            >>> float(fc["confidence_pct"].iloc[0])
            25.0
            >>> fc["acq_datetime"].iloc[0].strftime("%Y-%m-%d %H:%M")
            '2024-08-01 13:25'
            >>> fc.crs.to_epsg()
            4326

            ```
    """
    if df is None or df.empty:
        result = empty_fc()
    else:
        frame = pd.DataFrame(index=df.index)
        frame["latitude"] = pd.to_numeric(df.get("latitude"), errors="coerce")
        frame["longitude"] = pd.to_numeric(df.get("longitude"), errors="coerce")
        frame["acq_datetime"] = _acq_datetime(df)
        frame["sensor"] = sensor
        frame["satellite"] = _as_string(df.get("satellite"))
        raw_confidence = df.get("confidence")
        frame["confidence"] = _as_string(raw_confidence)
        frame["confidence_pct"] = _confidence_pct(raw_confidence, family)
        frame["brightness_k"] = _brightness(df, family)
        frame["frp"] = _numeric(df, "frp")
        frame["daynight"] = _as_string(df.get("daynight"))

        # `min_confidence` applies only to families whose confidence_pct is a
        # true 0-100 percent; non-percent families (GOES) intentionally skip
        # the filter (thresholding a ~0-1 provider scale would drop every
        # row). The backend warns about that once per download, not here per
        # chunk.
        if min_confidence is not None and family in PERCENT_CONFIDENCE_FAMILIES:
            frame = frame[frame["confidence_pct"] >= min_confidence]
        if day_night is not None:
            frame = frame[frame["daynight"] == day_night]

        if frame.empty:
            result = empty_fc()
        else:
            frame = frame.reset_index(drop=True)
            for column, dtype in ATTRIBUTE_COLUMNS.items():
                if column == "acq_datetime":
                    frame[column] = frame[column].astype(dtype)
                elif dtype == "string":
                    frame[column] = frame[column].astype("string")
            geometry = gpd.points_from_xy(frame["longitude"], frame["latitude"])
            gdf = gpd.GeoDataFrame(
                frame[list(ATTRIBUTE_COLUMNS)],
                geometry=gpd.GeoSeries(geometry, crs=DETECTION_CRS),
                crs=DETECTION_CRS,
            )
            result = FeatureCollection(gdf)
    return result

empty_fc()

Return an empty FeatureCollection with the canonical schema.

Used for an empty CSV, an out-of-coverage window, or a request whose filters dropped every row, so callers always get the same columns and dtypes back regardless of hit count.

Returns:

Name	Type	Description
FeatureCollection	FeatureCollection	Zero rows, the :data:ATTRIBUTE_COLUMNS columns with their declared dtypes, an empty geometry column, CRS EPSG:4326.

Examples:

• The schema is present even with no rows:

  >>> from earthlens.firms.events import empty_fc, ATTRIBUTE_COLUMNS
  >>> fc = empty_fc()
  >>> len(fc)
  0
  >>> set(ATTRIBUTE_COLUMNS).issubset(fc.columns)
  True
  >>> fc.crs.to_epsg()
  4326
  

Source code in src/earthlens/firms/events.py

def empty_fc() -> FeatureCollection:
    """Return an empty `FeatureCollection` with the canonical schema.

    Used for an empty CSV, an out-of-coverage window, or a request whose
    filters dropped every row, so callers always get the same columns
    and dtypes back regardless of hit count.

    Returns:
        FeatureCollection: Zero rows, the :data:`ATTRIBUTE_COLUMNS`
            columns with their declared dtypes, an empty `geometry`
            column, CRS `EPSG:4326`.

    Examples:
        - The schema is present even with no rows:
            ```python
            >>> from earthlens.firms.events import empty_fc, ATTRIBUTE_COLUMNS
            >>> fc = empty_fc()
            >>> len(fc)
            0
            >>> set(ATTRIBUTE_COLUMNS).issubset(fc.columns)
            True
            >>> fc.crs.to_epsg()
            4326

            ```
    """
    frame = pd.DataFrame(
        {
            column: pd.Series([], dtype=dtype)
            for column, dtype in ATTRIBUTE_COLUMNS.items()
        }
    )
    gdf = gpd.GeoDataFrame(
        frame, geometry=gpd.GeoSeries([], crs=DETECTION_CRS), crs=DETECTION_CRS
    )
    return FeatureCollection(gdf)

earthlens.firms.catalog

Sensor dispatch table for the NASA FIRMS active-fire backend.

FIRMS is a fixed set of active-fire sensors queried through one area CSV endpoint, not a curated dataset catalogue, so this "catalog" is small: a handful of rows mapping a FIRMS source code ("VIIRS_SNPP_NRT", "MODIS_NRT", …) to a little metadata. It follows the ECMWF / GEE convention where a sensor plays the "dataset" role and its CSV columns play the "variable" role — a :class:Sensor nests a columns: map of :class:SensorColumn rows.

Like gdacs_data_catalog.yaml / fdsn_data_catalog.yaml there is no available_* index: the listed sensors are the whole FIRMS universe, so an "available vs curated" split would just duplicate the map. Unlike those two, FIRMS does ship a probe + audit pair (tools/firms/) because the per-sensor CSV column schema varies by sensor family (MODIS reports a numeric confidence; VIIRS reports a categorical l/n/h), which is exactly the kind of drift a probe pins down.

:class:Catalog is a thin :class:earthlens.base.AbstractCatalog subclass that loads the bundled firms_data_catalog.yaml and exposes each row as a :class:Sensor, keyed by code under the inherited datasets field — which is what gives it the cat["MODIS_NRT"] / "MODIS_NRT" in cat / len(cat) dict-like surface and the did-you-mean error for free. :data:CATALOG_PATH is the path to the bundled YAML and is monkey-patchable in tests.

Catalog

Bases: AbstractCatalog

Sensor catalog for the NASA FIRMS backend.

Reads the bundled firms_data_catalog.yaml (shipped as package data) and exposes its sensors: block as a map of :class:Sensor rows, keyed by FIRMS source code under the inherited :attr:datasets field. Instantiate with no arguments (Catalog()); :func:model_post_init loads and validates the YAML in one pass. Resolve a sensor with :meth:get_sensor (a thin alias over :meth:~earthlens.base.AbstractCatalog.get_dataset) and a single column with :meth:get_column.

There is no available_* index — the listed sensors are the whole FIRMS universe (a deliberate deviation from the ECMWF/GEE catalogs, shared with GDACS/FDSN).

Attributes:

Name	Type	Description
datasets	dict[str, Sensor]	Map from the FIRMS source code to its :class:Sensor row.

Examples:

• List sensor codes and resolve one:

  >>> from earthlens.firms import Catalog
  >>> cat = Catalog()
  >>> cat.codes()  # doctest: +NORMALIZE_WHITESPACE
  ['GOES_NRT', 'LANDSAT_NRT', 'MODIS_NRT', 'MODIS_SP',
   'VIIRS_NOAA20_NRT', 'VIIRS_NOAA20_SP', 'VIIRS_NOAA21_NRT',
   'VIIRS_SNPP_NRT', 'VIIRS_SNPP_SP']
  >>> cat.get_sensor("MODIS_NRT").family
  'MODIS'
  >>> "MODIS_NRT" in cat
  True
  

• An unknown code raises with a did-you-mean hint:

  >>> from earthlens.firms import Catalog
  >>> Catalog().get_sensor("MODIS_NR")  # doctest: +ELLIPSIS
  Traceback (most recent call last):
      ...
  ValueError: 'MODIS_NR' is not in the FIRMS sensor catalog. Known sensors: [...]. Did you mean 'MODIS_NRT'?
  

Source code in src/earthlens/firms/catalog.py

class Catalog(AbstractCatalog):
    """Sensor catalog for the NASA FIRMS backend.

    Reads the bundled `firms_data_catalog.yaml` (shipped as package
    data) and exposes its `sensors:` block as a map of :class:`Sensor`
    rows, keyed by FIRMS source code under the inherited :attr:`datasets`
    field. Instantiate with no arguments (`Catalog()`);
    :func:`model_post_init` loads and validates the YAML in one pass.
    Resolve a sensor with :meth:`get_sensor` (a thin alias over
    :meth:`~earthlens.base.AbstractCatalog.get_dataset`) and a single
    column with :meth:`get_column`.

    There is no `available_*` index — the listed sensors are the whole
    FIRMS universe (a deliberate deviation from the ECMWF/GEE catalogs,
    shared with GDACS/FDSN).

    Attributes:
        datasets: Map from the FIRMS source code to its :class:`Sensor`
            row.

    Examples:
        - List sensor codes and resolve one:
            ```python
            >>> from earthlens.firms import Catalog
            >>> cat = Catalog()
            >>> cat.codes()  # doctest: +NORMALIZE_WHITESPACE
            ['GOES_NRT', 'LANDSAT_NRT', 'MODIS_NRT', 'MODIS_SP',
             'VIIRS_NOAA20_NRT', 'VIIRS_NOAA20_SP', 'VIIRS_NOAA21_NRT',
             'VIIRS_SNPP_NRT', 'VIIRS_SNPP_SP']
            >>> cat.get_sensor("MODIS_NRT").family
            'MODIS'
            >>> "MODIS_NRT" in cat
            True

            ```
        - An unknown code raises with a did-you-mean hint:
            ```python
            >>> from earthlens.firms import Catalog
            >>> Catalog().get_sensor("MODIS_NR")  # doctest: +ELLIPSIS
            Traceback (most recent call last):
                ...
            ValueError: 'MODIS_NR' is not in the FIRMS sensor catalog. Known sensors: [...]. Did you mean 'MODIS_NRT'?

            ```
    """

    _catalog_kind: str = "FIRMS sensor catalog"
    _entry_noun: str = "sensors"

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

    def model_post_init(self, __context: Any) -> None:
        """Auto-load the bundled catalog when no sensors were supplied.

        `Catalog()` with no args reads :data:`CATALOG_PATH`; passing
        `datasets=...` skips the disk read (used in tests).

        Raises:
            ValueError: Propagated from :meth:`load` when the YAML is
                missing, empty, or has a malformed sensor row.
        """
        if not self.datasets:
            loaded = Catalog.load()
            self.datasets = loaded.datasets
        super().model_post_init(__context)

    @classmethod
    def load(cls, catalog_path: Path | None = None) -> Catalog:
        """Read the FIRMS sensor catalog from disk.

        Args:
            catalog_path: Path to the catalog YAML. Defaults to the
                module-level :data:`CATALOG_PATH`.

        Returns:
            A fully-populated :class:`Catalog`.

        Raises:
            ValueError: If the file has no `sensors:` block, or a row
                fails :class:`Sensor` validation.
        """
        catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
        resolved = str(catalog_path.resolve())
        try:
            mtime = catalog_path.stat().st_mtime_ns
        except FileNotFoundError:
            mtime = 0
        key = (resolved, mtime)
        cached = _CATALOG_CACHE.get(key)
        if cached is not None:
            return cls(datasets=dict(cached))
        data = load_yaml_strict(catalog_path) or {}
        sensors_yaml = data.get("sensors") or {}
        if not sensors_yaml:
            raise ValueError(
                f"{catalog_path} is missing or has an empty 'sensors:' block. "
                "The FIRMS catalog must list at least one sensor."
            )
        sensors: dict[str, Sensor] = {}
        for code, body in sensors_yaml.items():
            try:
                sensors[code] = Sensor(**dict(body or {}))
            except ValidationError as exc:
                raise ValueError(
                    f"{catalog_path} sensor {code!r} failed validation:\n{exc}"
                ) from exc
        _CATALOG_CACHE[key] = sensors
        return cls(datasets=dict(sensors))

    def get_catalog(self) -> dict[str, Sensor]:
        """Return the sensor map (satisfies the abstract contract).

        Returns:
            dict[str, Sensor]: Same object as :attr:`datasets`.
        """
        return self.datasets

    def get_sensor(self, code: str) -> Sensor:
        """Return the :class:`Sensor` for `code`, with a did-you-mean hint.

        Thin alias over
        :meth:`~earthlens.base.AbstractCatalog.get_dataset`.

        Args:
            code: A FIRMS source code (`"VIIRS_SNPP_NRT"`, `"MODIS_NRT"`,
                …).

        Returns:
            Sensor: The matching sensor row.

        Raises:
            ValueError: If `code` is not a registered FIRMS sensor.
        """
        return self.get_dataset(code)

    def get_column(self, code: str, column: str) -> SensorColumn:
        """Return one column's metadata for a `(sensor, column)` pair.

        Args:
            code: A FIRMS source code as it appears in :attr:`datasets`.
            column: A CSV column name declared under that sensor.

        Returns:
            SensorColumn: The matching column metadata.

        Raises:
            ValueError: If `code` is not a registered sensor.
            KeyError: If `column` is not declared under that sensor.

        Examples:
            - Read a column's units:
                ```python
                >>> from earthlens.firms import Catalog
                >>> Catalog().get_column("MODIS_NRT", "confidence").units
                '%'

                ```
        """
        return self.get_sensor(code).columns[column]

    def get_variable(self, code: str, column: str) -> SensorColumn:
        """Leaf accessor for the shared two-arg get_variable contract.

        Alias of :meth:`get_column` so the FIRMS leaf is reachable under
        the same `get_variable(dataset_key, variable_name)` verb the
        other two-level catalogs use.

        Args:
            code: A FIRMS source code.
            column: A CSV column name declared under that sensor.

        Returns:
            SensorColumn: The matching column metadata.
        """
        return self.get_column(code, column)

    def codes(self) -> list[str]:
        """Return the registered FIRMS sensor codes, sorted.

        Returns:
            list[str]: The sensor codes (`["MODIS_NRT", "MODIS_SP", ...]`).
        """
        return sorted(self.datasets)

codes()

Return the registered FIRMS sensor codes, sorted.

Returns:

Type	Description
list[str]	list[str]: The sensor codes (["MODIS_NRT", "MODIS_SP", ...]).

Source code in src/earthlens/firms/catalog.py

def codes(self) -> list[str]:
    """Return the registered FIRMS sensor codes, sorted.

    Returns:
        list[str]: The sensor codes (`["MODIS_NRT", "MODIS_SP", ...]`).
    """
    return sorted(self.datasets)

get_catalog()

Return the sensor map (satisfies the abstract contract).

Returns:

Type	Description
dict[str, Sensor]	dict[str, Sensor]: Same object as :attr:datasets.

Source code in src/earthlens/firms/catalog.py

def get_catalog(self) -> dict[str, Sensor]:
    """Return the sensor map (satisfies the abstract contract).

    Returns:
        dict[str, Sensor]: Same object as :attr:`datasets`.
    """
    return self.datasets

get_column(code, column)

Return one column's metadata for a (sensor, column) pair.

Parameters:

Name	Type	Description	Default
code	str	A FIRMS source code as it appears in :attr:datasets.	required
column	str	A CSV column name declared under that sensor.	required

Returns:

Name	Type	Description
SensorColumn	SensorColumn	The matching column metadata.

Raises:

Type	Description
ValueError	If code is not a registered sensor.
KeyError	If column is not declared under that sensor.

Examples:

• Read a column's units:

  >>> from earthlens.firms import Catalog
  >>> Catalog().get_column("MODIS_NRT", "confidence").units
  '%'
  

Source code in src/earthlens/firms/catalog.py

def get_column(self, code: str, column: str) -> SensorColumn:
    """Return one column's metadata for a `(sensor, column)` pair.

    Args:
        code: A FIRMS source code as it appears in :attr:`datasets`.
        column: A CSV column name declared under that sensor.

    Returns:
        SensorColumn: The matching column metadata.

    Raises:
        ValueError: If `code` is not a registered sensor.
        KeyError: If `column` is not declared under that sensor.

    Examples:
        - Read a column's units:
            ```python
            >>> from earthlens.firms import Catalog
            >>> Catalog().get_column("MODIS_NRT", "confidence").units
            '%'

            ```
    """
    return self.get_sensor(code).columns[column]

get_sensor(code)

Return the :class:Sensor for code, with a did-you-mean hint.

Thin alias over :meth:~earthlens.base.AbstractCatalog.get_dataset.

Parameters:

Name	Type	Description	Default
code	str	A FIRMS source code ("VIIRS_SNPP_NRT", "MODIS_NRT", …).	required

Returns:

Name	Type	Description
Sensor	Sensor	The matching sensor row.

Raises:

Type	Description
ValueError	If code is not a registered FIRMS sensor.

Source code in src/earthlens/firms/catalog.py

def get_sensor(self, code: str) -> Sensor:
    """Return the :class:`Sensor` for `code`, with a did-you-mean hint.

    Thin alias over
    :meth:`~earthlens.base.AbstractCatalog.get_dataset`.

    Args:
        code: A FIRMS source code (`"VIIRS_SNPP_NRT"`, `"MODIS_NRT"`,
            …).

    Returns:
        Sensor: The matching sensor row.

    Raises:
        ValueError: If `code` is not a registered FIRMS sensor.
    """
    return self.get_dataset(code)

get_variable(code, column)

Leaf accessor for the shared two-arg get_variable contract.

Alias of :meth:get_column so the FIRMS leaf is reachable under the same get_variable(dataset_key, variable_name) verb the other two-level catalogs use.

Parameters:

Name	Type	Description	Default
code	str	A FIRMS source code.	required
column	str	A CSV column name declared under that sensor.	required

Returns:

Name	Type	Description
SensorColumn	SensorColumn	The matching column metadata.

Source code in src/earthlens/firms/catalog.py

def get_variable(self, code: str, column: str) -> SensorColumn:
    """Leaf accessor for the shared two-arg get_variable contract.

    Alias of :meth:`get_column` so the FIRMS leaf is reachable under
    the same `get_variable(dataset_key, variable_name)` verb the
    other two-level catalogs use.

    Args:
        code: A FIRMS source code.
        column: A CSV column name declared under that sensor.

    Returns:
        SensorColumn: The matching column metadata.
    """
    return self.get_column(code, column)

load(catalog_path=None) classmethod

Read the FIRMS sensor catalog from disk.

Parameters:

Name	Type	Description	Default
catalog_path	Path | None	Path to the catalog YAML. Defaults to the module-level :data:CATALOG_PATH.	None

Returns:

Type	Description
Catalog	A fully-populated :class:Catalog.

Raises:

Type	Description
ValueError	If the file has no sensors: block, or a row fails :class:Sensor validation.

Source code in src/earthlens/firms/catalog.py

@classmethod
def load(cls, catalog_path: Path | None = None) -> Catalog:
    """Read the FIRMS sensor catalog from disk.

    Args:
        catalog_path: Path to the catalog YAML. Defaults to the
            module-level :data:`CATALOG_PATH`.

    Returns:
        A fully-populated :class:`Catalog`.

    Raises:
        ValueError: If the file has no `sensors:` block, or a row
            fails :class:`Sensor` validation.
    """
    catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
    resolved = str(catalog_path.resolve())
    try:
        mtime = catalog_path.stat().st_mtime_ns
    except FileNotFoundError:
        mtime = 0
    key = (resolved, mtime)
    cached = _CATALOG_CACHE.get(key)
    if cached is not None:
        return cls(datasets=dict(cached))
    data = load_yaml_strict(catalog_path) or {}
    sensors_yaml = data.get("sensors") or {}
    if not sensors_yaml:
        raise ValueError(
            f"{catalog_path} is missing or has an empty 'sensors:' block. "
            "The FIRMS catalog must list at least one sensor."
        )
    sensors: dict[str, Sensor] = {}
    for code, body in sensors_yaml.items():
        try:
            sensors[code] = Sensor(**dict(body or {}))
        except ValidationError as exc:
            raise ValueError(
                f"{catalog_path} sensor {code!r} failed validation:\n{exc}"
            ) from exc
    _CATALOG_CACHE[key] = sensors
    return cls(datasets=dict(sensors))

model_post_init(__context)

Auto-load the bundled catalog when no sensors were supplied.

Catalog() with no args reads :data:CATALOG_PATH; passing datasets=... skips the disk read (used in tests).

Raises:

Type	Description
ValueError	Propagated from :meth:load when the YAML is missing, empty, or has a malformed sensor row.

Source code in src/earthlens/firms/catalog.py

def model_post_init(self, __context: Any) -> None:
    """Auto-load the bundled catalog when no sensors were supplied.

    `Catalog()` with no args reads :data:`CATALOG_PATH`; passing
    `datasets=...` skips the disk read (used in tests).

    Raises:
        ValueError: Propagated from :meth:`load` when the YAML is
            missing, empty, or has a malformed sensor row.
    """
    if not self.datasets:
        loaded = Catalog.load()
        self.datasets = loaded.datasets
    super().model_post_init(__context)

Sensor

Bases: BaseModel

One FIRMS sensor's dispatch row (the "dataset" analog).

The FIRMS source code is the parent key in :attr:Catalog.datasets and is repeated here as :attr:code so a :class:Sensor carries its own identity when passed around outside the catalog.

Attributes:

Name	Type	Description
code	str	FIRMS source code ("VIIRS_SNPP_NRT", "MODIS_NRT", …) — the value passed in variables=[...] and used as the URL source path segment.
name	str	Human-readable sensor name used in logs and docs.
family	Literal['MODIS', 'VIIRS', 'GOES', 'LANDSAT']	"MODIS", "VIIRS", "GOES", or "LANDSAT" — selects the confidence / brightness schema handling in :mod:earthlens.firms.events. MODIS and GOES report numeric confidence; VIIRS reports the categorical token l/n/h and LANDSAT reports l/m/h. Brightness comes from brightness (MODIS), bright_ti4 (VIIRS / GOES), or is absent (LANDSAT carries no brightness or FRP column).
resolution_m	int	Nominal nadir pixel size in metres (375 for VIIRS, 1000 for MODIS).
temporal	Temporal	The sensor's coverage window and quality tier.
columns	dict[str, SensorColumn]	Per-column metadata keyed by CSV column name.

Examples:

• Inspect a sensor's resolution and a column:

  >>> from earthlens.firms import Catalog
  >>> sensor = Catalog().get_sensor("VIIRS_SNPP_NRT")
  >>> sensor.resolution_m
  375
  >>> sensor.columns["frp"].units
  'MW'
  

Source code in src/earthlens/firms/catalog.py

class Sensor(BaseModel):
    """One FIRMS sensor's dispatch row (the "dataset" analog).

    The FIRMS source code is the parent key in :attr:`Catalog.datasets`
    and is repeated here as :attr:`code` so a :class:`Sensor` carries its
    own identity when passed around outside the catalog.

    Attributes:
        code: FIRMS source code (`"VIIRS_SNPP_NRT"`, `"MODIS_NRT"`, …) —
            the value passed in `variables=[...]` and used as the URL
            `source` path segment.
        name: Human-readable sensor name used in logs and docs.
        family: `"MODIS"`, `"VIIRS"`, `"GOES"`, or `"LANDSAT"` — selects
            the confidence / brightness schema handling in
            :mod:`earthlens.firms.events`. MODIS and GOES report numeric
            confidence; VIIRS reports the categorical token `l`/`n`/`h`
            and LANDSAT reports `l`/`m`/`h`. Brightness comes from
            `brightness` (MODIS), `bright_ti4` (VIIRS / GOES), or is
            absent (LANDSAT carries no brightness or FRP column).
        resolution_m: Nominal nadir pixel size in metres (375 for VIIRS,
            1000 for MODIS).
        temporal: The sensor's coverage window and quality tier.
        columns: Per-column metadata keyed by CSV column name.

    Examples:
        - Inspect a sensor's resolution and a column:
            ```python
            >>> from earthlens.firms import Catalog
            >>> sensor = Catalog().get_sensor("VIIRS_SNPP_NRT")
            >>> sensor.resolution_m
            375
            >>> sensor.columns["frp"].units
            'MW'

            ```
    """

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

    code: str
    name: str = ""
    family: Literal["MODIS", "VIIRS", "GOES", "LANDSAT"]
    resolution_m: int
    temporal: Temporal = Field(default_factory=Temporal)
    columns: dict[str, SensorColumn] = Field(default_factory=dict)

SensorColumn

Bases: BaseModel

One FIRMS CSV column's metadata (the "variable" analog).

A frozen value object describing a single column a sensor emits in its area-CSV response. Mirrors the ECMWF / GEE per-variable row, but minimal: FIRMS CSV columns carry no request-shaping parameters, only descriptive metadata.

Attributes:

Name	Type	Description
units	str	Physical unit of the column ("K", "MW", "%", or "1" for the dimensionless VIIRS confidence token).
long_name	str	Human-readable description used in docs and logs.

Examples:

• Build a column row directly:

  >>> from earthlens.firms import SensorColumn
  >>> col = SensorColumn(units="MW", long_name="Fire radiative power")
  >>> col.units
  'MW'
  

Source code in src/earthlens/firms/catalog.py

class SensorColumn(BaseModel):
    """One FIRMS CSV column's metadata (the "variable" analog).

    A frozen value object describing a single column a sensor emits in
    its area-CSV response. Mirrors the ECMWF / GEE per-variable row, but
    minimal: FIRMS CSV columns carry no request-shaping parameters, only
    descriptive metadata.

    Attributes:
        units: Physical unit of the column (`"K"`, `"MW"`, `"%"`, or
            `"1"` for the dimensionless VIIRS confidence token).
        long_name: Human-readable description used in docs and logs.

    Examples:
        - Build a column row directly:
            ```python
            >>> from earthlens.firms import SensorColumn
            >>> col = SensorColumn(units="MW", long_name="Fire radiative power")
            >>> col.units
            'MW'

            ```
    """

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

    units: str = ""
    long_name: str = ""

Temporal

Bases: BaseModel

A sensor's coverage window and quality tier.

Attributes:

Name	Type	Description
start	date | None	First date the sensor has data for, or None if unknown.
end	date | None	Last date covered, or None for an ongoing sensor.
quality	Literal['NRT', 'SP']	"NRT" (near-real-time, last ~2 months) or "SP" (standard-quality archive). Drives the :class:~earthlens.firms.FIRMS out-of-coverage warning: a request for an old window against an *_NRT sensor is silently empty upstream, so the backend warns and names the *_SP variant.

Examples:

• An ongoing NRT sensor:

  >>> from earthlens.firms.catalog import Temporal
  >>> t = Temporal(start="2012-01-20", quality="NRT")
  >>> t.end is None
  True
  

Source code in src/earthlens/firms/catalog.py

class Temporal(BaseModel):
    """A sensor's coverage window and quality tier.

    Attributes:
        start: First date the sensor has data for, or `None` if unknown.
        end: Last date covered, or `None` for an ongoing sensor.
        quality: `"NRT"` (near-real-time, last ~2 months) or `"SP"`
            (standard-quality archive). Drives the
            :class:`~earthlens.firms.FIRMS` out-of-coverage warning: a
            request for an old window against an `*_NRT` sensor is
            silently empty upstream, so the backend warns and names the
            `*_SP` variant.

    Examples:
        - An ongoing NRT sensor:
            ```python
            >>> from earthlens.firms.catalog import Temporal
            >>> t = Temporal(start="2012-01-20", quality="NRT")
            >>> t.end is None
            True

            ```
    """

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

    start: dt.date | None = None
    end: dt.date | None = None
    quality: Literal["NRT", "SP"] = "NRT"

clear_catalog_cache()

Empty the module-level catalog parse cache (for tests that rewrite YAML).

Source code in src/earthlens/firms/catalog.py

def clear_catalog_cache() -> None:
    """Empty the module-level catalog parse cache (for tests that rewrite YAML)."""
    _CATALOG_CACHE.clear()

earthlens.firms.auth

Credentials and MAP_KEY resolution for the NASA FIRMS backend.

Hosts :class:FirmsAuth, an :class:earthlens.base.AbstractAuth subclass that resolves a single FIRMS MAP_KEY from, in priority order, an explicit api_key= argument or the FIRMS_MAP_KEY environment variable. FIRMS requires a (free) key on every request; there is no username/password and no saved config-file dance, so this is the same single-secret shape as :class:earthlens.openaq.OpenaqAuth — mirrored on the CmemsAuth resolution chain but without the toolbox login.

Unlike OpenAQ (which attaches its key as an X-API-Key header via a dedicated client), FIRMS sends the MAP_KEY as a path segment in the request URL, so there is no separate client module: the resolved key is read back via the :attr:FirmsAuth.api_key property and dropped into the URL by :class:earthlens.firms.FIRMS directly.

The shape:

• :class:FirmsCredentials is a frozen pydantic value object carrying the optional key as a :class:pydantic.SecretStr.
• :class:FirmsAuth binds those credentials and resolves the key in :meth:FirmsAuth.configure — explicit key first, then the FIRMS_MAP_KEY env var, then a clear :class:AuthenticationError naming the free-registration URL (never an interactive prompt).
• configure() is idempotent — a second call after :meth:FirmsAuth.is_authenticated returns True short-circuits, so it is safe to call from long-lived workers.

AuthenticationError

Bases: AuthenticationError

Raised when no usable FIRMS MAP_KEY can be resolved.

Carries a message that names a fix: pass api_key= to EarthLens(...).authenticate(), set the FIRMS_MAP_KEY environment variable, or request a free key at firms.modaps.eosdis.nasa.gov/api/map_key/. A subclass of the cross-backend :class:earthlens.base.AuthenticationError so callers can catch every backend's auth failure with one except clause.

Source code in src/earthlens/firms/auth.py

class AuthenticationError(_BaseAuthenticationError):
    """Raised when no usable FIRMS `MAP_KEY` can be resolved.

    Carries a message that names a fix: pass `api_key=` to
    `EarthLens(...).authenticate()`, set the `FIRMS_MAP_KEY` environment
    variable, or request a free key at
    `firms.modaps.eosdis.nasa.gov/api/map_key/`. A subclass of the
    cross-backend :class:`earthlens.base.AuthenticationError` so callers
    can catch every backend's auth failure with one `except` clause.
    """

FirmsAuth

Bases: AbstractAuth[FirmsCredentials]

Resolve and hold the FIRMS MAP_KEY.

Implements the :class:earthlens.base.AbstractAuth contract for a single-secret backend. Construction does not touch the environment; :meth:configure performs the resolution and is idempotent. After a successful configure(), the key is available via the :attr:api_key property for the backend to drop into the request URL.

The class is a context manager (inherited from :class:AbstractAuth): with FirmsAuth(creds) as auth: ... calls configure() on enter and the default no-op close() on exit — there is no per-instance resource to release.

Attributes:

Name	Type	Description
_creds		The :class:FirmsCredentials passed at construction.

Examples:

• Resolve an explicit key:

  >>> from earthlens.firms import FirmsAuth, FirmsCredentials
  >>> auth = FirmsAuth(FirmsCredentials(api_key="k"))
  >>> auth.is_authenticated()
  False
  >>> auth.configure()
  >>> auth.is_authenticated()
  True
  >>> auth.api_key
  'k'
  

Source code in src/earthlens/firms/auth.py

class FirmsAuth(AbstractAuth[FirmsCredentials]):
    """Resolve and hold the FIRMS `MAP_KEY`.

    Implements the :class:`earthlens.base.AbstractAuth` contract for a
    single-secret backend. Construction does not touch the environment;
    :meth:`configure` performs the resolution and is idempotent. After a
    successful `configure()`, the key is available via the
    :attr:`api_key` property for the backend to drop into the request
    URL.

    The class is a context manager (inherited from
    :class:`AbstractAuth`): `with FirmsAuth(creds) as auth: ...` calls
    `configure()` on enter and the default no-op `close()` on exit —
    there is no per-instance resource to release.

    Attributes:
        _creds: The :class:`FirmsCredentials` passed at construction.

    Examples:
        - Resolve an explicit key:
            ```python
            >>> from earthlens.firms import FirmsAuth, FirmsCredentials
            >>> auth = FirmsAuth(FirmsCredentials(api_key="k"))
            >>> auth.is_authenticated()
            False
            >>> auth.configure()
            >>> auth.is_authenticated()
            True
            >>> auth.api_key
            'k'

            ```
    """

    def __init__(self, credentials: FirmsCredentials) -> None:
        """Store credentials; does not resolve the key yet.

        Args:
            credentials: The :class:`FirmsCredentials` value object
                carrying the optional `MAP_KEY`.
        """
        super().__init__(credentials)
        self._configured = False
        self._key: str | None = None

    def configure(self) -> None:
        """Resolve the `MAP_KEY` so subsequent requests can authenticate.

        Idempotent — short-circuits when :meth:`is_authenticated`
        already returns `True`. On the first call, resolves the key in
        this order: the explicit `api_key` on the credentials, then the
        `FIRMS_MAP_KEY` environment variable.

        Raises:
            AuthenticationError: When neither source supplies a key. The
                message names the `api_key=` argument, the
                `FIRMS_MAP_KEY` env var, and the free-registration URL —
                it never blocks on an interactive prompt.
        """
        if self.is_authenticated():
            return
        key = (
            self._creds.api_key.get_secret_value()
            if self._creds.api_key is not None
            else os.environ.get("FIRMS_MAP_KEY")
        )
        if not key:
            raise AuthenticationError(
                "no FIRMS MAP_KEY available: pass api_key= to "
                "EarthLens(...).authenticate() or set the FIRMS_MAP_KEY "
                f"environment variable. Request a free key at {_MAP_KEY_URL}."
            )
        self._key = key
        self._configured = True

    def is_authenticated(self) -> bool:
        """Return `True` once :meth:`configure` has resolved a key.

        Cheap predicate — does not call the network. A return of `True`
        means a usable key is held by this instance.

        Returns:
            bool: `True` after a successful :meth:`configure`, `False`
                before.
        """
        return self._configured

    @property
    def api_key(self) -> str:
        """The resolved `MAP_KEY`; valid only after :meth:`configure`.

        Returns:
            str: The FIRMS `MAP_KEY` string.

        Raises:
            AuthenticationError: When read before :meth:`configure` has
                resolved a key.
        """
        if self._key is None:
            raise AuthenticationError(
                "FirmsAuth.configure() has not run yet; no MAP_KEY resolved."
            )
        return self._key

api_key property

The resolved MAP_KEY; valid only after :meth:configure.

Returns:

Name	Type	Description
str	str	The FIRMS MAP_KEY string.

Raises:

Type	Description
AuthenticationError	When read before :meth:configure has resolved a key.

__init__(credentials)

Store credentials; does not resolve the key yet.

Parameters:

Name	Type	Description	Default
credentials	FirmsCredentials	The :class:FirmsCredentials value object carrying the optional MAP_KEY.	required

Source code in src/earthlens/firms/auth.py

def __init__(self, credentials: FirmsCredentials) -> None:
    """Store credentials; does not resolve the key yet.

    Args:
        credentials: The :class:`FirmsCredentials` value object
            carrying the optional `MAP_KEY`.
    """
    super().__init__(credentials)
    self._configured = False
    self._key: str | None = None

configure()

Resolve the MAP_KEY so subsequent requests can authenticate.

Idempotent — short-circuits when :meth:is_authenticated already returns True. On the first call, resolves the key in this order: the explicit api_key on the credentials, then the FIRMS_MAP_KEY environment variable.

Raises:

Type	Description
AuthenticationError	When neither source supplies a key. The message names the api_key= argument, the FIRMS_MAP_KEY env var, and the free-registration URL — it never blocks on an interactive prompt.

Source code in src/earthlens/firms/auth.py

def configure(self) -> None:
    """Resolve the `MAP_KEY` so subsequent requests can authenticate.

    Idempotent — short-circuits when :meth:`is_authenticated`
    already returns `True`. On the first call, resolves the key in
    this order: the explicit `api_key` on the credentials, then the
    `FIRMS_MAP_KEY` environment variable.

    Raises:
        AuthenticationError: When neither source supplies a key. The
            message names the `api_key=` argument, the
            `FIRMS_MAP_KEY` env var, and the free-registration URL —
            it never blocks on an interactive prompt.
    """
    if self.is_authenticated():
        return
    key = (
        self._creds.api_key.get_secret_value()
        if self._creds.api_key is not None
        else os.environ.get("FIRMS_MAP_KEY")
    )
    if not key:
        raise AuthenticationError(
            "no FIRMS MAP_KEY available: pass api_key= to "
            "EarthLens(...).authenticate() or set the FIRMS_MAP_KEY "
            f"environment variable. Request a free key at {_MAP_KEY_URL}."
        )
    self._key = key
    self._configured = True

is_authenticated()

Return True once :meth:configure has resolved a key.

Cheap predicate — does not call the network. A return of True means a usable key is held by this instance.

Returns:

Name	Type	Description
bool	bool	True after a successful :meth:configure, False before.

Source code in src/earthlens/firms/auth.py

def is_authenticated(self) -> bool:
    """Return `True` once :meth:`configure` has resolved a key.

    Cheap predicate — does not call the network. A return of `True`
    means a usable key is held by this instance.

    Returns:
        bool: `True` after a successful :meth:`configure`, `False`
            before.
    """
    return self._configured

FirmsCredentials

Bases: BaseModel

Frozen value object holding the FIRMS MAP_KEY.

The key is optional at construction time: None means "resolve from the FIRMS_MAP_KEY environment variable at :meth:FirmsAuth.configure time". The real "is there a usable key?" gate is :meth:FirmsAuth.configure, not this model.

Attributes:

Name	Type	Description
api_key	SecretStr | None	The FIRMS MAP_KEY, stored as a :class:pydantic.SecretStr so it is never echoed by repr(creds) or in logs. None defers resolution to the environment variable.

Examples:

• Build from an explicit key; the secret is hidden in repr:

  >>> from earthlens.firms import FirmsCredentials
  >>> creds = FirmsCredentials(api_key="topsecret")
  >>> creds.api_key.get_secret_value()
  'topsecret'
  >>> "topsecret" in repr(creds)
  False
  

• The key is optional — rely on the environment instead:

  >>> from earthlens.firms import FirmsCredentials
  >>> FirmsCredentials().api_key is None
  True
  

Source code in src/earthlens/firms/auth.py

class FirmsCredentials(BaseModel):
    """Frozen value object holding the FIRMS `MAP_KEY`.

    The key is optional at construction time: `None` means "resolve from
    the `FIRMS_MAP_KEY` environment variable at
    :meth:`FirmsAuth.configure` time". The real "is there a usable key?"
    gate is :meth:`FirmsAuth.configure`, not this model.

    Attributes:
        api_key: The FIRMS `MAP_KEY`, stored as a
            :class:`pydantic.SecretStr` so it is never echoed by
            `repr(creds)` or in logs. `None` defers resolution to the
            environment variable.

    Examples:
        - Build from an explicit key; the secret is hidden in `repr`:
            ```python
            >>> from earthlens.firms import FirmsCredentials
            >>> creds = FirmsCredentials(api_key="topsecret")
            >>> creds.api_key.get_secret_value()
            'topsecret'
            >>> "topsecret" in repr(creds)
            False

            ```
        - The key is optional — rely on the environment instead:
            ```python
            >>> from earthlens.firms import FirmsCredentials
            >>> FirmsCredentials().api_key is None
            True

            ```
    """

    model_config = ConfigDict(frozen=True)

    api_key: SecretStr | None = None