Sensor.Community — API reference

Sensor.Community air-quality data source subpackage — earthlens.sensor_community. Background, usage, and licence notes are covered under the other pages in this section; this page is the rendered API.

earthlens.sensor_community

Sensor.Community crowdsourced air-quality backend.

Returns readings from the Sensor.Community low-cost-sensor network as a long-format pandas.DataFrame (one row per measurement), the same tabular shape as earthlens.openaq.

This is a tabular backend: the result is per-row station observations, not a gridded array, so SensorCommunity.OUTPUT_KIND is "tabular" and the earthlens.earthlens.EarthLens facade rejects an aggregate= argument for it.

The archive has one CSV per (sensor, day) but no bbox index, so the backend discovers active sensors in the bbox via the live JSON API, then fetches each discovered sensor's per-day archive CSV over the date range. Historical coverage is therefore limited to sensors currently reporting in the bbox. Readings are crowdsourced from low-cost sensors and licensed under the ODbL; every download() emits a LicenseWarning.

Public surface (re-exported from this package):

• SensorCommunity — the backend; instantiate with a date range, a bbox, and variables=[pollutant, ...], then call SensorCommunity.download.
• Catalog — pydantic-backed loader for the bundled sensor_community_data_catalog.yaml pollutant dispatch table.
• Pollutant — one pollutant's dispatch row (name, column, sensor_types, units, display_name, group).
• LicenseWarning — emitted on every download() to flag the ODbL / low-cost-sensor quality caveat.
• CATALOG_PATH — path to the bundled pollutant YAML; monkey-patchable in tests.

Examples:

• List the registered pollutants:

  >>> from earthlens.sensor_community import Catalog
  >>> sorted(Catalog().pollutants)
  ['humidity', 'pm1', 'pm10', 'pm25', 'pressure', 'temperature']
  

Catalog

Bases: AbstractCatalog

Pollutant catalog for the Sensor.Community backend.

Reads the bundled sensor_community_data_catalog.yaml (shipped as package data) and exposes its pollutants: block as a map of Pollutant rows. Instantiate with no arguments (Catalog()); model_post_init loads and validates the YAML in one pass.

Attributes:

Name	Type	Description
pollutants	dict[str, Pollutant]	Map from the user-facing pollutant name to its Pollutant dispatch row.

Examples:

• Resolve names to the union of serving sensor types and to CSV columns:

  >>> from earthlens.sensor_community import Catalog
  >>> cat = Catalog()
  >>> "sds011" in cat.sensor_types_for(["pm25"])
  True
  >>> cat.columns_for(["pm25", "pm10"])
  {'P2': 'pm25', 'P1': 'pm10'}
  

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

class Catalog(AbstractCatalog):
    """Pollutant catalog for the Sensor.Community backend.

    Reads the bundled `sensor_community_data_catalog.yaml` (shipped as
    package data) and exposes its `pollutants:` block as a map of
    `Pollutant` rows. Instantiate with no arguments (`Catalog()`);
    `model_post_init` loads and validates the YAML in one pass.

    Attributes:
        pollutants: Map from the user-facing pollutant name to its
            `Pollutant` dispatch row.

    Examples:
        - Resolve names to the union of serving sensor types and to CSV
          columns:
            ```python
            >>> from earthlens.sensor_community import Catalog
            >>> cat = Catalog()
            >>> "sds011" in cat.sensor_types_for(["pm25"])
            True
            >>> cat.columns_for(["pm25", "pm10"])
            {'P2': 'pm25', 'P1': 'pm10'}

            ```
    """

    _catalog_kind: str = "Sensor.Community pollutant catalog"
    _entry_noun: str = "pollutants"

    #: The pollutant rows live in the base `datasets` field so the inherited
    #: dict surface works unchanged. `pollutants` is the domain-named alias.
    datasets: dict[str, Pollutant] = Field(default_factory=dict)

    @model_validator(mode="before")
    @classmethod
    def _accept_pollutants_alias(cls, data: Any) -> Any:
        """Accept the `pollutants=` kwarg as an alias for `datasets`.

        Callers and tests construct `Catalog(pollutants={...})`. The rows
        live in the base `datasets` field, so rewrite that key on the way
        in. An explicit `datasets=` always wins.

        Args:
            data: The raw model input (a mapping when constructed with
                keyword arguments).

        Returns:
            The input with `pollutants` renamed to `datasets`, untouched
            otherwise.
        """
        if isinstance(data, dict) and "pollutants" in data and "datasets" not in data:
            data = dict(data)
            data["datasets"] = data.pop("pollutants")
        return data

    @property
    def pollutants(self) -> dict[str, Pollutant]:
        """The pollutant map — alias for the base `datasets` field.

        Returns:
            dict[str, Pollutant]: The same mapping stored in `datasets`.
        """
        return self.datasets

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

        Returns:
            dict[str, Any]: The `datasets` read from
                the bundled catalog.
        """
        return {"datasets": Catalog.load().datasets}

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

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

        Returns:
            A fully-populated `Catalog`.

        Raises:
            ValueError: If `catalog_path` does not exist, or if the file has no
                `pollutants:` block, or a row fails `Pollutant` validation.
        """
        catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
        cached = load_catalog(
            catalog_path,
            _CATALOG_CACHE,
            _parse_sensor_community_catalog,
            provider="Sensor.Community",
        )
        return cls(datasets=dict(cached))

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

        Returns:
            dict[str, Pollutant]: Same object as `datasets` / `pollutants`.
        """
        return self.datasets

    def get_pollutant(self, name: str) -> Pollutant:
        """Resolve a pollutant name to its `Pollutant` row.

        Thin wrapper over the inherited `get_dataset`, which raises a
        `ValueError` with a did-you-mean hint on an unknown name.

        Args:
            name: A user-facing pollutant name (`"pm25"`, `"temperature"`).

        Returns:
            Pollutant: The matching dispatch row.

        Raises:
            ValueError: If `name` is not a known pollutant.
        """
        return cast("Pollutant", self.get_dataset(name))

    def sensor_types_for(self, names: list[str]) -> set[str]:
        """Return the union of serving sensor-type slugs for `names`.

        Args:
            names: User-facing pollutant names to resolve.

        Returns:
            set[str]: Every archive sensor-type slug whose CSV carries at
                least one of the requested pollutants.

        Raises:
            ValueError: If any name is unknown (via `get_pollutant`).
        """
        types: set[str] = set()
        for name in names:
            types.update(self.get_pollutant(name).sensor_types)
        return types

    def columns_for(self, names: list[str]) -> dict[str, str]:
        """Return the CSV `column` -> pollutant name map for `names`.

        Used at parse time to pull every requested pollutant's value out
        of one sensor CSV in a single pass.

        Args:
            names: User-facing pollutant names to resolve.

        Returns:
            dict[str, str]: Each requested pollutant's CSV column mapped
                to its name (`{"P2": "pm25", "P1": "pm10"}`).

        Raises:
            ValueError: If any name is unknown (via `get_pollutant`).
        """
        return {self.get_pollutant(name).column: name for name in names}

pollutants property

The pollutant map — alias for the base datasets field.

Returns:

Type	Description
dict[str, Pollutant]	dict[str, Pollutant]: The same mapping stored in datasets.

columns_for(names)

Return the CSV column -> pollutant name map for names.

Used at parse time to pull every requested pollutant's value out of one sensor CSV in a single pass.

Parameters:

Name	Type	Description	Default
names	list[str]	User-facing pollutant names to resolve.	required

Returns:

Type	Description
dict[str, str]	dict[str, str]: Each requested pollutant's CSV column mapped to its name ({"P2": "pm25", "P1": "pm10"}).

Raises:

Type	Description
ValueError	If any name is unknown (via get_pollutant).

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

def columns_for(self, names: list[str]) -> dict[str, str]:
    """Return the CSV `column` -> pollutant name map for `names`.

    Used at parse time to pull every requested pollutant's value out
    of one sensor CSV in a single pass.

    Args:
        names: User-facing pollutant names to resolve.

    Returns:
        dict[str, str]: Each requested pollutant's CSV column mapped
            to its name (`{"P2": "pm25", "P1": "pm10"}`).

    Raises:
        ValueError: If any name is unknown (via `get_pollutant`).
    """
    return {self.get_pollutant(name).column: name for name in names}

get_catalog()

Return the pollutant map (satisfies the abstract contract).

Returns:

Type	Description
dict[str, Pollutant]	dict[str, Pollutant]: Same object as datasets / pollutants.

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

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

    Returns:
        dict[str, Pollutant]: Same object as `datasets` / `pollutants`.
    """
    return self.datasets

get_pollutant(name)

Resolve a pollutant name to its Pollutant row.

Thin wrapper over the inherited get_dataset, which raises a ValueError with a did-you-mean hint on an unknown name.

Parameters:

Name	Type	Description	Default
name	str	A user-facing pollutant name ("pm25", "temperature").	required

Returns:

Name	Type	Description
Pollutant	Pollutant	The matching dispatch row.

Raises:

Type	Description
ValueError	If name is not a known pollutant.

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

def get_pollutant(self, name: str) -> Pollutant:
    """Resolve a pollutant name to its `Pollutant` row.

    Thin wrapper over the inherited `get_dataset`, which raises a
    `ValueError` with a did-you-mean hint on an unknown name.

    Args:
        name: A user-facing pollutant name (`"pm25"`, `"temperature"`).

    Returns:
        Pollutant: The matching dispatch row.

    Raises:
        ValueError: If `name` is not a known pollutant.
    """
    return cast("Pollutant", self.get_dataset(name))

load(catalog_path=None) classmethod

Read the Sensor.Community pollutant catalog from disk.

Parameters:

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

Returns:

Type	Description
Catalog	A fully-populated Catalog.

Raises:

Type	Description
ValueError	If catalog_path does not exist, or if the file has no pollutants: block, or a row fails Pollutant validation.

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

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

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

    Returns:
        A fully-populated `Catalog`.

    Raises:
        ValueError: If `catalog_path` does not exist, or if the file has no
            `pollutants:` block, or a row fails `Pollutant` validation.
    """
    catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
    cached = load_catalog(
        catalog_path,
        _CATALOG_CACHE,
        _parse_sensor_community_catalog,
        provider="Sensor.Community",
    )
    return cls(datasets=dict(cached))

sensor_types_for(names)

Return the union of serving sensor-type slugs for names.

Parameters:

Name	Type	Description	Default
names	list[str]	User-facing pollutant names to resolve.	required

Returns:

Type	Description
set[str]	set[str]: Every archive sensor-type slug whose CSV carries at least one of the requested pollutants.

Raises:

Type	Description
ValueError	If any name is unknown (via get_pollutant).

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

def sensor_types_for(self, names: list[str]) -> set[str]:
    """Return the union of serving sensor-type slugs for `names`.

    Args:
        names: User-facing pollutant names to resolve.

    Returns:
        set[str]: Every archive sensor-type slug whose CSV carries at
            least one of the requested pollutants.

    Raises:
        ValueError: If any name is unknown (via `get_pollutant`).
    """
    types: set[str] = set()
    for name in names:
        types.update(self.get_pollutant(name).sensor_types)
    return types

LicenseWarning

Bases: UserWarning

Warns that Sensor.Community data carries ODbL / quality obligations.

Sensor.Community measurements are crowdsourced from low-cost sensors and licensed under the Open Database License (ODbL): redistribution must keep the attribution and share-alike terms, and the readings are not reference-grade. The backend emits this once per download() so a downstream user is told rather than discovering it silently.

Source code in libs/providers/atmosphere/src/earthlens/sensor_community/_helpers.py

class LicenseWarning(UserWarning):
    """Warns that Sensor.Community data carries ODbL / quality obligations.

    Sensor.Community measurements are crowdsourced from low-cost sensors
    and licensed under the Open Database License (ODbL): redistribution
    must keep the attribution and share-alike terms, and the readings are
    not reference-grade. The backend emits this once per `download()` so a
    downstream user is told rather than discovering it silently.
    """

Pollutant

Bases: BaseModel

One Sensor.Community pollutant's dispatch row.

The user-facing name is the parent key in Catalog.pollutants and is also stored on the row as name so a resolved Pollutant is self-describing.

Attributes:

Name	Type	Description
name	str	Short machine name ("pm25", "temperature"); matches the catalog key.
column	str	The CSV column this pollutant is read from ("P2" for PM2.5, "temperature").
sensor_types	list[str]	Archive sensor-type slugs whose per-sensor CSV carries column (["sds011", "pms5003", ...]). Used to filter discovery and choose which archive files to fetch.
units	str	The reporting unit ("µg/m³", "°C"). Sensor.Community reports pressure in pascals.
display_name	str	Human-readable label for docs / plots ("PM2.5").
group	PollutantGroup	Coarse classification — "particulate", "meteorological", or "other".

Examples:

• Build a row directly:

  >>> from earthlens.sensor_community import Pollutant
  >>> p = Pollutant(name="pm25", column="P2", sensor_types=["sds011"])
  >>> (p.column, p.sensor_types)
  ('P2', ['sds011'])
  

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

class Pollutant(BaseModel):
    """One Sensor.Community pollutant's dispatch row.

    The user-facing name is the parent key in `Catalog.pollutants` and is
    also stored on the row as `name` so a resolved `Pollutant` is
    self-describing.

    Attributes:
        name: Short machine name (`"pm25"`, `"temperature"`); matches the
            catalog key.
        column: The CSV column this pollutant is read from (`"P2"` for
            PM2.5, `"temperature"`).
        sensor_types: Archive sensor-type slugs whose per-sensor CSV
            carries `column` (`["sds011", "pms5003", ...]`). Used to
            filter discovery and choose which archive files to fetch.
        units: The reporting unit (`"µg/m³"`, `"°C"`). Sensor.Community
            reports pressure in pascals.
        display_name: Human-readable label for docs / plots (`"PM2.5"`).
        group: Coarse classification — `"particulate"`, `"meteorological"`,
            or `"other"`.

    Examples:
        - Build a row directly:
            ```python
            >>> from earthlens.sensor_community import Pollutant
            >>> p = Pollutant(name="pm25", column="P2", sensor_types=["sds011"])
            >>> (p.column, p.sensor_types)
            ('P2', ['sds011'])

            ```
    """

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

    name: str
    column: str
    sensor_types: list[str] = Field(min_length=1)
    units: str = ""
    display_name: str = ""
    group: PollutantGroup = "other"

SensorCommunity

Bases: AbstractDataSource

Sensor.Community air-quality backend (long-format tabular output).

Discovers active sensors in the request bbox via the live JSON API, then fetches each sensor's per-day archive CSV over the date window, returning a long-format pandas.DataFrame (one row per measurement). There is no authentication — both hosts are public.

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	"tabular" — the result is per-row station observations, so the facade rejects aggregate= with NotImplementedError.

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

class SensorCommunity(AbstractDataSource):
    """Sensor.Community air-quality backend (long-format tabular output).

    Discovers active sensors in the request bbox via the live JSON API,
    then fetches each sensor's per-day archive CSV over the date window,
    returning a long-format `pandas.DataFrame` (one row per measurement).
    There is no authentication — both hosts are public.

    Attributes:
        OUTPUT_KIND: `"tabular"` — the result is per-row station
            observations, so the facade rejects `aggregate=` with
            `NotImplementedError`.
    """

    OUTPUT_KIND: OutputKind = "tabular"

    AGGREGATE_REFUSAL_REASON = "readings are tabular per-row station data, not gridded rasters, so there is no meaningful gridded reduction"

    def __init__(
        self,
        start: str,
        end: str,
        variables: list[str],
        lat_lim: list[float],
        lon_lim: list[float],
        temporal_resolution: str = "raw",
        path: Path | str | None = None,
        fmt: str = "%Y-%m-%d",
        session: requests.Session | None = None,
        client: SensorCommunityClient | None = None,
        file_format: FileFormat = "csv",
    ):
        """Initialise a Sensor.Community backend instance.

        Args:
            start: Inclusive start of the observation window, as a string
                parsed with `fmt`.
            end: Inclusive end of the observation window.
            variables: List of pollutant names to fetch (`["pm25"]`,
                `["temperature", "humidity"]`). Mapped to CSV columns +
                serving sensor types via the catalog. An empty list
                defaults to `["pm25"]`.
            lat_lim: `[lat_min, lat_max]` bounding-box latitudes in
                degrees.
            lon_lim: `[lon_min, lon_max]` bounding-box longitudes in
                degrees.
            temporal_resolution: Recorded for provenance; Sensor.Community
                has no server-side rollup. Accepts `"raw"` (default),
                `"hourly"`, or `"daily"` (the facade default).
            path: Output directory for the written CSV / Parquet. Created
                by the parent class if absent.
            fmt: `strptime` format for `start` / `end`.
            session: An existing `requests.Session` to reuse (used to
                build the default client). Injectable for tests.
            client: A `SensorCommunityClient` to reuse. Injectable so
                tests supply a fake transport; when `None` (default) one
                is built lazily from `session`.
            file_format: Output format — `"csv"` (default) or `"parquet"`.
        """
        if isinstance(variables, dict):
            raise TypeError(
                "SensorCommunity `variables` must be a list of pollutant names "
                "(e.g. ['pm25', 'pm10']), not a mapping."
            )
        self._session = session
        self._client_obj = client
        self._file_format: FileFormat = file_format
        self._catalog = Catalog()
        self._columns: dict[str, str] | None = None
        self._units: dict[str, str] | None = None
        super().__init__(
            start=start,
            end=end,
            variables=list(variables) or list(_DEFAULT_PARAMETERS),
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim,
            lon_lim=lon_lim,
            fmt=fmt,
            path=path,
        )

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

        Args:
            start: Inclusive start date string.
            end: Inclusive end date string.
            temporal_resolution: Provenance label (`"raw"`, `"hourly"`, or
                `"daily"`); does not change the request.
            fmt: `strptime` format tried first for a string `start` /
                `end`; a non-matching string falls back to an ISO-8601
                parse, and a `datetime` / `date` ignores it.

        Returns:
            TemporalExtent: Frozen model with the parsed endpoints.

        Raises:
            ValueError: If `temporal_resolution` is not accepted, or
                `start` parses to a date later than `end`.
        """
        if temporal_resolution not in _ACCEPTED_RESOLUTIONS:
            raise ValueError(
                f"temporal_resolution must be one of "
                f"{sorted(_ACCEPTED_RESOLUTIONS)}, got {temporal_resolution!r}."
            )
        start_dt = to_datetime(start, fmt)
        end_dt = to_datetime(end, fmt)
        return TemporalExtent(
            start_date=start_dt,
            end_date=end_dt,
            resolution=temporal_resolution,
            dates=pd.DatetimeIndex([start_dt, end_dt]),
        )

    def _client(self) -> SensorCommunityClient:
        """Build (once) and return the injectable HTTP client.

        Returns:
            SensorCommunityClient: The cached client (injected, or built
                from `session`).
        """
        if self._client_obj is None:
            self._client_obj = SensorCommunityClient(session=self._session)
        return self._client_obj

    def _days(self) -> list[str]:
        """Return the `YYYY-MM-DD` archive days spanning the request window.

        Returns:
            list[str]: One date string per day from `start_date` to
                `end_date`, inclusive.
        """
        start = self.time.start_date.date()
        end = self.time.end_date.date()
        span = (end - start).days
        return [
            (start + dt.timedelta(days=offset)).isoformat()
            for offset in range(span + 1)
        ]

    def _column_map(self) -> dict[str, str]:
        """CSV column -> pollutant name for the requested pollutants (cached)."""
        if self._columns is None:
            self._columns = self._catalog.columns_for(cast("list[str]", self.vars))
        return self._columns

    def _unit_map(self) -> dict[str, str]:
        """Pollutant name -> reporting unit for the requested pollutants (cached)."""
        if self._units is None:
            self._units = {
                name: self._catalog.get_pollutant(name).units for name in self.vars
            }
        return self._units

    def _search(self) -> list[RemoteProduct]:
        """Discover active sensors in the bbox via the live JSON API.

        Returns:
            list[RemoteProduct]: One product per unique `(sensor_id,
                sensor_type)` active in the bbox whose type serves a
                requested pollutant; `id` is the sensor id and `metadata`
                carries `sensor_type` / `lat` / `lon`.
        """
        wanted = self._catalog.sensor_types_for(cast("list[str]", self.vars))
        snapshot = self._client().live_snapshot()
        sensors = sensors_in_bbox(
            snapshot,
            (self.space.south, self.space.north),
            (self.space.west, self.space.east),
            wanted,
        )
        if not sensors:
            logger.warning(
                "Sensor.Community search: no live sensor of the requested "
                "type(s) is currently reporting in the bbox; historical "
                "coverage is limited to sensors active now."
            )
        return [
            RemoteProduct(id=sensor["sensor_id"], metadata=sensor) for sensor in sensors
        ]

    def _fetch_one(self, product: RemoteProduct) -> pd.DataFrame:
        """Fetch one sensor's per-day archive CSVs over the date window.

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

        Returns:
            pd.DataFrame: The sensor's readings in the long schema (empty
                when it reported nothing across the window).
        """
        columns = self._column_map()
        units = self._unit_map()
        sensor_type = product.metadata["sensor_type"]
        frames: list[pd.DataFrame] = []
        for day in self._days():
            text = self._client().archive_csv(day, sensor_type, product.id)
            if text is None:
                logger.info(
                    f"Sensor.Community: no archive file for sensor "
                    f"{product.id} ({sensor_type}) on {day}; skipped."
                )
                continue
            frames.append(
                frame_from_csv(text, columns, units, default_sensor_type=sensor_type)
            )
        non_empty = [frame for frame in frames if not frame.empty]
        if not non_empty:
            return empty_frame()
        combined = pd.concat(non_empty, ignore_index=True)
        # `concat` copied every frame, so the per-sensor sources are dead
        # weight from here. Both lists have to be cleared — either alone frees
        # nothing, since each holds the same frames. This does not lower the
        # peak (reached inside the concat) but stops them being carried through
        # the window filter and the return.
        frames.clear()
        non_empty.clear()
        return combined

    def _api(self) -> list[pd.DataFrame]:
        """Compose `_search` and `_fetch_one` into the canonical C3 shape.

        Contract-only: `download` overrides the write path and calls
        `_search_fetch_each` directly (to concat + window itself), mirroring
        the `earthlens.openaq` sibling, so this is not on the live path.
        """
        return self._search_fetch_each(desc="Sensor.Community sensors", unit="sensor")

    def _window(self) -> tuple[pd.Timestamp, pd.Timestamp]:
        """Return the `[lower, upper)` UTC filter bounds for the request.

        A date-granular end (midnight) is extended by one day so the whole
        end day is inclusive — the common path, and identical to AirNow /
        EEA. A non-midnight `end` (only reachable via an hour-aware `fmt`)
        yields a **half-open** `[start, end)` window here; AirNow instead
        treats its end hour as inclusive (its API takes an hourly range), so
        hour-granular callers should account for that one-endpoint
        difference.

        Returns:
            tuple[pd.Timestamp, pd.Timestamp]: `(lower, upper)`, tz-aware
                UTC; `upper` is exclusive.
        """
        lower = pd.Timestamp(self.time.start_date, tz="UTC")
        end = self.time.end_date
        if end.hour == 0 and end.minute == 0:
            upper = pd.Timestamp(end, tz="UTC") + pd.Timedelta(days=1)
        else:
            upper = pd.Timestamp(end, tz="UTC")
        return lower, upper

    def download(
        self,
        progress_bar: bool = True,
        limit: int | None = None,
    ) -> pd.DataFrame:
        """Discover + fetch readings, write them to `path`, return the frame.

        Emits a `LicenseWarning` (ODbL), runs the live-API discovery then
        the per-sensor archive fetch under a `tqdm` bar, concatenates and
        windows the readings to the exact date range, writes the
        long-format result to `path` as CSV (or Parquet), and returns it.
        An empty result returns — and writes — a schema-only DataFrame.

        Args:
            progress_bar: Show the per-sensor `tqdm` bar. Defaults to
                `True`.
            limit: Cap on the total readings fetched, across every discovered
                sensor. Applied as each sensor's frame arrives, so a sensor
                past the cap never has its daily archive files downloaded.
                `None` (the default) fetches everything. The cap is on rows
                *fetched*, before the window filter, so the returned frame can
                be shorter than the cap.

        Returns:
            pd.DataFrame: The long-format readings (schema columns,
                `datetime_utc` tz-aware UTC). Empty (schema-only) when
                nothing matched.
        """
        self._limit = self.check_limit(limit)
        warnings.warn(_LICENSE_TEXT, LicenseWarning, stacklevel=2)

        frames = self._search_fetch_each(
            progress_bar=progress_bar, desc="Sensor.Community sensors", unit="sensor"
        )
        non_empty = [frame for frame in frames if not frame.empty]
        # Release the per-sensor frames as we go: `concat` copies, so holding
        # the sources alongside the combined frame — and then alongside the
        # windowed copy — keeps up to three full copies of the request in RAM.
        frames.clear()
        if non_empty:
            combined = pd.concat(non_empty, ignore_index=True)
            non_empty.clear()
            lower, upper = self._window()
            mask = (combined["datetime_utc"] >= lower) & (
                combined["datetime_utc"] < upper
            )
            df = combined[mask].reset_index(drop=True)
            del combined
        else:
            df = empty_frame()

        out_path = self._output_path()
        if self._file_format == "parquet":
            df.to_parquet(out_path, index=False)
        else:
            df.to_csv(out_path, index=False)

        if len(df):
            logger.info(
                f"Sensor.Community download summary: {len(df)} reading(s) "
                f"across {df['station_id'].nunique()} sensor(s) written to "
                f"{out_path}"
            )
        else:
            logger.warning(
                "Sensor.Community download summary: no readings matched the "
                f"request; wrote an empty (schema-only) frame to {out_path}"
            )
        return df

    def _output_path(self) -> Path:
        """Compose the per-request output file path under `root_dir`."""
        ext = "parquet" if self._file_format == "parquet" else "csv"
        params = "-".join(self.vars)
        start = self.time.start_date.strftime("%Y%m%d")
        end = self.time.end_date.strftime("%Y%m%d")
        return self.root_dir / f"sensor_community_{params}_{start}_{end}.{ext}"

__init__(start, end, variables, lat_lim, lon_lim, temporal_resolution='raw', path=None, fmt='%Y-%m-%d', session=None, client=None, file_format='csv')

Initialise a Sensor.Community backend instance.

Parameters:

Name	Type	Description	Default
start	str	Inclusive start of the observation window, as a string parsed with fmt.	required
end	str	Inclusive end of the observation window.	required
variables	list[str]	List of pollutant names to fetch (["pm25"], ["temperature", "humidity"]). Mapped to CSV columns + serving sensor types via the catalog. An empty list defaults to ["pm25"].	required
lat_lim	list[float]	[lat_min, lat_max] bounding-box latitudes in degrees.	required
lon_lim	list[float]	[lon_min, lon_max] bounding-box longitudes in degrees.	required
temporal_resolution	str	Recorded for provenance; Sensor.Community has no server-side rollup. Accepts "raw" (default), "hourly", or "daily" (the facade default).	'raw'
path	Path | str | None	Output directory for the written CSV / Parquet. Created by the parent class if absent.	None
fmt	str	strptime format for start / end.	'%Y-%m-%d'
session	Session | None	An existing requests.Session to reuse (used to build the default client). Injectable for tests.	None
client	SensorCommunityClient | None	A SensorCommunityClient to reuse. Injectable so tests supply a fake transport; when None (default) one is built lazily from session.	None
file_format	FileFormat	Output format — "csv" (default) or "parquet".	'csv'

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

def __init__(
    self,
    start: str,
    end: str,
    variables: list[str],
    lat_lim: list[float],
    lon_lim: list[float],
    temporal_resolution: str = "raw",
    path: Path | str | None = None,
    fmt: str = "%Y-%m-%d",
    session: requests.Session | None = None,
    client: SensorCommunityClient | None = None,
    file_format: FileFormat = "csv",
):
    """Initialise a Sensor.Community backend instance.

    Args:
        start: Inclusive start of the observation window, as a string
            parsed with `fmt`.
        end: Inclusive end of the observation window.
        variables: List of pollutant names to fetch (`["pm25"]`,
            `["temperature", "humidity"]`). Mapped to CSV columns +
            serving sensor types via the catalog. An empty list
            defaults to `["pm25"]`.
        lat_lim: `[lat_min, lat_max]` bounding-box latitudes in
            degrees.
        lon_lim: `[lon_min, lon_max]` bounding-box longitudes in
            degrees.
        temporal_resolution: Recorded for provenance; Sensor.Community
            has no server-side rollup. Accepts `"raw"` (default),
            `"hourly"`, or `"daily"` (the facade default).
        path: Output directory for the written CSV / Parquet. Created
            by the parent class if absent.
        fmt: `strptime` format for `start` / `end`.
        session: An existing `requests.Session` to reuse (used to
            build the default client). Injectable for tests.
        client: A `SensorCommunityClient` to reuse. Injectable so
            tests supply a fake transport; when `None` (default) one
            is built lazily from `session`.
        file_format: Output format — `"csv"` (default) or `"parquet"`.
    """
    if isinstance(variables, dict):
        raise TypeError(
            "SensorCommunity `variables` must be a list of pollutant names "
            "(e.g. ['pm25', 'pm10']), not a mapping."
        )
    self._session = session
    self._client_obj = client
    self._file_format: FileFormat = file_format
    self._catalog = Catalog()
    self._columns: dict[str, str] | None = None
    self._units: dict[str, str] | None = None
    super().__init__(
        start=start,
        end=end,
        variables=list(variables) or list(_DEFAULT_PARAMETERS),
        temporal_resolution=temporal_resolution,
        lat_lim=lat_lim,
        lon_lim=lon_lim,
        fmt=fmt,
        path=path,
    )

download(progress_bar=True, limit=None)

Discover + fetch readings, write them to path, return the frame.

Emits a LicenseWarning (ODbL), runs the live-API discovery then the per-sensor archive fetch under a tqdm bar, concatenates and windows the readings to the exact date range, writes the long-format result to path as CSV (or Parquet), and returns it. An empty result returns — and writes — a schema-only DataFrame.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Show the per-sensor tqdm bar. Defaults to True.	True
limit	int | None	Cap on the total readings fetched, across every discovered sensor. Applied as each sensor's frame arrives, so a sensor past the cap never has its daily archive files downloaded. None (the default) fetches everything. The cap is on rows fetched, before the window filter, so the returned frame can be shorter than the cap.	None

Returns:

Type	Description
DataFrame	pd.DataFrame: The long-format readings (schema columns, datetime_utc tz-aware UTC). Empty (schema-only) when nothing matched.

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

def download(
    self,
    progress_bar: bool = True,
    limit: int | None = None,
) -> pd.DataFrame:
    """Discover + fetch readings, write them to `path`, return the frame.

    Emits a `LicenseWarning` (ODbL), runs the live-API discovery then
    the per-sensor archive fetch under a `tqdm` bar, concatenates and
    windows the readings to the exact date range, writes the
    long-format result to `path` as CSV (or Parquet), and returns it.
    An empty result returns — and writes — a schema-only DataFrame.

    Args:
        progress_bar: Show the per-sensor `tqdm` bar. Defaults to
            `True`.
        limit: Cap on the total readings fetched, across every discovered
            sensor. Applied as each sensor's frame arrives, so a sensor
            past the cap never has its daily archive files downloaded.
            `None` (the default) fetches everything. The cap is on rows
            *fetched*, before the window filter, so the returned frame can
            be shorter than the cap.

    Returns:
        pd.DataFrame: The long-format readings (schema columns,
            `datetime_utc` tz-aware UTC). Empty (schema-only) when
            nothing matched.
    """
    self._limit = self.check_limit(limit)
    warnings.warn(_LICENSE_TEXT, LicenseWarning, stacklevel=2)

    frames = self._search_fetch_each(
        progress_bar=progress_bar, desc="Sensor.Community sensors", unit="sensor"
    )
    non_empty = [frame for frame in frames if not frame.empty]
    # Release the per-sensor frames as we go: `concat` copies, so holding
    # the sources alongside the combined frame — and then alongside the
    # windowed copy — keeps up to three full copies of the request in RAM.
    frames.clear()
    if non_empty:
        combined = pd.concat(non_empty, ignore_index=True)
        non_empty.clear()
        lower, upper = self._window()
        mask = (combined["datetime_utc"] >= lower) & (
            combined["datetime_utc"] < upper
        )
        df = combined[mask].reset_index(drop=True)
        del combined
    else:
        df = empty_frame()

    out_path = self._output_path()
    if self._file_format == "parquet":
        df.to_parquet(out_path, index=False)
    else:
        df.to_csv(out_path, index=False)

    if len(df):
        logger.info(
            f"Sensor.Community download summary: {len(df)} reading(s) "
            f"across {df['station_id'].nunique()} sensor(s) written to "
            f"{out_path}"
        )
    else:
        logger.warning(
            "Sensor.Community download summary: no readings matched the "
            f"request; wrote an empty (schema-only) frame to {out_path}"
        )
    return df

earthlens.sensor_community.backend

Backend that fetches crowdsourced air-quality data from Sensor.Community.

SensorCommunity(AbstractDataSource) returns readings from the Sensor.Community low-cost-sensor network as a long-format pandas.DataFrame (one row per measurement), the same tabular shape as earthlens.openaq.

This is a tabular backend: the result is per-row station observations, not a gridded array, so OUTPUT_KIND = "tabular" and the earthlens.earthlens.EarthLens facade rejects an aggregate= argument.

Transport (a search/fetch split, like OpenAQ). The archive has one CSV per (sensor, day) but no bbox index, so _search first hits the live JSON API (data.sensor.community) to discover which sensors are active in the request bbox; _fetch then pulls each discovered sensor's per-day archive CSV (archive.sensor.community) over the date range, ;-parses it, and extracts the requested pollutant columns. A missing daily file is logged and skipped (never a silent gap). Because discovery uses the live snapshot, historical coverage is limited to sensors currently reporting in the bbox.

Data quality: readings are crowdsourced from low-cost sensors and licensed under the ODbL; every download() emits a LicenseWarning.

Pollutant selection: variables is a list[str] of pollutant names (["pm25"], ["pm25", "pm10"], ["temperature", "humidity"]), mapped to CSV columns + serving sensor types via the bundled catalog.

SensorCommunity

Bases: AbstractDataSource

Sensor.Community air-quality backend (long-format tabular output).

Discovers active sensors in the request bbox via the live JSON API, then fetches each sensor's per-day archive CSV over the date window, returning a long-format pandas.DataFrame (one row per measurement). There is no authentication — both hosts are public.

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	"tabular" — the result is per-row station observations, so the facade rejects aggregate= with NotImplementedError.

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

class SensorCommunity(AbstractDataSource):
    """Sensor.Community air-quality backend (long-format tabular output).

    Discovers active sensors in the request bbox via the live JSON API,
    then fetches each sensor's per-day archive CSV over the date window,
    returning a long-format `pandas.DataFrame` (one row per measurement).
    There is no authentication — both hosts are public.

    Attributes:
        OUTPUT_KIND: `"tabular"` — the result is per-row station
            observations, so the facade rejects `aggregate=` with
            `NotImplementedError`.
    """

    OUTPUT_KIND: OutputKind = "tabular"

    AGGREGATE_REFUSAL_REASON = "readings are tabular per-row station data, not gridded rasters, so there is no meaningful gridded reduction"

    def __init__(
        self,
        start: str,
        end: str,
        variables: list[str],
        lat_lim: list[float],
        lon_lim: list[float],
        temporal_resolution: str = "raw",
        path: Path | str | None = None,
        fmt: str = "%Y-%m-%d",
        session: requests.Session | None = None,
        client: SensorCommunityClient | None = None,
        file_format: FileFormat = "csv",
    ):
        """Initialise a Sensor.Community backend instance.

        Args:
            start: Inclusive start of the observation window, as a string
                parsed with `fmt`.
            end: Inclusive end of the observation window.
            variables: List of pollutant names to fetch (`["pm25"]`,
                `["temperature", "humidity"]`). Mapped to CSV columns +
                serving sensor types via the catalog. An empty list
                defaults to `["pm25"]`.
            lat_lim: `[lat_min, lat_max]` bounding-box latitudes in
                degrees.
            lon_lim: `[lon_min, lon_max]` bounding-box longitudes in
                degrees.
            temporal_resolution: Recorded for provenance; Sensor.Community
                has no server-side rollup. Accepts `"raw"` (default),
                `"hourly"`, or `"daily"` (the facade default).
            path: Output directory for the written CSV / Parquet. Created
                by the parent class if absent.
            fmt: `strptime` format for `start` / `end`.
            session: An existing `requests.Session` to reuse (used to
                build the default client). Injectable for tests.
            client: A `SensorCommunityClient` to reuse. Injectable so
                tests supply a fake transport; when `None` (default) one
                is built lazily from `session`.
            file_format: Output format — `"csv"` (default) or `"parquet"`.
        """
        if isinstance(variables, dict):
            raise TypeError(
                "SensorCommunity `variables` must be a list of pollutant names "
                "(e.g. ['pm25', 'pm10']), not a mapping."
            )
        self._session = session
        self._client_obj = client
        self._file_format: FileFormat = file_format
        self._catalog = Catalog()
        self._columns: dict[str, str] | None = None
        self._units: dict[str, str] | None = None
        super().__init__(
            start=start,
            end=end,
            variables=list(variables) or list(_DEFAULT_PARAMETERS),
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim,
            lon_lim=lon_lim,
            fmt=fmt,
            path=path,
        )

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

        Args:
            start: Inclusive start date string.
            end: Inclusive end date string.
            temporal_resolution: Provenance label (`"raw"`, `"hourly"`, or
                `"daily"`); does not change the request.
            fmt: `strptime` format tried first for a string `start` /
                `end`; a non-matching string falls back to an ISO-8601
                parse, and a `datetime` / `date` ignores it.

        Returns:
            TemporalExtent: Frozen model with the parsed endpoints.

        Raises:
            ValueError: If `temporal_resolution` is not accepted, or
                `start` parses to a date later than `end`.
        """
        if temporal_resolution not in _ACCEPTED_RESOLUTIONS:
            raise ValueError(
                f"temporal_resolution must be one of "
                f"{sorted(_ACCEPTED_RESOLUTIONS)}, got {temporal_resolution!r}."
            )
        start_dt = to_datetime(start, fmt)
        end_dt = to_datetime(end, fmt)
        return TemporalExtent(
            start_date=start_dt,
            end_date=end_dt,
            resolution=temporal_resolution,
            dates=pd.DatetimeIndex([start_dt, end_dt]),
        )

    def _client(self) -> SensorCommunityClient:
        """Build (once) and return the injectable HTTP client.

        Returns:
            SensorCommunityClient: The cached client (injected, or built
                from `session`).
        """
        if self._client_obj is None:
            self._client_obj = SensorCommunityClient(session=self._session)
        return self._client_obj

    def _days(self) -> list[str]:
        """Return the `YYYY-MM-DD` archive days spanning the request window.

        Returns:
            list[str]: One date string per day from `start_date` to
                `end_date`, inclusive.
        """
        start = self.time.start_date.date()
        end = self.time.end_date.date()
        span = (end - start).days
        return [
            (start + dt.timedelta(days=offset)).isoformat()
            for offset in range(span + 1)
        ]

    def _column_map(self) -> dict[str, str]:
        """CSV column -> pollutant name for the requested pollutants (cached)."""
        if self._columns is None:
            self._columns = self._catalog.columns_for(cast("list[str]", self.vars))
        return self._columns

    def _unit_map(self) -> dict[str, str]:
        """Pollutant name -> reporting unit for the requested pollutants (cached)."""
        if self._units is None:
            self._units = {
                name: self._catalog.get_pollutant(name).units for name in self.vars
            }
        return self._units

    def _search(self) -> list[RemoteProduct]:
        """Discover active sensors in the bbox via the live JSON API.

        Returns:
            list[RemoteProduct]: One product per unique `(sensor_id,
                sensor_type)` active in the bbox whose type serves a
                requested pollutant; `id` is the sensor id and `metadata`
                carries `sensor_type` / `lat` / `lon`.
        """
        wanted = self._catalog.sensor_types_for(cast("list[str]", self.vars))
        snapshot = self._client().live_snapshot()
        sensors = sensors_in_bbox(
            snapshot,
            (self.space.south, self.space.north),
            (self.space.west, self.space.east),
            wanted,
        )
        if not sensors:
            logger.warning(
                "Sensor.Community search: no live sensor of the requested "
                "type(s) is currently reporting in the bbox; historical "
                "coverage is limited to sensors active now."
            )
        return [
            RemoteProduct(id=sensor["sensor_id"], metadata=sensor) for sensor in sensors
        ]

    def _fetch_one(self, product: RemoteProduct) -> pd.DataFrame:
        """Fetch one sensor's per-day archive CSVs over the date window.

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

        Returns:
            pd.DataFrame: The sensor's readings in the long schema (empty
                when it reported nothing across the window).
        """
        columns = self._column_map()
        units = self._unit_map()
        sensor_type = product.metadata["sensor_type"]
        frames: list[pd.DataFrame] = []
        for day in self._days():
            text = self._client().archive_csv(day, sensor_type, product.id)
            if text is None:
                logger.info(
                    f"Sensor.Community: no archive file for sensor "
                    f"{product.id} ({sensor_type}) on {day}; skipped."
                )
                continue
            frames.append(
                frame_from_csv(text, columns, units, default_sensor_type=sensor_type)
            )
        non_empty = [frame for frame in frames if not frame.empty]
        if not non_empty:
            return empty_frame()
        combined = pd.concat(non_empty, ignore_index=True)
        # `concat` copied every frame, so the per-sensor sources are dead
        # weight from here. Both lists have to be cleared — either alone frees
        # nothing, since each holds the same frames. This does not lower the
        # peak (reached inside the concat) but stops them being carried through
        # the window filter and the return.
        frames.clear()
        non_empty.clear()
        return combined

    def _api(self) -> list[pd.DataFrame]:
        """Compose `_search` and `_fetch_one` into the canonical C3 shape.

        Contract-only: `download` overrides the write path and calls
        `_search_fetch_each` directly (to concat + window itself), mirroring
        the `earthlens.openaq` sibling, so this is not on the live path.
        """
        return self._search_fetch_each(desc="Sensor.Community sensors", unit="sensor")

    def _window(self) -> tuple[pd.Timestamp, pd.Timestamp]:
        """Return the `[lower, upper)` UTC filter bounds for the request.

        A date-granular end (midnight) is extended by one day so the whole
        end day is inclusive — the common path, and identical to AirNow /
        EEA. A non-midnight `end` (only reachable via an hour-aware `fmt`)
        yields a **half-open** `[start, end)` window here; AirNow instead
        treats its end hour as inclusive (its API takes an hourly range), so
        hour-granular callers should account for that one-endpoint
        difference.

        Returns:
            tuple[pd.Timestamp, pd.Timestamp]: `(lower, upper)`, tz-aware
                UTC; `upper` is exclusive.
        """
        lower = pd.Timestamp(self.time.start_date, tz="UTC")
        end = self.time.end_date
        if end.hour == 0 and end.minute == 0:
            upper = pd.Timestamp(end, tz="UTC") + pd.Timedelta(days=1)
        else:
            upper = pd.Timestamp(end, tz="UTC")
        return lower, upper

    def download(
        self,
        progress_bar: bool = True,
        limit: int | None = None,
    ) -> pd.DataFrame:
        """Discover + fetch readings, write them to `path`, return the frame.

        Emits a `LicenseWarning` (ODbL), runs the live-API discovery then
        the per-sensor archive fetch under a `tqdm` bar, concatenates and
        windows the readings to the exact date range, writes the
        long-format result to `path` as CSV (or Parquet), and returns it.
        An empty result returns — and writes — a schema-only DataFrame.

        Args:
            progress_bar: Show the per-sensor `tqdm` bar. Defaults to
                `True`.
            limit: Cap on the total readings fetched, across every discovered
                sensor. Applied as each sensor's frame arrives, so a sensor
                past the cap never has its daily archive files downloaded.
                `None` (the default) fetches everything. The cap is on rows
                *fetched*, before the window filter, so the returned frame can
                be shorter than the cap.

        Returns:
            pd.DataFrame: The long-format readings (schema columns,
                `datetime_utc` tz-aware UTC). Empty (schema-only) when
                nothing matched.
        """
        self._limit = self.check_limit(limit)
        warnings.warn(_LICENSE_TEXT, LicenseWarning, stacklevel=2)

        frames = self._search_fetch_each(
            progress_bar=progress_bar, desc="Sensor.Community sensors", unit="sensor"
        )
        non_empty = [frame for frame in frames if not frame.empty]
        # Release the per-sensor frames as we go: `concat` copies, so holding
        # the sources alongside the combined frame — and then alongside the
        # windowed copy — keeps up to three full copies of the request in RAM.
        frames.clear()
        if non_empty:
            combined = pd.concat(non_empty, ignore_index=True)
            non_empty.clear()
            lower, upper = self._window()
            mask = (combined["datetime_utc"] >= lower) & (
                combined["datetime_utc"] < upper
            )
            df = combined[mask].reset_index(drop=True)
            del combined
        else:
            df = empty_frame()

        out_path = self._output_path()
        if self._file_format == "parquet":
            df.to_parquet(out_path, index=False)
        else:
            df.to_csv(out_path, index=False)

        if len(df):
            logger.info(
                f"Sensor.Community download summary: {len(df)} reading(s) "
                f"across {df['station_id'].nunique()} sensor(s) written to "
                f"{out_path}"
            )
        else:
            logger.warning(
                "Sensor.Community download summary: no readings matched the "
                f"request; wrote an empty (schema-only) frame to {out_path}"
            )
        return df

    def _output_path(self) -> Path:
        """Compose the per-request output file path under `root_dir`."""
        ext = "parquet" if self._file_format == "parquet" else "csv"
        params = "-".join(self.vars)
        start = self.time.start_date.strftime("%Y%m%d")
        end = self.time.end_date.strftime("%Y%m%d")
        return self.root_dir / f"sensor_community_{params}_{start}_{end}.{ext}"

__init__(start, end, variables, lat_lim, lon_lim, temporal_resolution='raw', path=None, fmt='%Y-%m-%d', session=None, client=None, file_format='csv')

Initialise a Sensor.Community backend instance.

Parameters:

Name	Type	Description	Default
start	str	Inclusive start of the observation window, as a string parsed with fmt.	required
end	str	Inclusive end of the observation window.	required
variables	list[str]	List of pollutant names to fetch (["pm25"], ["temperature", "humidity"]). Mapped to CSV columns + serving sensor types via the catalog. An empty list defaults to ["pm25"].	required
lat_lim	list[float]	[lat_min, lat_max] bounding-box latitudes in degrees.	required
lon_lim	list[float]	[lon_min, lon_max] bounding-box longitudes in degrees.	required
temporal_resolution	str	Recorded for provenance; Sensor.Community has no server-side rollup. Accepts "raw" (default), "hourly", or "daily" (the facade default).	'raw'
path	Path | str | None	Output directory for the written CSV / Parquet. Created by the parent class if absent.	None
fmt	str	strptime format for start / end.	'%Y-%m-%d'
session	Session | None	An existing requests.Session to reuse (used to build the default client). Injectable for tests.	None
client	SensorCommunityClient | None	A SensorCommunityClient to reuse. Injectable so tests supply a fake transport; when None (default) one is built lazily from session.	None
file_format	FileFormat	Output format — "csv" (default) or "parquet".	'csv'

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

def __init__(
    self,
    start: str,
    end: str,
    variables: list[str],
    lat_lim: list[float],
    lon_lim: list[float],
    temporal_resolution: str = "raw",
    path: Path | str | None = None,
    fmt: str = "%Y-%m-%d",
    session: requests.Session | None = None,
    client: SensorCommunityClient | None = None,
    file_format: FileFormat = "csv",
):
    """Initialise a Sensor.Community backend instance.

    Args:
        start: Inclusive start of the observation window, as a string
            parsed with `fmt`.
        end: Inclusive end of the observation window.
        variables: List of pollutant names to fetch (`["pm25"]`,
            `["temperature", "humidity"]`). Mapped to CSV columns +
            serving sensor types via the catalog. An empty list
            defaults to `["pm25"]`.
        lat_lim: `[lat_min, lat_max]` bounding-box latitudes in
            degrees.
        lon_lim: `[lon_min, lon_max]` bounding-box longitudes in
            degrees.
        temporal_resolution: Recorded for provenance; Sensor.Community
            has no server-side rollup. Accepts `"raw"` (default),
            `"hourly"`, or `"daily"` (the facade default).
        path: Output directory for the written CSV / Parquet. Created
            by the parent class if absent.
        fmt: `strptime` format for `start` / `end`.
        session: An existing `requests.Session` to reuse (used to
            build the default client). Injectable for tests.
        client: A `SensorCommunityClient` to reuse. Injectable so
            tests supply a fake transport; when `None` (default) one
            is built lazily from `session`.
        file_format: Output format — `"csv"` (default) or `"parquet"`.
    """
    if isinstance(variables, dict):
        raise TypeError(
            "SensorCommunity `variables` must be a list of pollutant names "
            "(e.g. ['pm25', 'pm10']), not a mapping."
        )
    self._session = session
    self._client_obj = client
    self._file_format: FileFormat = file_format
    self._catalog = Catalog()
    self._columns: dict[str, str] | None = None
    self._units: dict[str, str] | None = None
    super().__init__(
        start=start,
        end=end,
        variables=list(variables) or list(_DEFAULT_PARAMETERS),
        temporal_resolution=temporal_resolution,
        lat_lim=lat_lim,
        lon_lim=lon_lim,
        fmt=fmt,
        path=path,
    )

download(progress_bar=True, limit=None)

Discover + fetch readings, write them to path, return the frame.

Emits a LicenseWarning (ODbL), runs the live-API discovery then the per-sensor archive fetch under a tqdm bar, concatenates and windows the readings to the exact date range, writes the long-format result to path as CSV (or Parquet), and returns it. An empty result returns — and writes — a schema-only DataFrame.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Show the per-sensor tqdm bar. Defaults to True.	True
limit	int | None	Cap on the total readings fetched, across every discovered sensor. Applied as each sensor's frame arrives, so a sensor past the cap never has its daily archive files downloaded. None (the default) fetches everything. The cap is on rows fetched, before the window filter, so the returned frame can be shorter than the cap.	None

Returns:

Type	Description
DataFrame	pd.DataFrame: The long-format readings (schema columns, datetime_utc tz-aware UTC). Empty (schema-only) when nothing matched.

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

def download(
    self,
    progress_bar: bool = True,
    limit: int | None = None,
) -> pd.DataFrame:
    """Discover + fetch readings, write them to `path`, return the frame.

    Emits a `LicenseWarning` (ODbL), runs the live-API discovery then
    the per-sensor archive fetch under a `tqdm` bar, concatenates and
    windows the readings to the exact date range, writes the
    long-format result to `path` as CSV (or Parquet), and returns it.
    An empty result returns — and writes — a schema-only DataFrame.

    Args:
        progress_bar: Show the per-sensor `tqdm` bar. Defaults to
            `True`.
        limit: Cap on the total readings fetched, across every discovered
            sensor. Applied as each sensor's frame arrives, so a sensor
            past the cap never has its daily archive files downloaded.
            `None` (the default) fetches everything. The cap is on rows
            *fetched*, before the window filter, so the returned frame can
            be shorter than the cap.

    Returns:
        pd.DataFrame: The long-format readings (schema columns,
            `datetime_utc` tz-aware UTC). Empty (schema-only) when
            nothing matched.
    """
    self._limit = self.check_limit(limit)
    warnings.warn(_LICENSE_TEXT, LicenseWarning, stacklevel=2)

    frames = self._search_fetch_each(
        progress_bar=progress_bar, desc="Sensor.Community sensors", unit="sensor"
    )
    non_empty = [frame for frame in frames if not frame.empty]
    # Release the per-sensor frames as we go: `concat` copies, so holding
    # the sources alongside the combined frame — and then alongside the
    # windowed copy — keeps up to three full copies of the request in RAM.
    frames.clear()
    if non_empty:
        combined = pd.concat(non_empty, ignore_index=True)
        non_empty.clear()
        lower, upper = self._window()
        mask = (combined["datetime_utc"] >= lower) & (
            combined["datetime_utc"] < upper
        )
        df = combined[mask].reset_index(drop=True)
        del combined
    else:
        df = empty_frame()

    out_path = self._output_path()
    if self._file_format == "parquet":
        df.to_parquet(out_path, index=False)
    else:
        df.to_csv(out_path, index=False)

    if len(df):
        logger.info(
            f"Sensor.Community download summary: {len(df)} reading(s) "
            f"across {df['station_id'].nunique()} sensor(s) written to "
            f"{out_path}"
        )
    else:
        logger.warning(
            "Sensor.Community download summary: no readings matched the "
            f"request; wrote an empty (schema-only) frame to {out_path}"
        )
    return df

earthlens.sensor_community.catalog

Pollutant dispatch table for the Sensor.Community backend.

Sensor.Community archives one CSV per (sensor, day); each CSV's measurement columns depend on the sensor type (particulate sensors report P0/P1/P2, climate sensors report temperature/humidity/pressure). Users pass earthlens pollutant names in variables=[...]; this module maps each name to the CSV column it lives in and the sensor_types (archive slugs) whose files carry it.

Like the OpenAQ / AirNow / EEA provider tables it is deliberately tiny and fixed, so there is no refresh / probe / audit tooling and no tools/sensor_community/ directory; adding a pollutant later is a hand-edit of one YAML row.

Catalog is a thin earthlens.base.AbstractCatalog subclass that loads the bundled sensor_community_data_catalog.yaml and exposes each row as a Pollutant. Resolve a single name with Catalog.get_pollutant (raises with a did-you-mean hint on an unknown name), the union of the sensor-type slugs for a list of names with Catalog.sensor_types_for, or the column -> name reverse map for a list of names with Catalog.columns_for. CATALOG_PATH is the path to the bundled YAML and is monkey-patchable in tests.

Catalog

Bases: AbstractCatalog

Pollutant catalog for the Sensor.Community backend.

Reads the bundled sensor_community_data_catalog.yaml (shipped as package data) and exposes its pollutants: block as a map of Pollutant rows. Instantiate with no arguments (Catalog()); model_post_init loads and validates the YAML in one pass.

Attributes:

Name	Type	Description
pollutants	dict[str, Pollutant]	Map from the user-facing pollutant name to its Pollutant dispatch row.

Examples:

• Resolve names to the union of serving sensor types and to CSV columns:

  >>> from earthlens.sensor_community import Catalog
  >>> cat = Catalog()
  >>> "sds011" in cat.sensor_types_for(["pm25"])
  True
  >>> cat.columns_for(["pm25", "pm10"])
  {'P2': 'pm25', 'P1': 'pm10'}
  

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

class Catalog(AbstractCatalog):
    """Pollutant catalog for the Sensor.Community backend.

    Reads the bundled `sensor_community_data_catalog.yaml` (shipped as
    package data) and exposes its `pollutants:` block as a map of
    `Pollutant` rows. Instantiate with no arguments (`Catalog()`);
    `model_post_init` loads and validates the YAML in one pass.

    Attributes:
        pollutants: Map from the user-facing pollutant name to its
            `Pollutant` dispatch row.

    Examples:
        - Resolve names to the union of serving sensor types and to CSV
          columns:
            ```python
            >>> from earthlens.sensor_community import Catalog
            >>> cat = Catalog()
            >>> "sds011" in cat.sensor_types_for(["pm25"])
            True
            >>> cat.columns_for(["pm25", "pm10"])
            {'P2': 'pm25', 'P1': 'pm10'}

            ```
    """

    _catalog_kind: str = "Sensor.Community pollutant catalog"
    _entry_noun: str = "pollutants"

    #: The pollutant rows live in the base `datasets` field so the inherited
    #: dict surface works unchanged. `pollutants` is the domain-named alias.
    datasets: dict[str, Pollutant] = Field(default_factory=dict)

    @model_validator(mode="before")
    @classmethod
    def _accept_pollutants_alias(cls, data: Any) -> Any:
        """Accept the `pollutants=` kwarg as an alias for `datasets`.

        Callers and tests construct `Catalog(pollutants={...})`. The rows
        live in the base `datasets` field, so rewrite that key on the way
        in. An explicit `datasets=` always wins.

        Args:
            data: The raw model input (a mapping when constructed with
                keyword arguments).

        Returns:
            The input with `pollutants` renamed to `datasets`, untouched
            otherwise.
        """
        if isinstance(data, dict) and "pollutants" in data and "datasets" not in data:
            data = dict(data)
            data["datasets"] = data.pop("pollutants")
        return data

    @property
    def pollutants(self) -> dict[str, Pollutant]:
        """The pollutant map — alias for the base `datasets` field.

        Returns:
            dict[str, Pollutant]: The same mapping stored in `datasets`.
        """
        return self.datasets

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

        Returns:
            dict[str, Any]: The `datasets` read from
                the bundled catalog.
        """
        return {"datasets": Catalog.load().datasets}

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

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

        Returns:
            A fully-populated `Catalog`.

        Raises:
            ValueError: If `catalog_path` does not exist, or if the file has no
                `pollutants:` block, or a row fails `Pollutant` validation.
        """
        catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
        cached = load_catalog(
            catalog_path,
            _CATALOG_CACHE,
            _parse_sensor_community_catalog,
            provider="Sensor.Community",
        )
        return cls(datasets=dict(cached))

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

        Returns:
            dict[str, Pollutant]: Same object as `datasets` / `pollutants`.
        """
        return self.datasets

    def get_pollutant(self, name: str) -> Pollutant:
        """Resolve a pollutant name to its `Pollutant` row.

        Thin wrapper over the inherited `get_dataset`, which raises a
        `ValueError` with a did-you-mean hint on an unknown name.

        Args:
            name: A user-facing pollutant name (`"pm25"`, `"temperature"`).

        Returns:
            Pollutant: The matching dispatch row.

        Raises:
            ValueError: If `name` is not a known pollutant.
        """
        return cast("Pollutant", self.get_dataset(name))

    def sensor_types_for(self, names: list[str]) -> set[str]:
        """Return the union of serving sensor-type slugs for `names`.

        Args:
            names: User-facing pollutant names to resolve.

        Returns:
            set[str]: Every archive sensor-type slug whose CSV carries at
                least one of the requested pollutants.

        Raises:
            ValueError: If any name is unknown (via `get_pollutant`).
        """
        types: set[str] = set()
        for name in names:
            types.update(self.get_pollutant(name).sensor_types)
        return types

    def columns_for(self, names: list[str]) -> dict[str, str]:
        """Return the CSV `column` -> pollutant name map for `names`.

        Used at parse time to pull every requested pollutant's value out
        of one sensor CSV in a single pass.

        Args:
            names: User-facing pollutant names to resolve.

        Returns:
            dict[str, str]: Each requested pollutant's CSV column mapped
                to its name (`{"P2": "pm25", "P1": "pm10"}`).

        Raises:
            ValueError: If any name is unknown (via `get_pollutant`).
        """
        return {self.get_pollutant(name).column: name for name in names}

pollutants property

The pollutant map — alias for the base datasets field.

Returns:

Type	Description
dict[str, Pollutant]	dict[str, Pollutant]: The same mapping stored in datasets.

columns_for(names)

Return the CSV column -> pollutant name map for names.

Used at parse time to pull every requested pollutant's value out of one sensor CSV in a single pass.

Parameters:

Name	Type	Description	Default
names	list[str]	User-facing pollutant names to resolve.	required

Returns:

Type	Description
dict[str, str]	dict[str, str]: Each requested pollutant's CSV column mapped to its name ({"P2": "pm25", "P1": "pm10"}).

Raises:

Type	Description
ValueError	If any name is unknown (via get_pollutant).

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

def columns_for(self, names: list[str]) -> dict[str, str]:
    """Return the CSV `column` -> pollutant name map for `names`.

    Used at parse time to pull every requested pollutant's value out
    of one sensor CSV in a single pass.

    Args:
        names: User-facing pollutant names to resolve.

    Returns:
        dict[str, str]: Each requested pollutant's CSV column mapped
            to its name (`{"P2": "pm25", "P1": "pm10"}`).

    Raises:
        ValueError: If any name is unknown (via `get_pollutant`).
    """
    return {self.get_pollutant(name).column: name for name in names}

get_catalog()

Return the pollutant map (satisfies the abstract contract).

Returns:

Type	Description
dict[str, Pollutant]	dict[str, Pollutant]: Same object as datasets / pollutants.

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

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

    Returns:
        dict[str, Pollutant]: Same object as `datasets` / `pollutants`.
    """
    return self.datasets

get_pollutant(name)

Resolve a pollutant name to its Pollutant row.

Thin wrapper over the inherited get_dataset, which raises a ValueError with a did-you-mean hint on an unknown name.

Parameters:

Name	Type	Description	Default
name	str	A user-facing pollutant name ("pm25", "temperature").	required

Returns:

Name	Type	Description
Pollutant	Pollutant	The matching dispatch row.

Raises:

Type	Description
ValueError	If name is not a known pollutant.

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

def get_pollutant(self, name: str) -> Pollutant:
    """Resolve a pollutant name to its `Pollutant` row.

    Thin wrapper over the inherited `get_dataset`, which raises a
    `ValueError` with a did-you-mean hint on an unknown name.

    Args:
        name: A user-facing pollutant name (`"pm25"`, `"temperature"`).

    Returns:
        Pollutant: The matching dispatch row.

    Raises:
        ValueError: If `name` is not a known pollutant.
    """
    return cast("Pollutant", self.get_dataset(name))

load(catalog_path=None) classmethod

Read the Sensor.Community pollutant catalog from disk.

Parameters:

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

Returns:

Type	Description
Catalog	A fully-populated Catalog.

Raises:

Type	Description
ValueError	If catalog_path does not exist, or if the file has no pollutants: block, or a row fails Pollutant validation.

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

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

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

    Returns:
        A fully-populated `Catalog`.

    Raises:
        ValueError: If `catalog_path` does not exist, or if the file has no
            `pollutants:` block, or a row fails `Pollutant` validation.
    """
    catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
    cached = load_catalog(
        catalog_path,
        _CATALOG_CACHE,
        _parse_sensor_community_catalog,
        provider="Sensor.Community",
    )
    return cls(datasets=dict(cached))

sensor_types_for(names)

Return the union of serving sensor-type slugs for names.

Parameters:

Name	Type	Description	Default
names	list[str]	User-facing pollutant names to resolve.	required

Returns:

Type	Description
set[str]	set[str]: Every archive sensor-type slug whose CSV carries at least one of the requested pollutants.

Raises:

Type	Description
ValueError	If any name is unknown (via get_pollutant).

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

def sensor_types_for(self, names: list[str]) -> set[str]:
    """Return the union of serving sensor-type slugs for `names`.

    Args:
        names: User-facing pollutant names to resolve.

    Returns:
        set[str]: Every archive sensor-type slug whose CSV carries at
            least one of the requested pollutants.

    Raises:
        ValueError: If any name is unknown (via `get_pollutant`).
    """
    types: set[str] = set()
    for name in names:
        types.update(self.get_pollutant(name).sensor_types)
    return types

Pollutant

Bases: BaseModel

One Sensor.Community pollutant's dispatch row.

The user-facing name is the parent key in Catalog.pollutants and is also stored on the row as name so a resolved Pollutant is self-describing.

Attributes:

Name	Type	Description
name	str	Short machine name ("pm25", "temperature"); matches the catalog key.
column	str	The CSV column this pollutant is read from ("P2" for PM2.5, "temperature").
sensor_types	list[str]	Archive sensor-type slugs whose per-sensor CSV carries column (["sds011", "pms5003", ...]). Used to filter discovery and choose which archive files to fetch.
units	str	The reporting unit ("µg/m³", "°C"). Sensor.Community reports pressure in pascals.
display_name	str	Human-readable label for docs / plots ("PM2.5").
group	PollutantGroup	Coarse classification — "particulate", "meteorological", or "other".

Examples:

• Build a row directly:

  >>> from earthlens.sensor_community import Pollutant
  >>> p = Pollutant(name="pm25", column="P2", sensor_types=["sds011"])
  >>> (p.column, p.sensor_types)
  ('P2', ['sds011'])
  

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

class Pollutant(BaseModel):
    """One Sensor.Community pollutant's dispatch row.

    The user-facing name is the parent key in `Catalog.pollutants` and is
    also stored on the row as `name` so a resolved `Pollutant` is
    self-describing.

    Attributes:
        name: Short machine name (`"pm25"`, `"temperature"`); matches the
            catalog key.
        column: The CSV column this pollutant is read from (`"P2"` for
            PM2.5, `"temperature"`).
        sensor_types: Archive sensor-type slugs whose per-sensor CSV
            carries `column` (`["sds011", "pms5003", ...]`). Used to
            filter discovery and choose which archive files to fetch.
        units: The reporting unit (`"µg/m³"`, `"°C"`). Sensor.Community
            reports pressure in pascals.
        display_name: Human-readable label for docs / plots (`"PM2.5"`).
        group: Coarse classification — `"particulate"`, `"meteorological"`,
            or `"other"`.

    Examples:
        - Build a row directly:
            ```python
            >>> from earthlens.sensor_community import Pollutant
            >>> p = Pollutant(name="pm25", column="P2", sensor_types=["sds011"])
            >>> (p.column, p.sensor_types)
            ('P2', ['sds011'])

            ```
    """

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

    name: str
    column: str
    sensor_types: list[str] = Field(min_length=1)
    units: str = ""
    display_name: str = ""
    group: PollutantGroup = "other"

clear_catalog_cache()

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

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

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

earthlens.sensor_community._helpers

Client, parsing, and licence helpers for the Sensor.Community backend.

Sensor.Community exposes two hosts the backend needs:

• the live JSON API (data.sensor.community) — the last ~5 minutes of every sensor globally, bbox-filterable, used to discover which sensors are active in the request bbox (the archive has no bbox index);
• the archive (archive.sensor.community) — one CSV per (sensor, day), <date>/<date>_<sensor_type>_sensor_<id>.csv, ;-separated, used to fetch each discovered sensor's history over the date range.

SensorCommunityClient wraps an injectable requests.Session over both hosts with 429/Retry-After back-off; a missing archive file (404) returns None so the backend can log-and-skip without failing the whole request. LicenseWarning flags the ODbL / crowdsourced-quality caveat.

LicenseWarning

Bases: UserWarning

Warns that Sensor.Community data carries ODbL / quality obligations.

Sensor.Community measurements are crowdsourced from low-cost sensors and licensed under the Open Database License (ODbL): redistribution must keep the attribution and share-alike terms, and the readings are not reference-grade. The backend emits this once per download() so a downstream user is told rather than discovering it silently.

Source code in libs/providers/atmosphere/src/earthlens/sensor_community/_helpers.py

class LicenseWarning(UserWarning):
    """Warns that Sensor.Community data carries ODbL / quality obligations.

    Sensor.Community measurements are crowdsourced from low-cost sensors
    and licensed under the Open Database License (ODbL): redistribution
    must keep the attribution and share-alike terms, and the readings are
    not reference-grade. The backend emits this once per `download()` so a
    downstream user is told rather than discovering it silently.
    """

SensorCommunityClient

Injectable client over the Sensor.Community live + archive hosts.

Delegates the transport (session, 429/Retry-After back-off) to the shared earthlens.base.http.HttpClient, keeping only the two-host request shaping: the live JSON snapshot and one per-sensor archive CSV (404-tolerant).

Attributes:

Name	Type	Description
max_retries	int	Maximum number of 429 retries before raising.
backoff_factor	float	Base seconds for exponential back-off when no Retry-After header is present (wait = factor * 2**attempt).
timeout	Timeout	Per-request timeout in seconds — a float or a (connect, read) pair.

Source code in libs/providers/atmosphere/src/earthlens/sensor_community/_helpers.py

class SensorCommunityClient:
    """Injectable client over the Sensor.Community live + archive hosts.

    Delegates the transport (session, `429`/`Retry-After` back-off) to
    the shared `earthlens.base.http.HttpClient`, keeping only the
    two-host request shaping: the live JSON snapshot and one per-sensor
    archive CSV (`404`-tolerant).

    Attributes:
        max_retries: Maximum number of `429` retries before raising.
        backoff_factor: Base seconds for exponential back-off when no
            `Retry-After` header is present (wait = factor * 2**attempt).
        timeout: Per-request timeout in seconds — a float or a `(connect, read)` pair.
    """

    def __init__(
        self,
        *,
        session: requests.Session | None = None,
        max_retries: int = 3,
        backoff_factor: float = 1.0,
        timeout: Timeout = 60.0,
        sleep: Callable[[float], None] = time.sleep,
    ) -> None:
        """Build a client over both Sensor.Community hosts.

        Args:
            session: An existing `requests.Session` to reuse. Defaults to a
                fresh session. Injectable so tests supply a fake transport.
            max_retries: Maximum `429` retries before raising.
            backoff_factor: Base seconds for exponential back-off.
            timeout: Per-request timeout in seconds — a float or a
                `(connect, read)` pair.
            sleep: The sleep function used between retries. Defaults to
                `time.sleep`; injectable so tests run without real delays.
        """
        self._http = HttpClient(
            session=session if session is not None else requests.Session(),
            max_retries=max_retries,
            backoff_factor=backoff_factor,
            timeout=timeout,
            status_forcelist=(429,),
            max_backoff=None,
            sleep=sleep,
        )

    @property
    def max_retries(self) -> int:
        """Maximum `429` retries before the last error is raised."""
        return self._http.max_retries

    @property
    def backoff_factor(self) -> float:
        """Base seconds for exponential back-off (no `Retry-After`)."""
        return self._http.backoff_factor

    @property
    def timeout(self) -> Timeout:
        """Per-request timeout in seconds (a float or a `(connect, read)` pair)."""
        return self._http.timeout

    def live_snapshot(self) -> list[dict[str, Any]]:
        """Fetch the live JSON API's last-~5-minute global sensor snapshot.

        Returns:
            list[dict[str, Any]]: The array of live measurement records
                (each with `location` and `sensor` sub-objects).

        Raises:
            requests.HTTPError: On a non-`429` error status.
        """
        payload = self._http.get_json(LIVE_URL)
        return payload if isinstance(payload, list) else []

    def archive_csv(self, date: str, sensor_type: str, sensor_id: str) -> str | None:
        """Fetch one per-sensor daily archive CSV, or `None` when absent.

        Args:
            date: The archive day as `YYYY-MM-DD`.
            sensor_type: The archive sensor-type slug (`"sds011"`).
            sensor_id: The sensor's numeric id (as a string).

        Returns:
            str | None: The CSV text, or `None` when the file does not
                exist (`404`) — the sensor did not report that day.

        Raises:
            requests.HTTPError: On a non-`404`, non-`429` error status.
        """
        url = f"{ARCHIVE_URL}/{date}/{date}_{sensor_type}_sensor_{sensor_id}.csv"
        response = self._http.get(url, raise_for_status=False)
        if response.status_code == 404:
            return None
        response.raise_for_status()
        return response.text

backoff_factor property

Base seconds for exponential back-off (no Retry-After).

max_retries property

Maximum 429 retries before the last error is raised.

timeout property

Per-request timeout in seconds (a float or a (connect, read) pair).

__init__(*, session=None, max_retries=3, backoff_factor=1.0, timeout=60.0, sleep=time.sleep)

Build a client over both Sensor.Community hosts.

Parameters:

Name	Type	Description	Default
session	Session | None	An existing requests.Session to reuse. Defaults to a fresh session. Injectable so tests supply a fake transport.	None
max_retries	int	Maximum 429 retries before raising.	3
backoff_factor	float	Base seconds for exponential back-off.	1.0
timeout	Timeout	Per-request timeout in seconds — a float or a (connect, read) pair.	60.0
sleep	Callable[[float], None]	The sleep function used between retries. Defaults to time.sleep; injectable so tests run without real delays.	sleep

Source code in libs/providers/atmosphere/src/earthlens/sensor_community/_helpers.py

def __init__(
    self,
    *,
    session: requests.Session | None = None,
    max_retries: int = 3,
    backoff_factor: float = 1.0,
    timeout: Timeout = 60.0,
    sleep: Callable[[float], None] = time.sleep,
) -> None:
    """Build a client over both Sensor.Community hosts.

    Args:
        session: An existing `requests.Session` to reuse. Defaults to a
            fresh session. Injectable so tests supply a fake transport.
        max_retries: Maximum `429` retries before raising.
        backoff_factor: Base seconds for exponential back-off.
        timeout: Per-request timeout in seconds — a float or a
            `(connect, read)` pair.
        sleep: The sleep function used between retries. Defaults to
            `time.sleep`; injectable so tests run without real delays.
    """
    self._http = HttpClient(
        session=session if session is not None else requests.Session(),
        max_retries=max_retries,
        backoff_factor=backoff_factor,
        timeout=timeout,
        status_forcelist=(429,),
        max_backoff=None,
        sleep=sleep,
    )

archive_csv(date, sensor_type, sensor_id)

Fetch one per-sensor daily archive CSV, or None when absent.

Parameters:

Name	Type	Description	Default
date	str	The archive day as YYYY-MM-DD.	required
sensor_type	str	The archive sensor-type slug ("sds011").	required
sensor_id	str	The sensor's numeric id (as a string).	required

Returns:

Type	Description
str | None	str | None: The CSV text, or None when the file does not exist (404) — the sensor did not report that day.

Raises:

Type	Description
HTTPError	On a non-404, non-429 error status.

Source code in libs/providers/atmosphere/src/earthlens/sensor_community/_helpers.py

def archive_csv(self, date: str, sensor_type: str, sensor_id: str) -> str | None:
    """Fetch one per-sensor daily archive CSV, or `None` when absent.

    Args:
        date: The archive day as `YYYY-MM-DD`.
        sensor_type: The archive sensor-type slug (`"sds011"`).
        sensor_id: The sensor's numeric id (as a string).

    Returns:
        str | None: The CSV text, or `None` when the file does not
            exist (`404`) — the sensor did not report that day.

    Raises:
        requests.HTTPError: On a non-`404`, non-`429` error status.
    """
    url = f"{ARCHIVE_URL}/{date}/{date}_{sensor_type}_sensor_{sensor_id}.csv"
    response = self._http.get(url, raise_for_status=False)
    if response.status_code == 404:
        return None
    response.raise_for_status()
    return response.text

live_snapshot()

Fetch the live JSON API's last-~5-minute global sensor snapshot.

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: The array of live measurement records (each with location and sensor sub-objects).

Raises:

Type	Description
HTTPError	On a non-429 error status.

Source code in libs/providers/atmosphere/src/earthlens/sensor_community/_helpers.py

def live_snapshot(self) -> list[dict[str, Any]]:
    """Fetch the live JSON API's last-~5-minute global sensor snapshot.

    Returns:
        list[dict[str, Any]]: The array of live measurement records
            (each with `location` and `sensor` sub-objects).

    Raises:
        requests.HTTPError: On a non-`429` error status.
    """
    payload = self._http.get_json(LIVE_URL)
    return payload if isinstance(payload, list) else []

empty_frame()

Return an empty DataFrame with the exact long-format schema.

Returns:

Type	Description
DataFrame	pd.DataFrame: Zero rows, SCHEMA columns and dtypes.

Source code in libs/providers/atmosphere/src/earthlens/sensor_community/_helpers.py

def empty_frame() -> pd.DataFrame:
    """Return an empty DataFrame with the exact long-format schema.

    Returns:
        pd.DataFrame: Zero rows, `SCHEMA` columns and dtypes.
    """
    return pd.DataFrame({column: [] for column in SCHEMA}).astype(SCHEMA)

frame_from_csv(text, columns, units, default_sensor_type=None)

Reshape one per-sensor archive CSV into the backend's long schema.

For each requested pollutant whose CSV column is present, emits one row per reading (station_id / sensor_type / lat / lon / timestamp come from the CSV). Rows with a non-numeric value or unparseable timestamp are dropped.

Parameters:

Name	Type	Description	Default
text	str	The ;-separated CSV text of one (sensor, day) file.	required
columns	dict[str, str]	CSV column -> pollutant name for the requested pollutants ({"P2": "pm25", "P1": "pm10"}).	required
units	dict[str, str]	Pollutant name -> reporting unit string.	required
default_sensor_type	str | None	The archive sensor-type slug known from discovery, used for the sensor_type column when the CSV omits its own sensor_type column. None leaves it null in that (degenerate) case.	None

Returns:

Type	Description
DataFrame	pd.DataFrame: The readings in the SCHEMA columns / dtypes; empty when the CSV has none of the requested columns.

Source code in libs/providers/atmosphere/src/earthlens/sensor_community/_helpers.py

def frame_from_csv(
    text: str,
    columns: dict[str, str],
    units: dict[str, str],
    default_sensor_type: str | None = None,
) -> pd.DataFrame:
    """Reshape one per-sensor archive CSV into the backend's long schema.

    For each requested pollutant whose CSV `column` is present, emits one
    row per reading (`station_id` / `sensor_type` / `lat` / `lon` /
    `timestamp` come from the CSV). Rows with a non-numeric value or
    unparseable timestamp are dropped.

    Args:
        text: The `;`-separated CSV text of one (sensor, day) file.
        columns: CSV column -> pollutant name for the requested pollutants
            (`{"P2": "pm25", "P1": "pm10"}`).
        units: Pollutant name -> reporting unit string.
        default_sensor_type: The archive sensor-type slug known from
            discovery, used for the `sensor_type` column when the CSV omits
            its own `sensor_type` column. `None` leaves it null in that
            (degenerate) case.

    Returns:
        pd.DataFrame: The readings in the `SCHEMA` columns / dtypes; empty
            when the CSV has none of the requested columns.
    """
    raw = pd.read_csv(io.StringIO(text), sep=";")
    present = {col: name for col, name in columns.items() if col in raw.columns}
    if raw.empty or not present:
        return empty_frame()
    # A malformed CSV that has the value column but lacks a structural one must
    # be skipped, not raise KeyError and abort the whole multi-sensor download
    # (the "missing file is logged & skipped, never a silent gap" contract).
    missing = {"sensor_id", "timestamp", "lat", "lon"} - set(raw.columns)
    if missing:
        logger.warning(
            f"Sensor.Community: archive CSV missing structural column(s) "
            f"{sorted(missing)}; skipping this file."
        )
        return empty_frame()
    frames: list[pd.DataFrame] = []
    # Normalise the CSV's upper-case sensor type (`SDS011`) to the lower-case
    # archive slug (`sds011`) so the output column matches the discovery
    # metadata and the archive URL.
    sensor_type = (
        raw["sensor_type"].astype(str).str.lower()
        if "sensor_type" in raw
        else default_sensor_type
    )
    for col, name in present.items():
        sub = pd.DataFrame(index=raw.index)
        sub["station_id"] = raw["sensor_id"].astype(str)
        sub["sensor_type"] = sensor_type
        sub["parameter"] = name
        sub["datetime_utc"] = pd.to_datetime(
            raw["timestamp"], utc=True, errors="coerce"
        )
        sub["value"] = pd.to_numeric(raw[col], errors="coerce")
        sub["units"] = units.get(name, "")
        sub["lat"] = pd.to_numeric(raw["lat"], errors="coerce")
        sub["lon"] = pd.to_numeric(raw["lon"], errors="coerce")
        sub["provider"] = PROVIDER
        frames.append(sub)
    out = pd.concat(frames, ignore_index=True)
    out = out.dropna(subset=["value", "datetime_utc"]).reset_index(drop=True)
    if out.empty:
        return empty_frame()
    return out.astype(SCHEMA)

sensors_in_bbox(snapshot, lat_lim, lon_lim, wanted_types)

Filter a live snapshot to unique sensors in the bbox of wanted types.

Parameters:

Name	Type	Description	Default
snapshot	list[dict[str, Any]]	The live JSON API records from live_snapshot.	required
lat_lim	tuple[float, float]	(lat_min, lat_max) of the request bbox.	required
lon_lim	tuple[float, float]	(lon_min, lon_max) of the request bbox.	required
wanted_types	set[str]	Archive sensor-type slugs (lower-case) to keep.	required

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: One entry per unique (sensor_id, sensor_type) — {"sensor_id", "sensor_type", "lat", "lon"} — sorted by sensor id for determinism.

Source code in libs/providers/atmosphere/src/earthlens/sensor_community/_helpers.py

def sensors_in_bbox(
    snapshot: list[dict[str, Any]],
    lat_lim: tuple[float, float],
    lon_lim: tuple[float, float],
    wanted_types: set[str],
) -> list[dict[str, Any]]:
    """Filter a live snapshot to unique sensors in the bbox of wanted types.

    Args:
        snapshot: The live JSON API records from `live_snapshot`.
        lat_lim: `(lat_min, lat_max)` of the request bbox.
        lon_lim: `(lon_min, lon_max)` of the request bbox.
        wanted_types: Archive sensor-type slugs (lower-case) to keep.

    Returns:
        list[dict[str, Any]]: One entry per unique `(sensor_id,
            sensor_type)` — `{"sensor_id", "sensor_type", "lat", "lon"}` —
            sorted by sensor id for determinism.
    """
    lat_min, lat_max = sorted(lat_lim)
    lon_min, lon_max = sorted(lon_lim)
    seen: dict[tuple[str, str], dict[str, Any]] = {}
    for record in snapshot:
        location = record.get("location") or {}
        sensor = record.get("sensor") or {}
        sensor_type = ((sensor.get("sensor_type") or {}).get("name") or "").lower()
        if sensor_type not in wanted_types:
            continue
        try:
            lat = float(cast("Any", location.get("latitude")))
            lon = float(cast("Any", location.get("longitude")))
        except (TypeError, ValueError):
            continue
        if not (lat_min <= lat <= lat_max and lon_min <= lon <= lon_max):
            continue
        sensor_id = str(sensor.get("id"))
        key = (sensor_id, sensor_type)
        if key not in seen:
            seen[key] = {
                "sensor_id": sensor_id,
                "sensor_type": sensor_type,
                "lat": lat,
                "lon": lon,
            }
    return [seen[key] for key in sorted(seen)]