Skip to content

API Reference

Variable registry additions

ERA5 pressure levels. reanalysis-era5-pressure-levels (catalog short name era5.pressure_levels) now describes the CDS pressure-level fields u, v, w (omega, Pa s⁻¹), t, z (geopotential) and q, each with a cds alias so CDSSource(...).download("reanalysis-era5-pressure-levels", Path("uv.nc"), variables=["u", "v"], levels=PressureLevels([1000, 925])) emits the CDS form names without hand-typing them (era5.pressure_levels is the catalog short name for that dataset id — see describe("era5.pressure_levels")). The single-level preset gains blh (boundary-layer height). The wrf aliases on u, v and blh name the raw wrfout variables that map onto the same CF quantity; q carries no wrf alias because QVAPOR is a mixing ratio w.r.t. dry air and must be converted (q = r / (1 + r)) before it is CF specific humidity.

Methane / trace gas. xch4, xch4_bias_corrected, xch4_precision, column_averaging_kernel, ch4_profile_apriori, dry_air_subcolumns and qa_value carry the TROPOMI L2 product names as tropomi aliases; ch4_enhancement (ppm m) is the EMIT L2B plume-enhancement quantity, and ch4_column (kg m⁻², atmosphere_mass_content_of_methane) is the canonical column-mass output of the xrops.atm.gas.ch4 operators. All of them resolve through xrproducts.resolve and stamp CF attributes through apply_cf_attrs.

Local L2 products

LocalL2Source is the DataSource for satellite L2 product files that are already on disk — nothing is fetched (download() raises NotImplementedError). The catalog short names tropomi.ch4, emit.ch4 and ghgsat.ch4 (source="local") each map to a product opener that turns one file into a flat CF Dataset. The openers read NetCDF, and the core install ships no backend for it — install the local extra (pip install 'xrtoolz-products[local]', which adds netcdf4):

Short name Opener Kind Layout
tropomi.ch4 open_tropomi_ch4_l2(path, qa_min=0.5, keep_groups=False) DatasetKind.SWATH (time, scanline, ground_pixel[, layer, level, corner])
emit.ch4 open_emit_ch4_l2b(path, glt_path=None) DatasetKind.SCENE (time, y, x)
ghgsat.ch4 open_ghgsat_ch4_l2(path) DatasetKind.SCENE (time, y, x)

TROPOMI. The four product groups (PRODUCT and PRODUCT/SUPPORT_DATA/{DETAILED_RESULTS,INPUT_DATA,GEOLOCATIONS}) are read with xarray.open_datatree and merged on the shared swath dims. Variables are renamed through the registry's tropomi aliases (methane_mixing_ratio → xch4, methane_mixing_ratio_bias_corrected → xch4_bias_corrected, methane_mixing_ratio_precision → xch4_precision, methane_profile_apriori → ch4_profile_apriori, surface_pressure → sp; column_averaging_kernel, dry_air_subcolumns and qa_value keep their names, pressure_interval has no registry entry and is passed through), and latitude / longitude become the 2-D lat / lon coordinates. Time keeps the product's shape: the length-1 time index is the granule reference time (PRODUCT/time, seconds since 2010-01-01) and the per-scanline delta_time offset is folded into a scanline_time (time, scanline) datetime64 coordinate. qa_value < qa_min pixels are set to NaN (subset_where, shape preserved); keep_groups=True keeps the raw product names.

EMIT. The L2B enhancement raster (methane_plume_complex as distributed, or the registry alias ch4_enhancement — either becomes ch4_enhancement in ppm m — plus ch4_uncertainty when present) is read from a NetCDF container; rioxarray is not an xrproducts dependency, so GeoTIFF distributions have to be converted first. With glt_path the scene is orthorectified through the geometric lookup table (glt_x / glt_y, 1-based, 0 = no data) and the 2-D lon / lat of the orthorectified grid come from the GLT's own lon / lat or its GDAL geotransform attribute (full six-coefficient affine). A scalar time variable on the raster is carried through as the scene timestamp. Several scenes take one lookup table each via glt_paths=[...], zipped with paths=[...].

GHGSat. A per-plume NetCDF with xch4 (ppb) and/or ch4_enhancement on (y, x) and 2-D latitude / longitude, promoted to lon / lat coords.

LocalL2Source.open(dataset_id, path=..., bbox=..., time=..., qa_min=..., variables=...) applies the screening after the opener: bbox through subset_bbox on the 2-D coordinates (where(..., drop=True)) after the box is normalised to the dataset's longitude convention ([-180, 180] when any lon is negative, [0, 360] when any exceeds 180 — a box that still crosses the antimeridian afterwards raises), time as a per-scanline mask on scanline_time for swaths or a mask on the time coordinate for scenes, and paths=[a, b] concatenates granules along time in the order given. subset_bbox, subset_where and subset_time now live in xrproducts.types (pure xarray) and are re-exported unchanged by xrops.geom, so the reader can screen without importing xrtoolz. See the TROPOMI recipe for a worked example.

Emission inventories. open_ch4_inventory(path, variable=None, units=None, source=None) opens a gridded CH₄ inventory (EDGAR, GFEI, EPA) as ch4_emission_flux in kg m⁻² s⁻¹ on ([time,] lat, lon), with lat ascending and lon wrapped to [-180, 180). The file's area-flux units are converted through UNIT_FACTORS (kg m⁻² s⁻¹, g m⁻² s⁻¹, Mg km⁻² a⁻¹ with a 365.25-day year); the factor and the original units are kept as attributes. Per-cell totals such as EDGAR's emi files (tonnes per cell, per year or per month) are rejected because the period is not in the file: read the flx files instead. It is a plain opener, not a catalog entry.

xrproducts

xrproducts — authenticated archive readers for xarray geodata.

xrproducts turns a typed :class:Request into a service-specific payload and downloads or opens an :class:xarray.Dataset from external geoscience archives (CMEMS, CDS, AEMET). The typed request vocabulary (:class:BBox, :class:TimeRange, :class:Variable, ...) lives in :mod:xrproducts.types; this top-level namespace re-exports the reader API and the common request types flat.

Reader API:

  • :class:DataSource, :class:DatasetInfo, :class:DatasetKind
  • Adapters: :class:CMEMSSource, :class:CDSSource, :class:AemetSource, :class:LocalL2Source (files already on disk; TROPOMI / EMIT / GHGSat methane L2 openers :func:open_tropomi_ch4_l2, :func:open_emit_ch4_l2b, :func:open_ghgsat_ch4_l2; gridded emission inventories through :func:open_ch4_inventory).
  • Credentials: :class:CMEMSCredentials, :class:CDSCredentials, :class:AEMETCredentials, :func:load_cmems, :func:load_cds, :func:load_aemet.
  • Catalog: :data:CATALOG, :class:CatalogEntry, :func:all_entries, :func:describe.
  • AEMET extras: :class:AemetArchive, :class:AemetError, :class:AemetAuthError, :class:AemetRateLimitError, :class:ArchiveCoverage.
  • CDS extras: :class:CDSInsituArchive, :class:CDSFormProfile, :data:INSITU, :data:REANALYSIS.

Request vocabulary (also available via :mod:xrproducts.types):

  • Spatial / temporal / vertical: :class:BBox, :class:TimeRange, :class:DepthRange, :class:PressureLevels.
  • Composition: :class:Request.
  • Stations: :class:Station, :class:StationCollection.
  • Variables: :class:Variable, :data:REGISTRY, :func:resolve, :func:register.
  • Validation: :func:validate_variable, :func:validate_dataset, :func:apply_cf_attrs, :class:ValidationReport, :class:Issue, :class:Severity.
  • Subsetting: :func:subset_bbox, :func:subset_where, :func:subset_time.

REANALYSIS module-attribute 

REANALYSIS = CDSFormProfile(family='reanalysis', format_key='format', format_default='netcdf', includes_product_type=True, uses_pressure_level=False, uses_time=True)

Profile for surface / single-level ERA5 reanalyses (and the fallback).

format=netcdf, product_type=reanalysis (source-configurable), area, year/month/day (all arrays), hourly time. Surface and land products reject pressure_level, so it is not emitted here — use :data:REANALYSIS_PRESSURE for pressure-level datasets.

AemetArchive 

AemetArchive(root: Path, source: AemetSource)

Append-only local mirror of AEMET station observations.

Everything (inventory + per-preset time series) lives under root as GeoParquet. Dense fields like future gridded AEMET products are out of scope for this class.

Parameters:

Name Type Description Default
root Path

Directory that will hold one file per preset.

 required
source AemetSource

:class:AemetSource used for fetches.

 required
Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/archive.py 
67
68
69
70
def __init__(self, root: Path, source: AemetSource) -> None:
    self.root = Path(root)
    self.source = source
    self.root.mkdir(parents=True, exist_ok=True)
sync_stations 
sync_stations() -> StationCollection

Refresh the cached station inventory and return it.

Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/archive.py 
74
75
76
77
78
def sync_stations(self) -> StationCollection:
    """Refresh the cached station inventory and return it."""
    collection = self.source.list_stations()
    self._write_stations(collection)
    return collection
load_stations 
load_stations() -> StationCollection

Return the inventory as a :class:StationCollection.

Refreshes from AEMET if no cached copy exists.

Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/archive.py 
80
81
82
83
84
85
86
87
88
89
90
91
def load_stations(self) -> StationCollection:
    """Return the inventory as a :class:`StationCollection`.

    Refreshes from AEMET if no cached copy exists.
    """
    path = self.stations_path
    if not path.is_file():
        return self.sync_stations()
    import geopandas as gpd

    gdf = gpd.read_parquet(path)
    return _stations_from_geodataframe(gdf)
load_stations_geodataframe 
load_stations_geodataframe()

Return the inventory as a :class:geopandas.GeoDataFrame.

Geometry is EPSG:4326 points. Refreshes from AEMET if no cached copy exists.

Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/archive.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def load_stations_geodataframe(self):
    """Return the inventory as a :class:`geopandas.GeoDataFrame`.

    Geometry is EPSG:4326 points. Refreshes from AEMET if no
    cached copy exists.
    """
    import geopandas as gpd

    if not self.stations_path.is_file():
        self.sync_stations()
    return gpd.read_parquet(self.stations_path)
sync 
sync(preset: str, *, stations: StationCollection | None = None, since: str | None = None, until: str | None = None) -> Dataset

Backfill or update a preset. Returns the freshly-fetched slice.

sync returns the in-memory xr.Dataset that was just fetched (useful for immediate inspection). The on-disk archive is written as a long-format GeoParquet file; call :meth:load to read it back as a :class:geopandas.GeoDataFrame.

Behaviour:

  • since defaults to one day after the archive's global latest stored time (the newest row across all stations), or the preset-specific floor (e.g. 1920 for daily / monthly) when the archive is empty. Per-station resume — so stations missing recent records get backfilled even when others are up-to-date — is not implemented; use an explicit since if you need that behaviour.
  • until defaults to now in UTC.
  • Re-running with the same window overwrites the overlapping slice; it is idempotent, not additive.
Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/archive.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def sync(
    self,
    preset: str,
    *,
    stations: StationCollection | None = None,
    since: str | None = None,
    until: str | None = None,
) -> xr.Dataset:
    """Backfill or update a preset. Returns the freshly-fetched slice.

    ``sync`` returns the in-memory ``xr.Dataset`` that was just
    fetched (useful for immediate inspection). The on-disk archive
    is written as a long-format GeoParquet file; call :meth:`load`
    to read it back as a :class:`geopandas.GeoDataFrame`.

    Behaviour:

    - ``since`` defaults to one day after the archive's **global**
      latest stored time (the newest row across all stations), or
      the preset-specific floor (e.g. 1920 for daily / monthly)
      when the archive is empty. Per-station resume — so stations
      missing recent records get backfilled even when others are
      up-to-date — is not implemented; use an explicit ``since``
      if you need that behaviour.
    - ``until`` defaults to ``now`` in UTC.
    - Re-running with the same window overwrites the overlapping
      slice; it is idempotent, not additive.
    """
    if preset not in _FETCH_TABLE:
        raise ValueError(
            f"unknown preset {preset!r}; expected one of {sorted(_FETCH_TABLE)}"
        )
    station_ids = self._station_ids(stations)
    start = self._resolve_start(preset, since)
    end = self._resolve_end(until)

    # Three cases for inverted windows (``start > end``):
    #
    # 1. Hourly preset — always fetch (endpoint is rolling 24h,
    #    ignores TimeRange entirely; stale data would otherwise
    #    accumulate until the next calendar day).
    # 2. Non-hourly + auto-resumed start (``since is None``) —
    #    ``_resolve_start`` set ``start = last + 1 day`` which
    #    overshoots ``end=now`` on routine same-day re-runs.
    #    Correct behaviour is a silent no-op.
    # 3. Non-hourly + caller-specified ``since`` that inverts —
    #    almost certainly a typo. Let ``TimeRange.parse`` raise
    #    so the user sees the error.
    inverted = pd.Timestamp(start, tz="UTC") > pd.Timestamp(end, tz="UTC")
    is_hourly = preset in _TIME_INSENSITIVE_PRESETS
    # Auto-resumed (since=None) + inverted + non-hourly → no-op.
    # Caller-specified inversion on a non-hourly preset → fall
    # through and let TimeRange.parse raise a clear error.
    if inverted and not is_hourly and since is None:
        return xr.Dataset()
    tr = None if (inverted and is_hourly) else TimeRange.parse(start, end)
    fresh = _FETCH_TABLE[preset](self.source, station_ids, tr)
    self._append(preset, fresh)
    return fresh
load 
load(preset: str)

Open the archive for preset as a :class:geopandas.GeoDataFrame.

The GeoDataFrame is long-format: one row per (station, time) sample, with the geometry column carrying each station's EPSG:4326 point.

Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/archive.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def load(self, preset: str):
    """Open the archive for ``preset`` as a :class:`geopandas.GeoDataFrame`.

    The GeoDataFrame is long-format: one row per
    ``(station, time)`` sample, with the ``geometry`` column
    carrying each station's EPSG:4326 point.
    """
    import geopandas as gpd

    path = self._preset_path(preset)
    if not path.exists():
        raise FileNotFoundError(
            f"archive for {preset!r} does not exist at {path}; "
            "run .sync(preset) first."
        )
    return gpd.read_parquet(path)
load_dataset 
load_dataset(preset: str) -> Dataset

Open the archive for preset as an (station, time) xarray dataset.

Convenience for workflows that prefer the cube view over the long-format GeoDataFrame.

Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/archive.py 
211
212
213
214
215
216
217
218
def load_dataset(self, preset: str) -> xr.Dataset:
    """Open the archive for ``preset`` as an ``(station, time)`` xarray dataset.

    Convenience for workflows that prefer the cube view over the
    long-format GeoDataFrame.
    """
    gdf = self.load(preset)
    return _geodataframe_to_dataset(gdf)
coverage 
coverage(preset: str) -> list[ArchiveCoverage]

Return per-station first / last / gap statistics for preset.

Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/archive.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
def coverage(self, preset: str) -> list[ArchiveCoverage]:
    """Return per-station first / last / gap statistics for ``preset``."""
    try:
        gdf = self.load(preset)
    except FileNotFoundError:
        return []

    # Choose a representative variable for gap counting; fall back
    # to the first value column if none of the usual suspects are
    # present.
    value_cols = _value_columns(gdf)
    pick = next(
        (
            v
            for v in ("air_temperature_daily_mean", "air_temperature")
            if v in value_cols
        ),
        value_cols[0] if value_cols else None,
    )
    all_times = np.sort(pd.to_datetime(gdf["time"].unique()))
    n_time = len(all_times)

    out: list[ArchiveCoverage] = []
    grouped = gdf.groupby("station_id", sort=True)
    for sid, sub in grouped:
        if pick is None or n_time == 0:
            out.append(
                ArchiveCoverage(
                    preset=preset,
                    station_id=str(sid),
                    first=None,
                    last=None,
                    n_timesteps=0,
                    gap_fraction=1.0,
                )
            )
            continue
        valid = sub[sub[pick].notna()]
        n_valid = len(valid)
        if n_valid == 0:
            out.append(
                ArchiveCoverage(
                    preset=preset,
                    station_id=str(sid),
                    first=None,
                    last=None,
                    n_timesteps=0,
                    gap_fraction=1.0,
                )
            )
            continue
        times = pd.to_datetime(valid["time"])
        # ``Series.min`` / ``max`` are typed as ``Timestamp | NaTType``;
        # we've already filtered to non-null rows so NaT isn't possible,
        # but ty doesn't know that.
        first_ts = cast("pd.Timestamp | None", pd.Timestamp(times.min()))
        last_ts = cast("pd.Timestamp | None", pd.Timestamp(times.max()))
        out.append(
            ArchiveCoverage(
                preset=preset,
                station_id=str(sid),
                first=first_ts,
                last=last_ts,
                n_timesteps=n_valid,
                gap_fraction=1.0 - n_valid / n_time,
            )
        )
    return out

AemetAuthError

Bases: AemetError

Missing or rejected API key.

AemetError

Bases: RuntimeError

Any AEMET-level failure (auth, rate limit, bad envelope, 5xx).

AemetRateLimitError

Bases: AemetError

Quota exhausted even after backoff.

AemetSource 

AemetSource(credentials: AEMETCredentials | None = None, client: Any | None = None, base_url: str = AEMET_BASE_URL, timeout_s: float = DEFAULT_TIMEOUT_S, max_retries: int = DEFAULT_MAX_RETRIES, max_workers: int = DEFAULT_MAX_WORKERS, min_interval_s: float = 0.0)

Bases: DataSource

Adapter for AEMET OpenData observation endpoints.

Parameters:

Name Type Description Default
credentials AEMETCredentials | None

Explicit credentials. When None, resolved from the environment / .env / ~/.aemet via :func:~xrproducts.load_aemet.

 None
client Any | None

Optional pre-built HTTP client (must expose a get(url, timeout=...) method returning an object with .status_code, .headers, .json(), .raise_for_status() — compatible with httpx.Client and test doubles).

 None
base_url str

Override the API root (useful for staging / mocks).

 AEMET_BASE_URL
timeout_s float

Per-request timeout.

 DEFAULT_TIMEOUT_S
max_retries int

Retries for 429 / transient network errors.

 DEFAULT_MAX_RETRIES
max_workers int

Thread pool size for parallel station fetches.

 DEFAULT_MAX_WORKERS
min_interval_s float

Minimum seconds between consecutive HTTP calls, enforced globally across the worker pool. AEMET's rate limit is ~150 requests per rolling minute — 0.5 (120 req/min) leaves comfortable headroom and avoids the burst-detection revocation we saw at max_workers=8. Set to 0.0 to disable pacing.

 0.0
Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/source.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def __init__(
    self,
    credentials: AEMETCredentials | None = None,
    client: Any | None = None,
    base_url: str = AEMET_BASE_URL,
    timeout_s: float = DEFAULT_TIMEOUT_S,
    max_retries: int = DEFAULT_MAX_RETRIES,
    max_workers: int = DEFAULT_MAX_WORKERS,
    min_interval_s: float = 0.0,
) -> None:
    self.credentials = credentials or load_aemet()
    self._client = client
    self.base_url = base_url.rstrip("/")
    self.timeout_s = timeout_s
    self.max_retries = max_retries
    self.max_workers = max_workers
    self.min_interval_s = max(0.0, min_interval_s)
    # Global pacing lock — a monotonic timestamp of the last
    # outbound HTTP call, shared across all worker threads.
    self._pace_lock = threading.Lock()
    self._pace_last = 0.0
    # Global 429 backoff — when any worker hits a rate limit, all
    # workers (including that one on its retry) wait until this
    # deadline. Without this shared pause, other workers keep the
    # bucket topped up while the 429'd worker backs off, so the
    # minute window never actually clears.
    self._rate_limited_until = 0.0
    # Scale factor on the 429 global-pause schedule. Production
    # runs want full minute-scale pauses to let AEMET's minute
    # bucket drain; tests set this to 0 to avoid actually sleeping.
    self.rate_limit_pause_scale = 1.0
download 
download(dataset_id: str, output: Path, *, variables: list[str | Variable] | None = None, bbox: BBox | None = None, time: TimeRange | None = None, depth: DepthRange | None = None, levels: PressureLevels | None = None, **extras: Any) -> Path

Download dataset_id to a NetCDF file at output.

Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/source.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def download(
    self,
    dataset_id: str,
    output: Path,
    *,
    variables: list[str | Variable] | None = None,
    bbox: BBox | None = None,
    time: TimeRange | None = None,
    depth: DepthRange | None = None,
    levels: PressureLevels | None = None,
    **extras: Any,
) -> Path:
    """Download ``dataset_id`` to a NetCDF file at ``output``."""
    ds = self.open(
        dataset_id,
        variables=variables,
        bbox=bbox,
        time=time,
        depth=depth,
        levels=levels,
        **extras,
    )
    output = Path(output)
    output.parent.mkdir(parents=True, exist_ok=True)
    ds.to_netcdf(output)
    return output
open 
open(dataset_id: str, *, variables: list[str | Variable] | None = None, bbox: BBox | None = None, time: TimeRange | None = None, depth: DepthRange | None = None, levels: PressureLevels | None = None, stations: Iterable[str] | StationCollection | None = None, **extras: Any) -> Dataset

Open dataset_id as an (station, time) xarray Dataset.

dataset_id is one of the preset IDs from :mod:~xrproducts._src.aemet.catalog (aemet_daily, aemet_hourly, aemet_monthly, aemet_pollution, aemet_stations, aemet_normals, aemet_extremes).

Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/source.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
def open(
    self,
    dataset_id: str,
    *,
    variables: list[str | Variable] | None = None,
    bbox: BBox | None = None,
    time: TimeRange | None = None,
    depth: DepthRange | None = None,
    levels: PressureLevels | None = None,
    stations: Iterable[str] | StationCollection | None = None,
    **extras: Any,
) -> xr.Dataset:
    """Open ``dataset_id`` as an ``(station, time)`` xarray Dataset.

    ``dataset_id`` is one of the preset IDs from
    :mod:`~xrproducts._src.aemet.catalog` (``aemet_daily``,
    ``aemet_hourly``, ``aemet_monthly``, ``aemet_pollution``,
    ``aemet_stations``, ``aemet_normals``, ``aemet_extremes``).
    """
    info = self.describe(dataset_id)
    kind = info.extras.get("aemet_kind", dataset_id)
    known = {
        "aemet_stations",
        "aemet_daily",
        "aemet_hourly",
        "aemet_monthly",
        "aemet_normals",
        "aemet_extremes",
        "aemet_pollution",
    }
    if kind not in known:
        raise ValueError(f"unknown AEMET dataset: {dataset_id}")

    # The inventory preset doesn't need station IDs — it *is* the
    # station list. All other presets require at least one station
    # ID or a BBox to derive them from.
    if kind == "aemet_stations":
        return _stations_to_dataset(self.list_stations(bbox=bbox))

    station_ids = self._resolve_station_ids(stations, bbox)

    if kind == "aemet_daily":
        return self.get_daily(station_ids, time=time, variables=variables)
    if kind == "aemet_hourly":
        return self.get_hourly(station_ids, time=time, variables=variables)
    if kind == "aemet_monthly":
        return self.get_monthly(station_ids, time=time, variables=variables)
    if kind == "aemet_normals":
        return self.get_normals(station_ids, variables=variables)
    if kind == "aemet_extremes":
        parameter = str(extras.get("parameter", "T"))
        return self.get_extremes(station_ids, parameter=parameter)
    # kind == "aemet_pollution"
    return self.get_pollution(station_ids, time=time, variables=variables)
list_stations 
list_stations(bbox: BBox | None = None) -> StationCollection

Return the full AEMET station inventory, optionally filtered.

Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/source.py 
323
324
325
326
327
328
329
330
331
332
def list_stations(self, bbox: BBox | None = None) -> StationCollection:
    """Return the full AEMET station inventory, optionally filtered."""
    payload = self._get_json(
        "/valores/climatologicos/inventarioestaciones/todasestaciones"
    )
    stations = tuple(_parse_station(row) for row in payload)
    collection = StationCollection(stations)
    if bbox is not None:
        collection = collection.within(bbox)
    return collection
get_daily 
get_daily(stations: Iterable[str], *, time: TimeRange | None = None, variables: list[str | Variable] | None = None) -> Dataset

Fetch daily climatology for stations in time.

Automatically chunks each station's window into 180-day slices (AEMET's undocumented per-request cap) and fetches in parallel.

Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/source.py 
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
def get_daily(
    self,
    stations: Iterable[str],
    *,
    time: TimeRange | None = None,
    variables: list[str | Variable] | None = None,
) -> xr.Dataset:
    """Fetch daily climatology for ``stations`` in ``time``.

    Automatically chunks each station's window into 180-day slices
    (AEMET's undocumented per-request cap) and fetches in parallel.
    """
    station_ids = tuple(stations)
    if not station_ids:
        raise ValueError("at least one station id required")
    if time is None:
        raise ValueError("daily endpoint requires a TimeRange")

    start_dt = time.start.to_pydatetime()
    end_dt = time.end.to_pydatetime()

    frames = self._fetch_station_windows(
        station_ids,
        start_dt,
        end_dt,
        chunk_days=DAILY_WINDOW_DAYS,
        url_fn=_daily_url,
        parse_fn=_parse_daily_rows,
    )
    full_index = pd.date_range(start_dt.date(), end_dt.date(), freq="1D", tz=UTC)
    ds = _frames_to_stations_dataset(
        frames,
        station_ids=station_ids,
        time_index=full_index,
        field_map=DAILY_FIELDS,
        passthrough=DAILY_PASSTHROUGH_FIELDS,
        endpoint_name="daily",
    )
    if variables:
        ds = _subset_variables(ds, variables)
    return ds
get_hourly 
get_hourly(stations: Iterable[str], *, time: TimeRange | None = None, variables: list[str | Variable] | None = None) -> Dataset

Fetch the latest ~24 h of hourly observations for stations.

The AEMET hourly endpoint is rolling — it returns whatever window of recent observations the station has published (typically the last 24 hours). There is no historical-window query, but time (when given) trims the returned dataset to the requested window.

Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/source.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
def get_hourly(
    self,
    stations: Iterable[str],
    *,
    time: TimeRange | None = None,
    variables: list[str | Variable] | None = None,
) -> xr.Dataset:
    """Fetch the latest ~24 h of hourly observations for ``stations``.

    The AEMET hourly endpoint is rolling — it returns whatever window
    of recent observations the station has published (typically the
    last 24 hours). There is no historical-window query, but ``time``
    (when given) trims the returned dataset to the requested window.
    """
    station_ids = tuple(stations)
    if not station_ids:
        raise ValueError("at least one station id required")

    def fetch(sid: str) -> list[dict[str, Any]]:
        return self._get_json(f"/observacion/convencional/datos/estacion/{sid}")

    rows_by_station: dict[str, list[dict[str, Any]]] = {}
    times_seen: set[datetime] = set()
    with ThreadPoolExecutor(max_workers=self.max_workers) as pool:
        futures = {pool.submit(fetch, sid): sid for sid in station_ids}
        for fut in as_completed(futures):
            sid = futures[fut]
            try:
                rows = fut.result() or []
            except AemetNoDataError:
                rows = []
            except (AemetAuthError, AemetRateLimitError):
                # Auth / quota failures are not per-chunk
                # problems — they fail the whole scrape, so we
                # let them propagate instead of silently
                # skipping every remaining station.
                raise
            except AemetError as exc:
                warnings.warn(
                    f"AEMET hourly failed for station {sid}: {exc}",
                    stacklevel=2,
                )
                rows = []
            rows_by_station[sid] = rows
            for row in rows:
                if "fint" in row:
                    times_seen.add(parse_aemet_datetime(row["fint"]))

    if not times_seen:
        raise AemetError("AEMET returned no hourly observations")
    time_index = pd.DatetimeIndex(sorted(times_seen), tz=UTC)

    frames: dict[str, pd.DataFrame] = {}
    for sid, rows in rows_by_station.items():
        if not rows:
            continue
        frames[sid] = _hourly_frame(rows)

    ds = _frames_to_stations_dataset(
        frames,
        station_ids=station_ids,
        time_index=time_index,
        field_map=HOURLY_FIELDS,
        passthrough=frozenset(),
        endpoint_name="hourly",
    )
    if time is not None and "time" in ds.coords:
        # ds time is naive UTC datetime64; match the TimeRange bounds.
        lo = time.start.tz_convert(UTC).tz_localize(None)
        hi = time.end.tz_convert(UTC).tz_localize(None)
        ds = ds.sel(time=slice(lo, hi))
    if variables:
        ds = _subset_variables(ds, variables)
    return ds
get_monthly 
get_monthly(stations: Iterable[str], *, time: TimeRange | None = None, variables: list[str | Variable] | None = None) -> Dataset

Fetch monthly aggregates for stations over time.

Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/source.py 
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
def get_monthly(
    self,
    stations: Iterable[str],
    *,
    time: TimeRange | None = None,
    variables: list[str | Variable] | None = None,
) -> xr.Dataset:
    """Fetch monthly aggregates for ``stations`` over ``time``."""
    station_ids = tuple(stations)
    if not station_ids:
        raise ValueError("at least one station id required")
    if time is None:
        raise ValueError("monthly endpoint requires a TimeRange")

    year_start = time.start.year
    year_end = time.end.year

    def fetch(sid: str, y1: int, y2: int) -> list[dict[str, Any]]:
        return self._get_json(
            f"/valores/climatologicos/mensualesanuales/datos/"
            f"anioini/{y1}/aniofin/{y2}/estacion/{sid}"
        )

    frames: dict[str, pd.DataFrame] = {}
    with ThreadPoolExecutor(max_workers=self.max_workers) as pool:
        futures = {}
        for sid in station_ids:
            for y1, y2 in _chunk_years(year_start, year_end, MONTHLY_WINDOW_YEARS):
                futures[pool.submit(fetch, sid, y1, y2)] = sid
        for fut in as_completed(futures):
            sid = futures[fut]
            try:
                rows = fut.result() or []
            except AemetNoDataError:
                # Chunk empty for this station (e.g. started after
                # the request window). Skip — a gap is still a
                # valid, observable outcome (full time axis + NaN).
                continue
            except (AemetAuthError, AemetRateLimitError):
                # Auth / quota failures are not per-chunk
                # problems — they fail the whole scrape, so we
                # let them propagate instead of silently
                # skipping every remaining station.
                raise
            except AemetError as exc:
                # Hard error on one chunk shouldn't nuke the
                # whole scrape; log and continue so the archive
                # gets whatever data we could reach. Re-running
                # later resumes from the archive's latest time.
                warnings.warn(
                    f"AEMET chunk failed for station {sid}: {exc}",
                    stacklevel=2,
                )
                continue
            if not rows:
                continue
            frame = _monthly_frame(rows)
            if sid in frames:
                frames[sid] = pd.concat([frames[sid], frame])
            else:
                frames[sid] = frame

    # Fetch chunks cover whole years (AEMET endpoint granularity),
    # but return only the months the caller actually asked for.
    # Without this trim, ``TimeRange.parse("2024-06", "2024-09")``
    # would still return Jan-Dec 2024 and incremental archive
    # syncs would repeatedly rewrite entire years.
    req_start = time.start.tz_convert(UTC) if time.start.tz else time.start
    req_end = time.end.tz_convert(UTC) if time.end.tz else time.end
    month_start = pd.Timestamp(
        year=req_start.year, month=req_start.month, day=1, tz=UTC
    )
    month_end = pd.Timestamp(year=req_end.year, month=req_end.month, day=1, tz=UTC)
    time_index = pd.date_range(month_start, month_end, freq="MS", tz=UTC)

    ds = _frames_to_stations_dataset(
        frames,
        station_ids=station_ids,
        time_index=time_index,
        field_map=MONTHLY_FIELDS,
        passthrough=frozenset(),
        endpoint_name="monthly",
    )
    if variables:
        ds = _subset_variables(ds, variables)
    return ds
get_normals 
get_normals(stations: Iterable[str], *, variables: list[str | Variable] | None = None) -> Dataset

Fetch 1981–2010 climate normals for stations.

Emitted as a (station, month) dataset with month=1..12. variables selects a subset of the normals fields.

Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/source.py 
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
def get_normals(
    self,
    stations: Iterable[str],
    *,
    variables: list[str | Variable] | None = None,
) -> xr.Dataset:
    """Fetch 1981–2010 climate normals for ``stations``.

    Emitted as a ``(station, month)`` dataset with month=1..12.
    ``variables`` selects a subset of the normals fields.
    """
    station_ids = tuple(stations)

    def fetch(sid: str) -> list[dict[str, Any]]:
        return self._get_json(f"/valores/climatologicos/normales/estacion/{sid}")

    rows_per: dict[str, list[dict[str, Any]]] = {}
    with ThreadPoolExecutor(max_workers=self.max_workers) as pool:
        futures = {pool.submit(fetch, sid): sid for sid in station_ids}
        for fut in as_completed(futures):
            sid = futures[fut]
            # Match the per-chunk tolerance used by daily / monthly
            # / pollution: a single station's missing payload
            # shouldn't abort a many-station batch.
            try:
                rows_per[sid] = fut.result() or []
            except AemetNoDataError:
                rows_per[sid] = []
            except (AemetAuthError, AemetRateLimitError):
                raise
            except AemetError as exc:
                warnings.warn(
                    f"AEMET normals failed for station {sid}: {exc}",
                    stacklevel=2,
                )
                rows_per[sid] = []
    ds = _normals_to_dataset(rows_per, station_ids)
    if variables:
        ds = _subset_variables(ds, variables)
    return ds
get_extremes 
get_extremes(stations: Iterable[str], *, parameter: str = 'T') -> Dataset

Fetch record extremes (P=precip, T=temp, V=wind) per station.

Extremes are a ragged, per-field set of records. We emit them as a minimal (station,) dataset with the raw dictionaries kept in attrs['records'] rather than guessing a schema that may vary per station.

Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/source.py 
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
def get_extremes(
    self, stations: Iterable[str], *, parameter: str = "T"
) -> xr.Dataset:
    """Fetch record extremes (P=precip, T=temp, V=wind) per station.

    Extremes are a ragged, per-field set of records. We emit them as
    a minimal ``(station,)`` dataset with the raw dictionaries kept
    in ``attrs['records']`` rather than guessing a schema that may
    vary per station.
    """
    station_ids = tuple(stations)
    if parameter not in {"P", "T", "V"}:
        raise ValueError(f"parameter must be one of P/T/V, got {parameter!r}")

    def fetch(sid: str) -> list[dict[str, Any]]:
        return self._get_json(
            f"/valores/climatologicos/valoresextremos/"
            f"parametro/{parameter}/estacion/{sid}"
        )

    rows_per: dict[str, list[dict[str, Any]]] = {}
    with ThreadPoolExecutor(max_workers=self.max_workers) as pool:
        futures = {pool.submit(fetch, sid): sid for sid in station_ids}
        for fut in as_completed(futures):
            sid = futures[fut]
            try:
                rows_per[sid] = fut.result() or []
            except AemetNoDataError:
                rows_per[sid] = []
            except (AemetAuthError, AemetRateLimitError):
                raise
            except AemetError as exc:
                warnings.warn(
                    f"AEMET extremes failed for station {sid}: {exc}",
                    stacklevel=2,
                )
                rows_per[sid] = []
    return _extremes_to_dataset(rows_per, station_ids, parameter)
get_pollution 
get_pollution(stations: Iterable[str], *, time: TimeRange | None = None, variables: list[str | Variable] | None = None) -> Dataset

Fetch EMEP background-pollution time series for stations.

AEMET returns irregular hourly/daily mixed samples — we pass them through as (station, time) keeping whatever time axis the payload reports. variables selects a subset of the pollutant fields.

Source code in packages/xrtoolz-products/src/xrproducts/_src/aemet/source.py 
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
def get_pollution(
    self,
    stations: Iterable[str],
    *,
    time: TimeRange | None = None,
    variables: list[str | Variable] | None = None,
) -> xr.Dataset:
    """Fetch EMEP background-pollution time series for ``stations``.

    AEMET returns irregular hourly/daily mixed samples — we pass
    them through as ``(station, time)`` keeping whatever time axis
    the payload reports. ``variables`` selects a subset of the
    pollutant fields.
    """
    station_ids = tuple(stations)

    def fetch(sid: str) -> list[dict[str, Any]]:
        return self._get_json(f"/red/especial/contaminacionfondo/estacion/{sid}")

    frames: dict[str, pd.DataFrame] = {}
    all_times: set[datetime] = set()
    with ThreadPoolExecutor(max_workers=self.max_workers) as pool:
        futures = {pool.submit(fetch, sid): sid for sid in station_ids}
        for fut in as_completed(futures):
            sid = futures[fut]
            try:
                rows = fut.result() or []
            except AemetNoDataError:
                continue
            except (AemetAuthError, AemetRateLimitError):
                # Auth / quota failures are not per-chunk
                # problems — they fail the whole scrape, so we
                # let them propagate instead of silently
                # skipping every remaining station.
                raise
            except AemetError as exc:
                warnings.warn(
                    f"AEMET pollution failed for station {sid}: {exc}",
                    stacklevel=2,
                )
                continue
            if not rows:
                continue
            frame = _pollution_frame(rows)
            frames[sid] = frame
            for ts in pd.DatetimeIndex(frame.index):
                all_times.add(ts.to_pydatetime())

    if time is not None and all_times:
        all_times = {
            t
            for t in all_times
            if time.start.to_pydatetime() <= t <= time.end.to_pydatetime()
        }
    time_index = pd.DatetimeIndex(sorted(all_times), tz=UTC)
    ds = _pollution_to_dataset(frames, station_ids, time_index)
    if variables:
        ds = _subset_variables(ds, variables)
    return ds

ArchiveCoverage dataclass 

ArchiveCoverage(preset: str, station_id: str, first: Timestamp | None, last: Timestamp | None, n_timesteps: int, gap_fraction: float)

Per-station span of what the archive already holds for a preset.

DatasetInfo dataclass 

DatasetInfo(dataset_id: str, source: str, title: str, kind: DatasetKind = GRIDDED, variables: tuple[Variable, ...] = (), spatial_coverage: BBox | None = None, temporal_coverage: tuple[str, str] | None = None, doi: str | None = None, license: str | None = None, notes: str | None = None, extras: dict[str, Any] = dict(), form_profile: Any = None)

Static description of a remote dataset.

Adapters return these from :meth:DataSource.describe and the catalog stores them alongside default request parameters.

DatasetKind

Bases: StrEnum

Discriminator for how a dataset is laid out once materialized.

  • gridded — regular lon/lat (+time) cube, the common case.
  • alongtrack — 1-D samples along a satellite ground track.
  • profiles — collection of vertical profiles at scattered locations (Argo floats, CTD casts, gliders).
  • trajectory — drifters / moving platforms.
  • stations — fixed points, one time-series per station. Maps to CF's featureType = "timeSeries" (CF 9.3) with dims (station, time) and per-station coords.
  • swath — satellite L2 swath on (time, scanline, ground_pixel) with 2-D lon/lat coordinates (TROPOMI).
  • scene — single L2 scene / plume raster on (y, x) with 2-D lon/lat coordinates (EMIT, GHGSat).

Downstream code (regridding, masking, plotting) branches on this; e.g. antimeridian BBox wrapping is meaningless for profiles.

DataSource

Bases: ABC

Abstract interface implemented by every adapter.

source_id instance-attribute 
source_id: str

Short slug used to look up per-source aliases in :class:Variable.

list_datasets abstractmethod 
list_datasets() -> list[DatasetInfo]

Return known datasets exposed by this source.

Source code in packages/xrtoolz-products/src/xrproducts/_src/base.py 
84
85
86
@abstractmethod
def list_datasets(self) -> list[DatasetInfo]:  # pragma: no cover - interface
    """Return known datasets exposed by this source."""
describe abstractmethod 
describe(dataset_id: str) -> DatasetInfo

Return a :class:DatasetInfo for dataset_id.

Source code in packages/xrtoolz-products/src/xrproducts/_src/base.py 
88
89
90
@abstractmethod
def describe(self, dataset_id: str) -> DatasetInfo:  # pragma: no cover - interface
    """Return a :class:`DatasetInfo` for ``dataset_id``."""
download abstractmethod 
download(dataset_id: str, output: Path, *, variables: list[str | Variable] | None = None, bbox: BBox | None = None, time: TimeRange | None = None, depth: DepthRange | None = None, levels: PressureLevels | None = None, **extras: Any) -> Path

Download dataset_id to output and return the path.

Source code in packages/xrtoolz-products/src/xrproducts/_src/base.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@abstractmethod
def download(
    self,
    dataset_id: str,
    output: Path,
    *,
    variables: list[str | Variable] | None = None,
    bbox: BBox | None = None,
    time: TimeRange | None = None,
    depth: DepthRange | None = None,
    levels: PressureLevels | None = None,
    **extras: Any,
) -> Path:  # pragma: no cover - interface
    """Download ``dataset_id`` to ``output`` and return the path."""
open abstractmethod 
open(dataset_id: str, *, variables: list[str | Variable] | None = None, bbox: BBox | None = None, time: TimeRange | None = None, depth: DepthRange | None = None, levels: PressureLevels | None = None, **extras: Any) -> Dataset

Open dataset_id lazily as an :class:xarray.Dataset.

Source code in packages/xrtoolz-products/src/xrproducts/_src/base.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
@abstractmethod
def open(
    self,
    dataset_id: str,
    *,
    variables: list[str | Variable] | None = None,
    bbox: BBox | None = None,
    time: TimeRange | None = None,
    depth: DepthRange | None = None,
    levels: PressureLevels | None = None,
    **extras: Any,
) -> xr.Dataset:  # pragma: no cover - interface
    """Open ``dataset_id`` lazily as an :class:`xarray.Dataset`."""

CatalogEntry dataclass 

CatalogEntry(source: str, dataset_id: str, defaults: dict[str, Any] | None = None)

Short-name-to-source mapping with optional request defaults.

CDSFormProfile dataclass 

CDSFormProfile(family: str, format_key: str = 'format', format_default: str = 'netcdf', fixed: Mapping[str, Any] = dict(), includes_product_type: bool = False, uses_area: bool = True, uses_pressure_level: bool = False, year_is_array: bool = True, uses_time: bool = False, required_extras: tuple[str, ...] = tuple())

Shape of a CDS retrieve() form for a dataset family.

Attributes:

Name Type Description
family str

Short slug identifying the family ("reanalysis", "insitu-land", "insitu-marine", ...). Used in logs / tests.
format_key str

Name of the form field that carries the output format. "format" for reanalysis; "data_format" for CDS in-situ (the API uses different names).
format_default str

Default value for format_key when the caller / source don't set it. "netcdf" for reanalysis; "csv" for in-situ.
fixed Mapping[str, Any]

Form keys this family always sends with a fixed value (e.g. {"version": "2_0_0"} for in-situ). Overridable via **extras.
includes_product_type bool

Whether product_type is part of the form.
uses_area bool

Whether bbox serialises to area.
uses_pressure_level bool

Whether :class:PressureLevels serialises to pressure_level.
year_is_array bool

Whether year accepts a list. In-situ products take year as a single string — time must be within one calendar year per request. The :class:~xrproducts._src.cds.archive.CDSInsituArchive chunks by year to respect this.
uses_time bool

Whether the family expects an hourly time form key. ERA5 reanalyses are hourly products and reject a request that sends only year/month/day; the adapter derives the hours from the request :class:TimeRange.
required_extras tuple[str, ...]

Keys the caller must supply via **extras (e.g. "time_aggregation" for in-situ-land).

CDSInsituArchive 

CDSInsituArchive(root: Path, preset: str, source: CDSSource | None = None, *, time_aggregation: str = 'daily', variables: tuple[Variable, ...] | None = None)

Append-only local mirror of a CDS in-situ preset.

Parameters:

Name Type Description Default
root Path

Directory that will hold one subtree per preset.

 required
preset str

Either "cds_insitu_land" or "cds_insitu_marine".

 required
source CDSSource | None

:class:CDSSource used for downloads. If None a default one is constructed (env-var / .env credentials).

 None
time_aggregation str

One of "sub_daily", "daily", "monthly" — determines the CDS form key of the same name and the on-disk partition cadence.

 'daily'
variables tuple[Variable, ...] | None

Restrict fetches to these variables. None fetches every variable advertised in the preset's DatasetInfo — CDS in-situ rejects requests with no variable key, so we fall back rather than send an empty list.

 None
Source code in packages/xrtoolz-products/src/xrproducts/_src/cds/archive.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def __init__(
    self,
    root: Path,
    preset: str,
    source: CDSSource | None = None,
    *,
    time_aggregation: str = "daily",
    variables: tuple[Variable, ...] | None = None,
) -> None:
    if preset not in PRESET_TO_DATASET:
        raise ValueError(
            f"unknown preset {preset!r}; expected one of "
            f"{sorted(PRESET_TO_DATASET)}"
        )
    if time_aggregation not in _VALID_TIME_AGGREGATIONS:
        raise ValueError(
            f"unknown time_aggregation {time_aggregation!r}; expected "
            f"one of {sorted(_VALID_TIME_AGGREGATIONS)}"
        )
    self.root = Path(root)
    self.preset = preset
    self.dataset_id = PRESET_TO_DATASET[preset]
    self.source = source if source is not None else CDSSource()
    # Marine ignores ``time_aggregation``; we keep the attribute
    # for bookkeeping (manifest / path derivation) but don't send
    # it on marine requests.
    self.time_aggregation = time_aggregation
    self.variables = variables
sync 
sync(start: str | Timestamp, end: str | Timestamp, *, bbox: BBox | None = None, since: str | Timestamp | None = None, overwrite: bool = False) -> DataFrame

Fetch year-chunks spanning [start, end] and append to disk.

Parameters:

Name Type Description Default
start str | Timestamp

Earliest year to fetch. Inclusive.

 required
end str | Timestamp

Latest year to fetch. Inclusive.

 required
bbox BBox | None

Optional server-side spatial filter. CDS in-situ accepts an area key, so we forward this on the request rather than filtering after download. The archive records the bbox as part of the manifest's "scope" and refuses to reuse partitions fetched under a different bbox — re-run on a fresh root if you want to change scope mid-archive.

 None
since str | Timestamp | None

If given, skip years strictly earlier than this and force a re-fetch of years at/after the cutoff even if they're already in the manifest (targeted refresh).

 None
overwrite bool

Re-download every year in the window regardless of the manifest.

 False

Returns:

Type Description
DataFrame

The concatenated freshly-fetched data (empty if everything
DataFrame

was already cached).
Source code in packages/xrtoolz-products/src/xrproducts/_src/cds/archive.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def sync(
    self,
    start: str | pd.Timestamp,
    end: str | pd.Timestamp,
    *,
    bbox: BBox | None = None,
    since: str | pd.Timestamp | None = None,
    overwrite: bool = False,
) -> pd.DataFrame:
    """Fetch year-chunks spanning ``[start, end]`` and append to disk.

    Args:
        start: Earliest year to fetch. Inclusive.
        end: Latest year to fetch. Inclusive.
        bbox: Optional server-side spatial filter. CDS in-situ
            accepts an ``area`` key, so we forward this on the
            request rather than filtering after download. The
            archive records the bbox as part of the manifest's
            "scope" and refuses to reuse partitions fetched under
            a different bbox — re-run on a fresh root if you want
            to change scope mid-archive.
        since: If given, skip years strictly earlier than this and
            force a re-fetch of years at/after the cutoff even if
            they're already in the manifest (targeted refresh).
        overwrite: Re-download every year in the window regardless
            of the manifest.

    Returns:
        The concatenated freshly-fetched data (empty if everything
        was already cached).
    """
    # CDS in-situ always takes one year per request (land + marine).
    year_chunks = _year_chunks(start, end)

    manifest = self._load_manifest()
    current_scope = _scope_fingerprint(
        bbox,
        self.variables,
        self.time_aggregation if self.preset == "cds_insitu_land" else None,
    )
    stored_scope = manifest.get("scope")
    scope_changed = stored_scope is not None and stored_scope != current_scope
    if scope_changed and not overwrite:
        raise ValueError(
            f"archive at {self.preset_root} was previously synced with "
            f"scope {stored_scope!r}; this call uses {current_scope!r}. "
            "Re-running with a different bbox / variables would silently "
            "leave already-marked years partial — rerun on a fresh root "
            "or call .sync(..., overwrite=True) to re-pull everything "
            "under the new scope."
        )
    if scope_changed:
        # overwrite=True under a new scope: reset stale state so neither
        # the manifest nor the parquet retains rows from the old scope.
        # ``_append`` merges by (station, time, variable), so without
        # clearing the data file, rows outside the new bbox / variable
        # set would survive a narrower re-sync.
        manifest["completed_chunks"] = []
        self.data_path.unlink(missing_ok=True)
        self.stations_path.unlink(missing_ok=True)
    done = set() if overwrite else set(manifest.get("completed_chunks", []))
    since_yr = _yr(since) if since is not None else None

    fetched: list[pd.DataFrame] = []
    for year_a, year_b in year_chunks:
        key = _chunk_key(year_a, year_b)
        # Drop years entirely before the ``since`` cutoff.
        if since_yr is not None and year_b < since_yr:
            continue
        # Skip "done" only when ``since`` wasn't supplied. With
        # ``since`` set the user is explicitly asking for a
        # targeted refresh of recent years.
        if since_yr is None and key in done:
            continue
        df = self._fetch_chunk(year_a, year_b, bbox=bbox)
        if not df.empty:
            self._append(df)
            fetched.append(df)
        # ``setdefault`` + ``append`` can duplicate entries when
        # ``overwrite=True`` re-fetches an already-listed year —
        # track completions as a sorted unique list instead.
        done_list = set(manifest.get("completed_chunks", []))
        done_list.add(key)
        manifest["completed_chunks"] = sorted(done_list)
        manifest["last_sync_utc"] = datetime.now(UTC).isoformat()
        manifest["preset"] = self.preset
        manifest["time_aggregation"] = self.time_aggregation
        manifest["scope"] = current_scope
        self._write_manifest(manifest)

    return pd.concat(fetched, ignore_index=True) if fetched else pd.DataFrame()
load 
load(start: str | Timestamp | None = None, end: str | Timestamp | None = None)

Open the archive as a :class:geopandas.GeoDataFrame.

Optionally filter to [start, end] inclusive.

Source code in packages/xrtoolz-products/src/xrproducts/_src/cds/archive.py 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
def load(
    self,
    start: str | pd.Timestamp | None = None,
    end: str | pd.Timestamp | None = None,
):
    """Open the archive as a :class:`geopandas.GeoDataFrame`.

    Optionally filter to ``[start, end]`` inclusive.
    """
    import geopandas as gpd

    if not self.data_path.exists():
        raise FileNotFoundError(
            f"archive for {self.preset!r} is empty; run .sync(...) first."
        )
    gdf = gpd.read_parquet(self.data_path)
    if start is not None:
        gdf = gdf[gdf["time"] >= _as_utc(start)]
    if end is not None:
        gdf = gdf[gdf["time"] <= _as_utc(end)]
    return gdf
load_dataset 
load_dataset(start: str | Timestamp | None = None, end: str | Timestamp | None = None) -> Dataset

Open the archive as an (station, time) xarray Dataset.

Source code in packages/xrtoolz-products/src/xrproducts/_src/cds/archive.py 
307
308
309
310
311
312
313
314
def load_dataset(
    self,
    start: str | pd.Timestamp | None = None,
    end: str | pd.Timestamp | None = None,
) -> xr.Dataset:
    """Open the archive as an ``(station, time)`` xarray Dataset."""
    gdf = self.load(start=start, end=end)
    return _long_to_dataset(gdf, source=self.preset)
load_stations 
load_stations()

Distinct (station_id, lon, lat) inventory as a GeoDataFrame.

Source code in packages/xrtoolz-products/src/xrproducts/_src/cds/archive.py 
316
317
318
319
320
321
322
323
324
def load_stations(self):
    """Distinct ``(station_id, lon, lat)`` inventory as a GeoDataFrame."""
    import geopandas as gpd

    if not self.stations_path.exists():
        raise FileNotFoundError(
            f"station inventory for {self.preset!r} is empty; run .sync(...) first."
        )
    return gpd.read_parquet(self.stations_path)
coverage 
coverage() -> list[ArchiveCoverage]

Per-station first/last/n_timesteps for the archive.

Source code in packages/xrtoolz-products/src/xrproducts/_src/cds/archive.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
def coverage(self) -> list[ArchiveCoverage]:
    """Per-station first/last/n_timesteps for the archive."""
    try:
        gdf = self.load()
    except FileNotFoundError:
        return []
    out: list[ArchiveCoverage] = []
    for sid, sub in gdf.groupby("station_id", sort=True):
        times = pd.to_datetime(sub["time"]).dropna()
        # In long-format each timestamp is repeated per variable,
        # so ``len(sub)`` over-counts. Use the unique-timestamp
        # count instead.
        n_unique = int(times.nunique())
        if n_unique == 0:
            first: pd.Timestamp | None = None
            last: pd.Timestamp | None = None
        else:
            first = cast("pd.Timestamp | None", pd.Timestamp(times.min()))
            last = cast("pd.Timestamp | None", pd.Timestamp(times.max()))
        out.append(
            ArchiveCoverage(
                preset=self.preset,
                station_id=str(sid),
                first=first,
                last=last,
                n_timesteps=n_unique,
            )
        )
    return out

CDSSource 

CDSSource(credentials: CDSCredentials | None = None, client: Any | None = None, format: str | None = None, product_type: str | None = 'reanalysis')

Bases: DataSource

Adapter around the cdsapi Python client.

Parameters:

Name Type Description Default
credentials CDSCredentials | None

Explicit :class:CDSCredentials. When None, credentials are resolved from env vars / ~/.cdsapirc.

 None
client Any | None

Optional pre-built cdsapi.Client (or test double).

 None
format str | None

CDS output format. When None (default) the family profile picks — "netcdf" for reanalyses, "zip" for in-situ. Override with "grib" / "netcdf" / "zip" to force a format for every request from this source.

 None
product_type str | None

Default product_type form entry when the dataset's profile admits it (reanalysis datasets; ignored for in-situ).

 'reanalysis'
Source code in packages/xrtoolz-products/src/xrproducts/_src/cds/source.py 
47
48
49
50
51
52
53
54
55
56
57
def __init__(
    self,
    credentials: CDSCredentials | None = None,
    client: Any | None = None,
    format: str | None = None,
    product_type: str | None = "reanalysis",
) -> None:
    self.credentials = credentials or load_cds()
    self._client = client
    self.format = format
    self.product_type = product_type
download 
download(dataset_id: str, output: Path, *, variables: list[str | Variable] | None = None, bbox: BBox | None = None, time: TimeRange | None = None, depth: DepthRange | None = None, levels: PressureLevels | None = None, **extras: Any) -> Path

Retrieve a dataset to output via cdsapi.Client.retrieve.

Source code in packages/xrtoolz-products/src/xrproducts/_src/cds/source.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def download(
    self,
    dataset_id: str,
    output: Path,
    *,
    variables: list[str | Variable] | None = None,
    bbox: BBox | None = None,
    time: TimeRange | None = None,
    depth: DepthRange | None = None,
    levels: PressureLevels | None = None,
    **extras: Any,
) -> Path:
    """Retrieve a dataset to ``output`` via ``cdsapi.Client.retrieve``."""
    output = Path(output)
    output.parent.mkdir(parents=True, exist_ok=True)
    form = self._build_form(
        dataset_id=dataset_id,
        variables=variables,
        bbox=bbox,
        time=time,
        levels=levels,
        extras=extras,
    )
    self._get_client().retrieve(dataset_id, form, str(output))
    return output
open 
open(dataset_id: str, *, variables: list[str | Variable] | None = None, bbox: BBox | None = None, time: TimeRange | None = None, depth: DepthRange | None = None, levels: PressureLevels | None = None, **extras: Any) -> Dataset

Download then open the resulting file with xarray.open_dataset.

CDS has no native lazy access path, so this always materialises a local file (under the request cache) before handing back an xr.Dataset. Both the cache file suffix and the xarray engine follow :attr:format, so reconfiguring to "grib" remains consistent end-to-end.

Source code in packages/xrtoolz-products/src/xrproducts/_src/cds/source.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def open(
    self,
    dataset_id: str,
    *,
    variables: list[str | Variable] | None = None,
    bbox: BBox | None = None,
    time: TimeRange | None = None,
    depth: DepthRange | None = None,
    levels: PressureLevels | None = None,
    **extras: Any,
) -> xr.Dataset:
    """Download then open the resulting file with ``xarray.open_dataset``.

    CDS has no native lazy access path, so this always materialises
    a local file (under the request cache) before handing back an
    ``xr.Dataset``. Both the cache file suffix and the xarray engine
    follow :attr:`format`, so reconfiguring to ``"grib"`` remains
    consistent end-to-end.
    """
    from xrproducts._src.cache import cache_path

    request = {
        "dataset_id": dataset_id,
        # Source-level form options change the bytes CDS returns, so they
        # must be part of the cache key — otherwise a source configured
        # with a different product_type/format would reuse a stale file.
        "product_type": self.product_type,
        "format": self.format,
        "variables": [v if isinstance(v, str) else v.name for v in variables or []],
        "bbox": bbox.__dict__ if bbox else None,
        "time": {
            "start": time.start.isoformat(),
            "end": time.end.isoformat(),
            "freq": time.freq,
        }
        if time is not None
        else None,
        "levels": levels.levels if levels else None,
        "extras": extras,
    }
    # Same format-resolution order as ``_build_form``. Respect the
    # profile's ``format_key`` (``format`` vs ``data_format``) and
    # treat ``None`` as "not provided" so a stale kwarg can't blank
    # the resolved format out.
    profile = resolve_profile(dataset_id)
    candidate = extras.get(profile.format_key) or extras.get("format")
    resolved_format = (
        candidate
        if isinstance(candidate, str) and candidate
        else (self.format or profile.format_default)
    )
    # Fail fast before the (potentially long, CDS-queued) download
    # when the output format isn't xarray-readable — use
    # ``CDSInsituArchive`` or ``download()`` for those.
    if resolved_format.lower() in {"zip", "csv"}:
        raise ValueError(
            f"CDS dataset {dataset_id!r} returns a {resolved_format} "
            "bundle, not an xarray-readable file. Use CDSInsituArchive "
            "or download() directly and parse the result yourself."
        )
    suffix = _suffix_for_format(resolved_format)
    path = cache_path(self.source_id, dataset_id, request, suffix=suffix)
    if not path.exists():
        self.download(
            dataset_id,
            path,
            variables=variables,
            bbox=bbox,
            time=time,
            depth=depth,
            levels=levels,
            **extras,
        )
    engine = _engine_for_format(resolved_format)
    try:
        ds = (
            xr.open_dataset(path, engine=engine)
            if engine
            else xr.open_dataset(path)
        )
    except (ImportError, ValueError) as exc:
        # cfgrib is not in the `cds` extra — it drags in the ecCodes
        # binaries and the CDS profiles default to NetCDF. Point at
        # the extra that provides it rather than at xarray's
        # "unrecognized engine" wording.
        if engine == "cfgrib":
            raise ImportError(
                f"Reading {resolved_format} output requires the 'cfgrib' "
                "backend (and the ecCodes binaries it wraps). Install "
                "with: pip install 'xrtoolz-products[grib]'"
            ) from exc
        raise
    # CDS may over-fetch (year/month/day cartesian); trim to the window.
    return _trim_to_time(ds, time)

CMEMSSource 

CMEMSSource(credentials: CMEMSCredentials | None = None, client: Any | None = None)

Bases: DataSource

Adapter around the copernicusmarine Python client.

Parameters:

Name Type Description Default
credentials CMEMSCredentials | None

Explicit :class:CMEMSCredentials. When None, credentials are resolved from env vars / ~/.cmems.

 None
client Any | None

Optional pre-built client (copernicusmarine module or test double). Useful for tests; production code leaves this None.

 None
Source code in packages/xrtoolz-products/src/xrproducts/_src/cmems/source.py 
39
40
41
42
43
44
45
def __init__(
    self,
    credentials: CMEMSCredentials | None = None,
    client: Any | None = None,
) -> None:
    self.credentials = credentials or load_cmems()
    self._client = client
download 
download(dataset_id: str, output: Path, *, variables: list[str | Variable] | None = None, bbox: BBox | None = None, time: TimeRange | None = None, depth: DepthRange | None = None, levels: PressureLevels | None = None, **extras: Any) -> Path

Download a subset to a NetCDF file at output.

Source code in packages/xrtoolz-products/src/xrproducts/_src/cmems/source.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def download(
    self,
    dataset_id: str,
    output: Path,
    *,
    variables: list[str | Variable] | None = None,
    bbox: BBox | None = None,
    time: TimeRange | None = None,
    depth: DepthRange | None = None,
    levels: PressureLevels | None = None,
    **extras: Any,
) -> Path:
    """Download a subset to a NetCDF file at ``output``."""
    output = Path(output)
    output.parent.mkdir(parents=True, exist_ok=True)
    kwargs = self._subset_kwargs(
        dataset_id=dataset_id,
        variables=variables,
        bbox=bbox,
        time=time,
        depth=depth,
        extras=extras,
    )
    self._get_client().subset(
        **kwargs,
        output_filename=output.name,
        output_directory=str(output.parent),
        **self._auth_kwargs(),
    )
    return output
open 
open(dataset_id: str, *, variables: list[str | Variable] | None = None, bbox: BBox | None = None, time: TimeRange | None = None, depth: DepthRange | None = None, levels: PressureLevels | None = None, **extras: Any) -> Dataset

Open a lazy dataset via copernicusmarine.open_dataset.

Source code in packages/xrtoolz-products/src/xrproducts/_src/cmems/source.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def open(
    self,
    dataset_id: str,
    *,
    variables: list[str | Variable] | None = None,
    bbox: BBox | None = None,
    time: TimeRange | None = None,
    depth: DepthRange | None = None,
    levels: PressureLevels | None = None,
    **extras: Any,
) -> xr.Dataset:
    """Open a lazy dataset via ``copernicusmarine.open_dataset``."""
    kwargs = self._subset_kwargs(
        dataset_id=dataset_id,
        variables=variables,
        bbox=bbox,
        time=time,
        depth=depth,
        extras=extras,
    )
    return self._get_client().open_dataset(**kwargs, **self._auth_kwargs())

AEMETCredentials dataclass 

AEMETCredentials(api_key: str)

API key for AEMET OpenData.

CDSCredentials dataclass 

CDSCredentials(url: str, key: str)

URL / key pair for the Climate Data Store.

CMEMSCredentials dataclass 

CMEMSCredentials(username: str, password: str)

Username / password pair for Copernicus Marine.

LocalL2Source

Bases: DataSource

Open satellite L2 product files from the local filesystem.

dataset_id selects a product opener from :data:LOCAL_DATASETS ("tropomi.ch4", "emit.ch4", "ghgsat.ch4"); the file path comes through path= (or paths= for multi-granule concatenation along time). bbox is applied with :func:xrproducts.subset_bbox on the 2-D lon/lat coordinates (after matching the box to the dataset's longitude convention) and time as a mask on the time coordinate (or on the per-scanline scanline_time for swaths). Nothing is fetched: :meth:download raises.

The openers read NetCDF, which the core install does not ship a backend for: pip install 'xrtoolz-products[local]'.

download 
download(dataset_id: str, output: Path, *, variables: list[str | Variable] | None = None, bbox: BBox | None = None, time: TimeRange | None = None, depth: DepthRange | None = None, levels: PressureLevels | None = None, **extras: Any) -> Path

Not supported: the product files are already on disk.

Raises:

Type Description
NotImplementedError

Always.
Source code in packages/xrtoolz-products/src/xrproducts/_src/local/source.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def download(
    self,
    dataset_id: str,
    output: Path,
    *,
    variables: list[str | Variable] | None = None,
    bbox: BBox | None = None,
    time: TimeRange | None = None,
    depth: DepthRange | None = None,
    levels: PressureLevels | None = None,
    **extras: Any,
) -> Path:
    """Not supported: the product files are already on disk.

    Raises:
        NotImplementedError: Always.
    """
    raise NotImplementedError("local files are already on disk")
open 
open(dataset_id: str, *, path: str | Path | None = None, paths: Sequence[str | Path] | None = None, glt_paths: Sequence[str | Path] | None = None, variables: list[str | Variable] | None = None, bbox: BBox | None = None, time: TimeRange | None = None, depth: DepthRange | None = None, levels: PressureLevels | None = None, qa_min: float | None = None, **extras: Any) -> Dataset

Open one or more product files as a single flat Dataset.

Parameters:

Name Type Description Default
dataset_id str

Catalog short name ("tropomi.ch4", "emit.ch4" or "ghgsat.ch4").

 required
path str | Path | None

The product file. Exactly one of path / paths.

 None
paths Sequence[str | Path] | None

Several granules / scenes, concatenated along time in the order given (they must share the swath or raster shape).

 None
glt_paths Sequence[str | Path] | None

One EMIT geometric lookup table per entry of paths (same order, same length). A single scene takes glt_path= through extras instead.

 None
variables list[str | Variable] | None

Registry names or :class:Variable instances to keep; None keeps everything the opener produced.

 None
bbox BBox | None

Keep only pixels whose 2-D lon / lat fall inside the box (where(..., drop=True) — scanlines / rows that fall entirely outside are dropped). The box is normalised to the dataset's longitude convention first ([-180, 180] when any lon is negative, [0, 360] when any exceeds 180), so either convention works.

 None
time TimeRange | None

Keep measurements inside the window. Swaths are masked on their per-scanline scanline_time; scenes on their time coordinate (a mask, so paths need not be in chronological order).

 None
depth DepthRange | None

Ignored (no vertical axis in these products).

 None
levels PressureLevels | None

Ignored (no vertical axis in these products).

 None
qa_min float | None

Forwarded to the opener when given (TROPOMI: qa_value < qa_min is masked). None leaves the opener's own default in force (0.5 for TROPOMI); pass 0.0 to keep every pixel.

 None
**extras Any

Forwarded to the opener (keep_groups=, glt_path= for a single EMIT scene ...).

 {}

Returns:

Type Description
Dataset

The screened Dataset.

Raises:

Type Description
KeyError

If dataset_id is not a local product, or a requested variable is absent from the opened file.
ValueError

If neither or both of path / paths are given; if glt_paths does not line up with paths or a single glt_path is given for several scenes; or if bbox crosses the antimeridian once expressed in the dataset's longitude convention.
Source code in packages/xrtoolz-products/src/xrproducts/_src/local/source.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def open(
    self,
    dataset_id: str,
    *,
    path: str | Path | None = None,
    paths: Sequence[str | Path] | None = None,
    glt_paths: Sequence[str | Path] | None = None,
    variables: list[str | Variable] | None = None,
    bbox: BBox | None = None,
    time: TimeRange | None = None,
    depth: DepthRange | None = None,
    levels: PressureLevels | None = None,
    qa_min: float | None = None,
    **extras: Any,
) -> xr.Dataset:
    """Open one or more product files as a single flat Dataset.

    Args:
        dataset_id: Catalog short name (``"tropomi.ch4"``, ``"emit.ch4"``
            or ``"ghgsat.ch4"``).
        path: The product file. Exactly one of ``path`` / ``paths``.
        paths: Several granules / scenes, concatenated along ``time``
            in the order given (they must share the swath or raster
            shape).
        glt_paths: One EMIT geometric lookup table per entry of
            ``paths`` (same order, same length). A single scene takes
            ``glt_path=`` through ``extras`` instead.
        variables: Registry names or :class:`Variable` instances to
            keep; ``None`` keeps everything the opener produced.
        bbox: Keep only pixels whose 2-D ``lon`` / ``lat`` fall inside
            the box (``where(..., drop=True)`` — scanlines / rows
            that fall entirely outside are dropped). The box is
            normalised to the dataset's longitude convention first
            (``[-180, 180]`` when any ``lon`` is negative, ``[0, 360]``
            when any exceeds 180), so either convention works.
        time: Keep measurements inside the window. Swaths are masked
            on their per-scanline ``scanline_time``; scenes on their
            ``time`` coordinate (a mask, so ``paths`` need not be in
            chronological order).
        depth: Ignored (no vertical axis in these products).
        levels: Ignored (no vertical axis in these products).
        qa_min: Forwarded to the opener when given (TROPOMI:
            ``qa_value < qa_min`` is masked). ``None`` leaves the
            opener's own default in force (``0.5`` for TROPOMI); pass
            ``0.0`` to keep every pixel.
        **extras: Forwarded to the opener (``keep_groups=``,
            ``glt_path=`` for a single EMIT scene ...).

    Returns:
        The screened Dataset.

    Raises:
        KeyError: If ``dataset_id`` is not a local product, or a
            requested variable is absent from the opened file.
        ValueError: If neither or both of ``path`` / ``paths`` are
            given; if ``glt_paths`` does not line up with ``paths`` or
            a single ``glt_path`` is given for several scenes; or if
            ``bbox`` crosses the antimeridian once expressed in the
            dataset's longitude convention.
    """
    opener = _OPENERS.get(dataset_id)
    if opener is None:
        raise KeyError(
            f"Unknown local dataset {dataset_id!r}; known: {sorted(_OPENERS)}"
        )
    if (path is None) == (paths is None):
        raise ValueError("pass exactly one of path= or paths=")
    if qa_min is not None:
        extras["qa_min"] = qa_min
    files = [path] if path is not None else list(paths or ())
    per_file = _per_file_extras(files, glt_paths, extras)
    parts = [
        opener(Path(p), **extras, **kw)
        for p, kw in zip(files, per_file, strict=True)
    ]
    ds = parts[0] if len(parts) == 1 else xr.concat(parts, dim="time")
    if variables:
        keep = {v.name for v in self._resolve_variables(variables)}
        if missing := sorted(keep - set(ds.data_vars)):
            raise KeyError(f"{missing} not in dataset; have {sorted(ds.data_vars)}")
        ds = ds.drop_vars([n for n in ds.data_vars if n not in keep])
    if bbox is not None:
        bbox = _match_lon_convention(bbox, ds["lon"])
        if bbox.crosses_antimeridian:
            raise ValueError(
                "BBox crosses the antimeridian in the dataset's longitude "
                "convention; split it at the antimeridian and open each half"
            )
        ds = subset_bbox(
            ds, (bbox.lon_min, bbox.lon_max), (bbox.lat_min, bbox.lat_max)
        )
    if time is not None:
        ds = _subset_time_range(ds, time)
    return ds

BBox dataclass 

BBox(lon_min: float, lon_max: float, lat_min: float, lat_max: float)

Axis-aligned lon/lat bounding box.

Longitude is accepted in either [-180, 180] or [0, 360] convention; use :meth:to_180 / :meth:to_360 to normalize. A lon_min > lon_max box is interpreted as crossing the antimeridian.

width property 
width: float

Longitude extent (handles antimeridian wrap).

height property 
height: float

Latitude extent in degrees.

from_tuple classmethod 
from_tuple(t: tuple[float, float, float, float]) -> BBox

Build from (lon_min, lon_max, lat_min, lat_max).

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/geometry.py 
35
36
37
38
39
@classmethod
def from_tuple(cls, t: tuple[float, float, float, float]) -> BBox:
    """Build from ``(lon_min, lon_max, lat_min, lat_max)``."""
    lon_min, lon_max, lat_min, lat_max = t
    return cls(lon_min, lon_max, lat_min, lat_max)
global_ classmethod 
global_() -> BBox

Whole-globe bounding box in [-180, 180] convention.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/geometry.py 
41
42
43
44
@classmethod
def global_(cls) -> BBox:
    """Whole-globe bounding box in ``[-180, 180]`` convention."""
    return cls(-180.0, 180.0, -90.0, 90.0)
to_180 
to_180() -> BBox

Return a copy with longitudes normalized to [-180, 180].

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/geometry.py 
66
67
68
69
70
71
72
73
74
75
76
def to_180(self) -> BBox:
    """Return a copy with longitudes normalized to ``[-180, 180]``."""

    def _w(x: float) -> float:
        return ((x + 180.0) % 360.0) - 180.0

    # A full-globe box has identical endpoints under modulo (both map
    # to -180); preserve the canonical span instead of collapsing it.
    if self.lon_max - self.lon_min == 360.0:
        return BBox(-180.0, 180.0, self.lat_min, self.lat_max)
    return BBox(_w(self.lon_min), _w(self.lon_max), self.lat_min, self.lat_max)
to_360 
to_360() -> BBox

Return a copy with longitudes normalized to [0, 360].

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/geometry.py 
78
79
80
81
82
83
84
85
86
def to_360(self) -> BBox:
    """Return a copy with longitudes normalized to ``[0, 360]``."""
    # ``-180 % 360 == 180 == 180 % 360`` would collapse a full-globe box
    # to a single meridian; preserve the canonical span instead.
    if self.lon_max - self.lon_min == 360.0:
        return BBox(0.0, 360.0, self.lat_min, self.lat_max)
    return BBox(
        self.lon_min % 360.0, self.lon_max % 360.0, self.lat_min, self.lat_max
    )
as_cmems 
as_cmems() -> dict[str, float]

copernicusmarine.subset kwargs.

Raises:

Type Description
ValueError

if the box crosses the antimeridian — passing lon_min > lon_max would send CMEMS an inverted window. Call :meth:to_360 first, or split the box at ±180°.
Source code in packages/xrtoolz-products/src/xrproducts/types/_src/geometry.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def as_cmems(self) -> dict[str, float]:
    """``copernicusmarine.subset`` kwargs.

    Raises:
        ValueError: if the box crosses the antimeridian — passing
            ``lon_min > lon_max`` would send CMEMS an inverted window.
            Call :meth:`to_360` first, or split the box at ±180°.
    """
    if self.crosses_antimeridian:
        raise ValueError(
            "BBox crosses the antimeridian; CMEMS would receive an "
            "inverted minimum/maximum longitude. Call .to_360() first "
            "or split the box at ±180° and request each half."
        )
    return {
        "minimum_longitude": self.lon_min,
        "maximum_longitude": self.lon_max,
        "minimum_latitude": self.lat_min,
        "maximum_latitude": self.lat_max,
    }
as_cds_area 
as_cds_area() -> list[float]

CDS area key: [North, West, South, East].

Raises:

Type Description
ValueError

if the box crosses the antimeridian — West > East is an inverted window. Call :meth:to_360 first, or split the box at ±180°.
Source code in packages/xrtoolz-products/src/xrproducts/types/_src/geometry.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def as_cds_area(self) -> list[float]:
    """CDS ``area`` key: ``[North, West, South, East]``.

    Raises:
        ValueError: if the box crosses the antimeridian — ``West > East``
            is an inverted window. Call :meth:`to_360` first, or split
            the box at ±180°.
    """
    if self.crosses_antimeridian:
        raise ValueError(
            "BBox crosses the antimeridian; CDS would receive an inverted "
            "West/East window. Call .to_360() first or split the box at "
            "±180° and request each half."
        )
    return [self.lat_max, self.lon_min, self.lat_min, self.lon_max]
as_xarray_sel 
as_xarray_sel(lon: str = 'lon', lat: str = 'lat', lat_descending: bool = False) -> dict[str, slice]

ds.sel(**bbox.as_xarray_sel()) selector.

Parameters:

Name Type Description Default
lon str

Name of the longitude coordinate.

 'lon'
lat str

Name of the latitude coordinate.

 'lat'
lat_descending bool

Set when the dataset's latitude index runs north-to-south (ERA5 and most CDS reanalysis output do). sel requires slice bounds in index order, so an ascending slice(lat_min, lat_max) selects zero points on such a grid. Infer it with bool(ds[lat][0] > ds[lat][-1]).

 False

Raises:

Type Description
ValueError

if the box crosses the antimeridian — a single slice(lon_min, lon_max) would select zero points in that case. Split the box at ±180° and select the halves separately, or call :meth:to_360 first.
Source code in packages/xrtoolz-products/src/xrproducts/types/_src/geometry.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def as_xarray_sel(
    self,
    lon: str = "lon",
    lat: str = "lat",
    lat_descending: bool = False,
) -> dict[str, slice]:
    """``ds.sel(**bbox.as_xarray_sel())`` selector.

    Args:
        lon: Name of the longitude coordinate.
        lat: Name of the latitude coordinate.
        lat_descending: Set when the dataset's latitude index runs
            north-to-south (ERA5 and most CDS reanalysis output do).
            ``sel`` requires slice bounds in index order, so an
            ascending ``slice(lat_min, lat_max)`` selects **zero**
            points on such a grid. Infer it with
            ``bool(ds[lat][0] > ds[lat][-1])``.

    Raises:
        ValueError: if the box crosses the antimeridian — a single
            ``slice(lon_min, lon_max)`` would select zero points in
            that case. Split the box at ±180° and select the halves
            separately, or call :meth:`to_360` first.
    """
    if self.crosses_antimeridian:
        raise ValueError(
            "BBox crosses the antimeridian; a single slice cannot "
            "represent the wrap. Split the box at ±180° and sel() "
            "each half, or call .to_360() first."
        )
    lat_slice = (
        slice(self.lat_max, self.lat_min)
        if lat_descending
        else slice(self.lat_min, self.lat_max)
    )
    return {
        lon: slice(self.lon_min, self.lon_max),
        lat: lat_slice,
    }

DepthRange dataclass 

DepthRange(min: float, max: float)

Continuous depth window (positive-down, metres).

Issue dataclass 

Issue(variable: str, severity: Severity, code: str, message: str)

Single validation finding.

PressureLevels dataclass 

PressureLevels(levels: tuple[int, ...])

Discrete pressure levels in hPa (CDS-style).

Request dataclass 

Request(variables: tuple[str, ...], bbox: BBox | None = None, time: TimeRange | None = None, depth: DepthRange | None = None, levels: PressureLevels | None = None, extras: dict[str, Any] | None = None)

Canonical request payload consumed by data adapters.

Not every field applies to every source; adapters pick what they need. Keeping one struct means the catalog and the UI speak the same shape.

Severity

Bases: StrEnum

Two-level severity for validation issues.

Station dataclass 

Station(id: str, name: str, lon: float, lat: float, altitude: float | None = None, wmo_id: str | None = None, source: str | None = None, city: str | None = None, province: str | None = None, community: str | None = None, timezone: str | None = None, start_time: str | None = None, end_time: str | None = None, active: bool | None = None, attrs: Mapping[str, Any] = dict())

One fixed observation station.

Coordinates are decimal degrees in the [-180, 180] / [-90, 90] convention. city / province / community reflect the administrative hierarchy exposed by AEMET and most national services; adapters that don't split this way can leave them None.

distance_to 
distance_to(lon: float, lat: float) -> float

Great-circle distance to (lon, lat) in kilometres.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/station.py 
69
70
71
def distance_to(self, lon: float, lat: float) -> float:
    """Great-circle distance to ``(lon, lat)`` in kilometres."""
    return _haversine_km(self.lon, self.lat, lon, lat)
in_bbox 
in_bbox(bbox: BBox) -> bool

Whether the station falls inside bbox (inclusive).

Station longitudes are stored in [-180, 180]; bbox may be supplied in either convention, so it is normalized to [-180, 180] first (which may turn a [0, 360] box into an antimeridian- crossing one, handled below).

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/station.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def in_bbox(self, bbox: BBox) -> bool:
    """Whether the station falls inside ``bbox`` (inclusive).

    Station longitudes are stored in ``[-180, 180]``; ``bbox`` may be
    supplied in either convention, so it is normalized to ``[-180, 180]``
    first (which may turn a ``[0, 360]`` box into an antimeridian-
    crossing one, handled below).
    """
    if not bbox.lat_min <= self.lat <= bbox.lat_max:
        return False
    bbox = bbox.to_180()
    if bbox.crosses_antimeridian:
        return self.lon >= bbox.lon_min or self.lon <= bbox.lon_max
    return bbox.lon_min <= self.lon <= bbox.lon_max

StationCollection dataclass 

StationCollection(stations: tuple[Station, ...] = ())

Immutable collection of :class:Station with set-like filters.

Supports iteration, len, containment, tuple-style indexing, and conversion to :class:pandas.DataFrame. All filter methods return a new collection rather than mutating in place, so chains like stations.filter(province="Cantabria").within(bbox) are safe.

from_iter classmethod 
from_iter(stations: Iterable[Station]) -> StationCollection

Build from any iterable of :class:Station.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/station.py 
134
135
136
137
@classmethod
def from_iter(cls, stations: Iterable[Station]) -> StationCollection:
    """Build from any iterable of :class:`Station`."""
    return cls(tuple(stations))
filter 
filter(*, bbox: BBox | None = None, city: str | Iterable[str] | None = None, province: str | Iterable[str] | None = None, community: str | Iterable[str] | None = None, source: str | Iterable[str] | None = None, active: bool | None = None, has_wmo: bool | None = None) -> StationCollection

Return a new collection keeping stations matching all filters.

String-valued filters accept either a single value or an iterable of values (OR-semantics within a filter, AND across filters). Matching is case-insensitive on the admin fields so users don't have to remember whether AEMET capitalises "Cantabria" or not.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/station.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def filter(
    self,
    *,
    bbox: BBox | None = None,
    city: str | Iterable[str] | None = None,
    province: str | Iterable[str] | None = None,
    community: str | Iterable[str] | None = None,
    source: str | Iterable[str] | None = None,
    active: bool | None = None,
    has_wmo: bool | None = None,
) -> StationCollection:
    """Return a new collection keeping stations matching *all* filters.

    String-valued filters accept either a single value or an iterable
    of values (OR-semantics within a filter, AND across filters).
    Matching is case-insensitive on the admin fields so users don't
    have to remember whether AEMET capitalises "Cantabria" or not.
    """
    city_set = _norm_set(city)
    province_set = _norm_set(province)
    community_set = _norm_set(community)
    source_set = _norm_set(source)

    def keep(s: Station) -> bool:
        if bbox is not None and not s.in_bbox(bbox):
            return False
        if city_set is not None and _norm(s.city) not in city_set:
            return False
        if province_set is not None and _norm(s.province) not in province_set:
            return False
        if community_set is not None and _norm(s.community) not in community_set:
            return False
        if source_set is not None and _norm(s.source) not in source_set:
            return False
        if active is not None and s.active != active:
            return False
        if has_wmo is True and not s.wmo_id:
            return False
        return not (has_wmo is False and s.wmo_id)

    return StationCollection(tuple(s for s in self.stations if keep(s)))
within 
within(bbox: BBox) -> StationCollection

Shorthand for filter(bbox=bbox).

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/station.py 
183
184
185
def within(self, bbox: BBox) -> StationCollection:
    """Shorthand for ``filter(bbox=bbox)``."""
    return self.filter(bbox=bbox)
nearest 
nearest(point: tuple[float, float], n: int = 1) -> StationCollection

Return the n stations closest to point = (lon, lat).

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/station.py 
187
188
189
190
191
192
193
def nearest(self, point: tuple[float, float], n: int = 1) -> StationCollection:
    """Return the ``n`` stations closest to ``point = (lon, lat)``."""
    if n <= 0:
        return StationCollection(())
    lon, lat = point
    ranked = sorted(self.stations, key=lambda s: s.distance_to(lon, lat))
    return StationCollection(tuple(ranked[:n]))
ids 
ids() -> tuple[str, ...]

Tuple of station IDs in collection order.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/station.py 
197
198
199
def ids(self) -> tuple[str, ...]:
    """Tuple of station IDs in collection order."""
    return tuple(s.id for s in self.stations)
cities 
cities() -> tuple[str, ...]

Sorted unique non-empty city values.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/station.py 
201
202
203
def cities(self) -> tuple[str, ...]:
    """Sorted unique non-empty ``city`` values."""
    return _sorted_unique(s.city for s in self.stations)
provinces 
provinces() -> tuple[str, ...]

Sorted unique non-empty province values.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/station.py 
205
206
207
def provinces(self) -> tuple[str, ...]:
    """Sorted unique non-empty ``province`` values."""
    return _sorted_unique(s.province for s in self.stations)
communities 
communities() -> tuple[str, ...]

Sorted unique non-empty community values.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/station.py 
209
210
211
def communities(self) -> tuple[str, ...]:
    """Sorted unique non-empty ``community`` values."""
    return _sorted_unique(s.community for s in self.stations)
to_dataframe 
to_dataframe() -> DataFrame

Return the collection as a :class:pandas.DataFrame.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/station.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def to_dataframe(self) -> pd.DataFrame:
    """Return the collection as a :class:`pandas.DataFrame`."""
    import pandas as pd

    if not self.stations:
        return pd.DataFrame(
            columns=[
                "id",
                "name",
                "lon",
                "lat",
                "altitude",
                "wmo_id",
                "source",
                "city",
                "province",
                "community",
                "timezone",
                "start_time",
                "end_time",
                "active",
            ]
        )
    rows = [
        {
            "id": s.id,
            "name": s.name,
            "lon": s.lon,
            "lat": s.lat,
            "altitude": s.altitude,
            "wmo_id": s.wmo_id,
            "source": s.source,
            "city": s.city,
            "province": s.province,
            "community": s.community,
            "timezone": s.timezone,
            "start_time": s.start_time,
            "end_time": s.end_time,
            "active": s.active,
        }
        for s in self.stations
    ]
    return pd.DataFrame(rows)

TimeRange dataclass 

TimeRange(start: Timestamp, end: Timestamp, freq: str | None = None)

Inclusive time window, optionally with a target sampling frequency.

freq is a pandas offset alias ("1D", "6H" ...) used only by serializers that need to materialize the range (e.g. the CDS year/month/day form). Adapters accepting continuous ranges (CMEMS) ignore freq.

parse classmethod 
parse(start: str | datetime | Timestamp, end: str | datetime | Timestamp, freq: str | None = None) -> TimeRange

Parse heterogeneous inputs into a TimeRange.

Accepts ISO strings, datetime, pandas.Timestamp. Strings without timezone are treated as UTC.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/time.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
@classmethod
def parse(
    cls,
    start: str | datetime | pd.Timestamp,
    end: str | datetime | pd.Timestamp,
    freq: str | None = None,
) -> TimeRange:
    """Parse heterogeneous inputs into a ``TimeRange``.

    Accepts ISO strings, ``datetime``, ``pandas.Timestamp``.
    Strings without timezone are treated as UTC.
    """
    return cls(start=_as_ts(start), end=_as_ts(end), freq=freq)
to_index 
to_index() -> DatetimeIndex

Materialize the range at :attr:freq (default daily).

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/time.py 
54
55
56
def to_index(self) -> pd.DatetimeIndex:
    """Materialize the range at :attr:`freq` (default daily)."""
    return pd.date_range(self.start, self.end, freq=self.freq or "1D")
as_cds_form 
as_cds_form() -> dict[str, list[str]]

Explode the range into the year/month/day form CDS expects.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/time.py 
66
67
68
69
70
71
72
def as_cds_form(self) -> dict[str, list[str]]:
    """Explode the range into the ``year/month/day`` form CDS expects."""
    idx = self.to_index()
    years = sorted({f"{t.year}" for t in idx})
    months = sorted({f"{t.month:02d}" for t in idx})
    days = sorted({f"{t.day:02d}" for t in idx})
    return {"year": years, "month": months, "day": days}
cds_times 
cds_times() -> list[str]

Hours-of-day ("HH:MM") the range touches, for the CDS time key.

Derived from :meth:to_index at :attr:freq (default daily, so ["00:00"]). ERA5-style reanalyses are hourly products and require this field alongside year/month/day; a finer freq (e.g. "6h") expands the returned hours accordingly.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/time.py 
74
75
76
77
78
79
80
81
82
83
def cds_times(self) -> list[str]:
    """Hours-of-day (``"HH:MM"``) the range touches, for the CDS ``time`` key.

    Derived from :meth:`to_index` at :attr:`freq` (default daily, so
    ``["00:00"]``). ERA5-style reanalyses are hourly products and
    require this field alongside ``year/month/day``; a finer ``freq``
    (e.g. ``"6h"``) expands the returned hours accordingly.
    """
    idx = self.to_index()
    return sorted({f"{t.hour:02d}:{t.minute:02d}" for t in idx})
as_xarray_sel 
as_xarray_sel(time: str = 'time') -> dict[str, slice]

ds.sel(**tr.as_xarray_sel()) selector.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/time.py 
85
86
87
def as_xarray_sel(self, time: str = "time") -> dict[str, slice]:
    """``ds.sel(**tr.as_xarray_sel())`` selector."""
    return {time: slice(self.start, self.end)}

ValidationReport dataclass 

ValidationReport(issues: list[Issue] = list())

Aggregate validation result with convenience accessors.

raise_if_errors 
raise_if_errors() -> None

Raise :class:ValueError if any error-level issues are present.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/validation.py 
63
64
65
66
67
68
def raise_if_errors(self) -> None:
    """Raise :class:`ValueError` if any error-level issues are present."""
    errs = self.errors()
    if errs:
        lines = [f"[{i.code}] {i.variable}: {i.message}" for i in errs]
        raise ValueError("Validation failed:\n  " + "\n  ".join(lines))

Variable dataclass 

Variable(name: str, standard_name: str | None = None, long_name: str | None = None, units: str | None = None, aliases: Mapping[str, str] = dict(), valid_range: tuple[float, float] | None = None, dtype: str | None = None, cmap: str | None = None)

CF-aware variable descriptor used across adapters and validators.

The class is an immutable dataclass so instances are hashable and safe to share; per-source identifiers live in aliases and are resolved with :meth:for_source.

for_source 
for_source(source: str) -> str

Return the identifier this variable uses in source.

Falls back to :attr:name if no alias is registered.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/variable.py 
56
57
58
59
60
61
def for_source(self, source: str) -> str:
    """Return the identifier this variable uses in ``source``.

    Falls back to :attr:`name` if no alias is registered.
    """
    return self.aliases.get(source, self.name)
cf_attrs 
cf_attrs() -> dict[str, str]

Return the CF-1.x metadata dict suitable for da.attrs.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/variable.py 
63
64
65
66
67
68
69
70
71
72
def cf_attrs(self) -> dict[str, str]:
    """Return the CF-1.x metadata dict suitable for ``da.attrs``."""
    attrs: dict[str, str] = {}
    if self.standard_name is not None:
        attrs["standard_name"] = self.standard_name
    if self.long_name is not None:
        attrs["long_name"] = self.long_name
    if self.units is not None:
        attrs["units"] = self.units
    return attrs

all_entries 

all_entries() -> dict[str, CatalogEntry]

Return a shallow copy of the unified catalog.

Source code in packages/xrtoolz-products/src/xrproducts/_src/catalog.py 
146
147
148
def all_entries() -> dict[str, CatalogEntry]:
    """Return a shallow copy of the unified catalog."""
    return dict(CATALOG)

describe 

describe(name: str) -> DatasetInfo

Look up a :class:DatasetInfo by short name.

Source code in packages/xrtoolz-products/src/xrproducts/_src/catalog.py 
151
152
153
154
155
156
157
158
159
160
161
162
def describe(name: str) -> DatasetInfo:
    """Look up a :class:`DatasetInfo` by short name."""
    entry = CATALOG[name]
    if entry.source == "cmems":
        return CMEMS_DATASETS[entry.dataset_id]
    if entry.source == "cds":
        return CDS_DATASETS[entry.dataset_id]
    if entry.source == "aemet":
        return AEMET_DATASETS[entry.dataset_id]
    if entry.source == "local":
        return LOCAL_DATASETS[entry.dataset_id]
    raise KeyError(f"Unknown source {entry.source!r}")

load_aemet 

load_aemet(api_key: str | None = None, path: Path | None = None) -> AEMETCredentials | None

Load AEMET credentials, or None if none could be found.

Resolution order:

  1. Explicit api_key argument.
  2. AEMET_API_KEY environment variable.
  3. .env file in the current working directory (one AEMET_API_KEY=... line).
  4. path (or ~/.aemet) parsed as a key: value config.

The .env step matches the most common local-dev workflow without introducing a python-dotenv dependency — we parse the single key we need and leave the file alone.

Source code in packages/xrtoolz-products/src/xrproducts/_src/credentials.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def load_aemet(
    api_key: str | None = None,
    path: Path | None = None,
) -> AEMETCredentials | None:
    """Load AEMET credentials, or ``None`` if none could be found.

    Resolution order:

    1. Explicit ``api_key`` argument.
    2. ``AEMET_API_KEY`` environment variable.
    3. ``.env`` file in the current working directory (one
       ``AEMET_API_KEY=...`` line).
    4. ``path`` (or ``~/.aemet``) parsed as a ``key: value`` config.

    The ``.env`` step matches the most common local-dev workflow
    without introducing a ``python-dotenv`` dependency — we parse the
    single key we need and leave the file alone.
    """
    if api_key:
        return AEMETCredentials(api_key=api_key)

    env_key = os.environ.get("AEMET_API_KEY")
    if env_key:
        return AEMETCredentials(api_key=env_key)

    # Walk up from CWD looking for ``.env``; matches how
    # ``python-dotenv`` behaves so a notebook opened in ``docs/``
    # still picks up the project-root ``.env``.
    for candidate in (Path.cwd(), *Path.cwd().parents):
        dotenv = candidate / ".env"
        if dotenv.is_file():
            parsed = _parse_kv(dotenv.read_text())
            k = parsed.get("aemet_api_key")
            if k:
                return AEMETCredentials(api_key=k)
            break

    cfg = path or Path.home() / ".aemet"
    if cfg.is_file():
        parsed = _parse_kv(cfg.read_text())
        k = parsed.get("api_key") or parsed.get("aemet_api_key")
        if k:
            return AEMETCredentials(api_key=k)

    return None

load_cds 

load_cds(url: str | None = None, key: str | None = None, path: Path | None = None) -> CDSCredentials | None

Load CDS credentials, or None if none could be found.

Resolution order:

  1. Explicit url + key arguments.
  2. Explicit path argument (parsed as a key: value config). When given, this bypasses env vars and .env lookup so the caller has a deterministic override.
  3. Environment variables CDSAPI_URL + CDSAPI_KEY.
  4. .env file walked up from CWD (CDSAPI_URL and CDSAPI_KEY keys), matching the python-dotenv lookup behaviour. A notebook started under docs/ still picks up the project-root .env.
  5. ~/.cdsapirc parsed as a key: value config.
Source code in packages/xrtoolz-products/src/xrproducts/_src/credentials.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def load_cds(
    url: str | None = None,
    key: str | None = None,
    path: Path | None = None,
) -> CDSCredentials | None:
    """Load CDS credentials, or ``None`` if none could be found.

    Resolution order:

    1. Explicit ``url`` + ``key`` arguments.
    2. Explicit ``path`` argument (parsed as a ``key: value`` config).
       When given, this bypasses env vars and ``.env`` lookup so the
       caller has a deterministic override.
    3. Environment variables ``CDSAPI_URL`` + ``CDSAPI_KEY``.
    4. ``.env`` file walked up from CWD (``CDSAPI_URL`` and
       ``CDSAPI_KEY`` keys), matching the ``python-dotenv`` lookup
       behaviour. A notebook started under ``docs/`` still picks up
       the project-root ``.env``.
    5. ``~/.cdsapirc`` parsed as a ``key: value`` config.
    """
    if url and key:
        return CDSCredentials(url=url, key=key)

    if path is not None:
        return _load_cds_from_file(path)

    env_url = os.environ.get("CDSAPI_URL")
    env_key = os.environ.get("CDSAPI_KEY")
    if env_url and env_key:
        return CDSCredentials(url=env_url, key=env_key)

    for candidate in (Path.cwd(), *Path.cwd().parents):
        dotenv = candidate / ".env"
        if dotenv.is_file():
            parsed = _parse_kv(dotenv.read_text())
            u = parsed.get("cdsapi_url")
            k = parsed.get("cdsapi_key")
            if u and k:
                return CDSCredentials(url=u, key=k)
            break

    return _load_cds_from_file(Path.home() / ".cdsapirc")

load_cmems 

load_cmems(username: str | None = None, password: str | None = None, path: Path | None = None) -> CMEMSCredentials | None

Load CMEMS credentials, or None if none could be found.

Resolution order:

  1. Explicit username + password arguments.
  2. Environment variables COPERNICUSMARINE_SERVICE_USERNAME + COPERNICUSMARINE_SERVICE_PASSWORD.
  3. .env file walked up from CWD (same keys), matching the python-dotenv lookup the CDS / AEMET loaders use, so a notebook started under docs/ still picks up the project-root .env.
  4. ~/.cmems (or path) parsed as a key: value config.
Source code in packages/xrtoolz-products/src/xrproducts/_src/credentials.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def load_cmems(
    username: str | None = None,
    password: str | None = None,
    path: Path | None = None,
) -> CMEMSCredentials | None:
    """Load CMEMS credentials, or ``None`` if none could be found.

    Resolution order:

    1. Explicit ``username`` + ``password`` arguments.
    2. Environment variables ``COPERNICUSMARINE_SERVICE_USERNAME`` +
       ``COPERNICUSMARINE_SERVICE_PASSWORD``.
    3. ``.env`` file walked up from CWD (same keys), matching the
       ``python-dotenv`` lookup the CDS / AEMET loaders use, so a
       notebook started under ``docs/`` still picks up the project-root
       ``.env``.
    4. ``~/.cmems`` (or ``path``) parsed as a ``key: value`` config.
    """
    if username and password:
        return CMEMSCredentials(username=username, password=password)

    env_u = os.environ.get("COPERNICUSMARINE_SERVICE_USERNAME")
    env_p = os.environ.get("COPERNICUSMARINE_SERVICE_PASSWORD")
    if env_u and env_p:
        return CMEMSCredentials(username=env_u, password=env_p)

    # Only walk up for ``.env`` when no explicit ``path`` override is given,
    # so callers passing ``path`` get a deterministic lookup.
    if path is None:
        for candidate in (Path.cwd(), *Path.cwd().parents):
            dotenv = candidate / ".env"
            if dotenv.is_file():
                parsed = _parse_kv(dotenv.read_text())
                u = parsed.get("copernicusmarine_service_username")
                p = parsed.get("copernicusmarine_service_password")
                if u and p:
                    return CMEMSCredentials(username=u, password=p)
                break

    cfg = path or Path.home() / ".cmems"
    if cfg.is_file():
        parsed = _parse_kv(cfg.read_text())
        u = parsed.get("username")
        p = parsed.get("password")
        if u and p:
            return CMEMSCredentials(username=u, password=p)

    return None

open_ch4_inventory 

open_ch4_inventory(path: str | Path, *, variable: str | None = None, units: str | None = None, source: str | None = None) -> Dataset

Open a gridded CH₄ emission inventory as a flux in kg m⁻² s⁻¹.

The flux variable is variable or, when omitted, the one data variable on both lat and lon (latitude / longitude are renamed first). Its units come from units or its units attribute and must be an area flux in :data:UNIT_FACTORS; the values are multiplied by that factor, so with f the factor, F [kg m⁻² s⁻¹] = f · F_file. Longitudes in [0, 360) are wrapped to [-180, 180), and both axes are sorted ascending.

Parameters:

Name Type Description Default
path str | Path

NetCDF file of the inventory (one sector or a total).

 required
variable str | None

Name of the flux variable; inferred when None.

 None
units str | None

Units of the flux, overriding the file's units attribute (for files that omit or misspell it).

 None
source str | None

Free-text provenance (e.g. "EDGAR v8.0"), stored in the source attribute.

 None

Returns:

Type Description
Dataset

Dataset with ch4_emission_flux on ([time,] lat, lon) in
Dataset

kg m⁻² s⁻¹, with units_in_file and conversion_factor
Dataset

attributes recording what was converted.

Raises:

Type Description
KeyError

If variable is not in the file, or no variable (or more than one) lies on lat and lon when inferring it.
ValueError

If the units are missing or not an accepted area flux.
Source code in packages/xrtoolz-products/src/xrproducts/_src/local/openers/inventory.py 
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def open_ch4_inventory(
    path: str | Path,
    *,
    variable: str | None = None,
    units: str | None = None,
    source: str | None = None,
) -> xr.Dataset:
    """Open a gridded CH₄ emission inventory as a flux in kg m⁻² s⁻¹.

    The flux variable is ``variable`` or, when omitted, the one data
    variable on both ``lat`` and ``lon`` (``latitude`` / ``longitude`` are
    renamed first). Its units come from ``units`` or its ``units``
    attribute and must be an area flux in :data:`UNIT_FACTORS`; the values
    are multiplied by that factor, so with ``f`` the factor,
    ``F [kg m⁻² s⁻¹] = f · F_file``. Longitudes in [0, 360) are wrapped to
    [-180, 180), and both axes are sorted ascending.

    Args:
        path: NetCDF file of the inventory (one sector or a total).
        variable: Name of the flux variable; inferred when ``None``.
        units: Units of the flux, overriding the file's ``units``
            attribute (for files that omit or misspell it).
        source: Free-text provenance (e.g. ``"EDGAR v8.0"``), stored in
            the ``source`` attribute.

    Returns:
        Dataset with ``ch4_emission_flux`` on ``([time,] lat, lon)`` in
        kg m⁻² s⁻¹, with ``units_in_file`` and ``conversion_factor``
        attributes recording what was converted.

    Raises:
        KeyError: If ``variable`` is not in the file, or no variable (or
            more than one) lies on ``lat`` and ``lon`` when inferring it.
        ValueError: If the units are missing or not an accepted area flux.
    """
    ds = promote_lonlat(xr.open_dataset(path))
    name = variable if variable is not None else _infer_flux_variable(ds, path)
    if name not in ds.data_vars:
        raise KeyError(f"{path!s} has no variable {name!r}")
    flux = ds[name]
    units_in_file = units if units is not None else flux.attrs.get("units")
    factor = _factor(units_in_file, name)

    flux = _wrap_and_sort(flux * factor)
    # Overwrite: under `xr.set_options(keep_attrs=True)` the product above
    # keeps the file's `units`, which would mislabel the converted values.
    flux = apply_cf_attrs(
        flux.rename(CH4_EMISSION_FLUX.name), CH4_EMISSION_FLUX, overwrite=True
    )
    flux.attrs["units_in_file"] = str(units_in_file)
    flux.attrs["conversion_factor"] = factor
    out = flux.to_dataset()
    out.attrs = dict(ds.attrs)
    if source is not None:
        out.attrs["source"] = source
    return out

open_emit_ch4_l2b 

open_emit_ch4_l2b(path: str | Path, *, glt_path: str | Path | None = None) -> Dataset

Open an EMIT L2B CH4 enhancement scene as a flat CF Dataset.

The raster carries the plume enhancement (ppm m) as either methane_plume_complex (the distributed variable name) or ch4_enhancement (the registry alias) — renamed to ch4_enhancement — and, when present, ch4_uncertainty on (y, x). With glt_path the scene is orthorectified through EMIT's geometric lookup table: glt_x / glt_y hold the 1-based raw-pixel column / row that fills each orthorectified cell (0 marks no data), and the 2-D lon / lat of the orthorectified grid come from the GLT file's own lon / lat variables or, failing that, its GDAL geotransform attribute (the full six-coefficient affine, rotation terms included). Without a GLT, 2-D lon / lat are taken from the raster file itself when it has them. A scalar time variable on the raster survives orthorectification and becomes the scene timestamp.

Parameters:

Name Type Description Default
path str | Path

NetCDF raster with methane_plume_complex / ch4_enhancement on (y, x) (EMIT's downtrack / crosstrack dims are renamed).

 required
glt_path str | Path | None

Optional geometric lookup table NetCDF.

 None

Returns:

Type Description
Dataset

Scene Dataset on (time, y, x) (time of length 1 when the
Dataset

file exposes a timestamp) with 2-D lon / lat coordinates.

Raises:

Type Description
KeyError

If neither methane_plume_complex nor ch4_enhancement is in the raster.
ValueError

If the GLT exposes neither lon/lat nor a geotransform.
Source code in packages/xrtoolz-products/src/xrproducts/_src/local/openers/emit.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def open_emit_ch4_l2b(
    path: str | Path,
    *,
    glt_path: str | Path | None = None,
) -> xr.Dataset:
    """Open an EMIT L2B CH4 enhancement scene as a flat CF Dataset.

    The raster carries the plume enhancement (ppm m) as either
    ``methane_plume_complex`` (the distributed variable name) or
    ``ch4_enhancement`` (the registry alias) — renamed to
    ``ch4_enhancement`` — and, when present, ``ch4_uncertainty`` on
    ``(y, x)``. With ``glt_path`` the scene is
    orthorectified through EMIT's geometric lookup table: ``glt_x`` /
    ``glt_y`` hold the 1-based raw-pixel column / row that fills each
    orthorectified cell (``0`` marks no data), and the 2-D ``lon`` /
    ``lat`` of the orthorectified grid come from the GLT file's own
    ``lon`` / ``lat`` variables or, failing that, its GDAL ``geotransform``
    attribute (the full six-coefficient affine, rotation terms included).
    Without a GLT, 2-D ``lon`` / ``lat`` are taken from the raster file
    itself when it has them. A scalar ``time`` variable on the raster
    survives orthorectification and becomes the scene timestamp.

    Args:
        path: NetCDF raster with ``methane_plume_complex`` /
            ``ch4_enhancement`` on ``(y, x)`` (EMIT's ``downtrack`` /
            ``crosstrack`` dims are renamed).
        glt_path: Optional geometric lookup table NetCDF.

    Returns:
        Scene Dataset on ``(time, y, x)`` (``time`` of length 1 when the
        file exposes a timestamp) with 2-D ``lon`` / ``lat`` coordinates.

    Raises:
        KeyError: If neither ``methane_plume_complex`` nor
            ``ch4_enhancement`` is in the raster.
        ValueError: If the GLT exposes neither ``lon``/``lat`` nor a
            ``geotransform``.
    """
    ds = xr.open_dataset(path)
    ds = ds.rename({k: v for k, v in _RASTER_DIMS.items() if k in ds.dims})
    candidates = (CH4_ENHANCEMENT.for_source("emit"), _PLUME_COMPLEX)
    name = next((n for n in candidates if n in ds.data_vars), None)
    if name is None:
        raise KeyError(f"{path!s} has none of {candidates!r}")
    ds = ds.rename({name: CH4_ENHANCEMENT.name}) if name != CH4_ENHANCEMENT.name else ds
    if glt_path is not None:
        ds = _orthorectify(ds, xr.open_dataset(glt_path))
    ds = promote_lonlat(ds)
    ds[CH4_ENHANCEMENT.name] = apply_cf_attrs(ds[CH4_ENHANCEMENT.name], CH4_ENHANCEMENT)
    return attach_scene_time(ds)

open_ghgsat_ch4_l2 

open_ghgsat_ch4_l2(path: str | Path) -> Dataset

Open a GHGSat per-plume NetCDF as a flat CF Dataset.

Releases carry xch4 (ppb) and/or ch4_enhancement on (y, x) with 2-D latitude / longitude; both are renamed through the registry (ghgsat alias, falling back to the canonical name) and the geolocation becomes lon / lat coordinates.

Parameters:

Name Type Description Default
path str | Path

Per-plume NetCDF file.

 required

Returns:

Type Description
Dataset

Scene Dataset on (time, y, x) (time of length 1 when the
Dataset

file exposes a timestamp) with 2-D lon / lat coordinates.

Raises:

Type Description
KeyError

If neither xch4 nor ch4_enhancement is present.
Source code in packages/xrtoolz-products/src/xrproducts/_src/local/openers/ghgsat.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def open_ghgsat_ch4_l2(path: str | Path) -> xr.Dataset:
    """Open a GHGSat per-plume NetCDF as a flat CF Dataset.

    Releases carry ``xch4`` (ppb) and/or ``ch4_enhancement`` on ``(y, x)``
    with 2-D ``latitude`` / ``longitude``; both are renamed through the
    registry (``ghgsat`` alias, falling back to the canonical name) and
    the geolocation becomes ``lon`` / ``lat`` coordinates.

    Args:
        path: Per-plume NetCDF file.

    Returns:
        Scene Dataset on ``(time, y, x)`` (``time`` of length 1 when the
        file exposes a timestamp) with 2-D ``lon`` / ``lat`` coordinates.

    Raises:
        KeyError: If neither ``xch4`` nor ``ch4_enhancement`` is present.
    """
    ds = xr.open_dataset(path)
    found = False
    for var in (XCH4, CH4_ENHANCEMENT):
        raw = var.for_source("ghgsat")
        if raw not in ds.data_vars:
            continue
        found = True
        if raw != var.name:
            ds = ds.rename({raw: var.name})
        ds[var.name] = apply_cf_attrs(ds[var.name], var)
    if not found:
        raise KeyError(
            f"{path!s} has neither {XCH4.for_source('ghgsat')!r} nor "
            f"{CH4_ENHANCEMENT.for_source('ghgsat')!r}"
        )
    return attach_scene_time(promote_lonlat(ds))

open_tropomi_ch4_l2 

open_tropomi_ch4_l2(path: str | Path, *, qa_min: float | None = 0.5, keep_groups: bool = False) -> Dataset

Flatten a TROPOMI CH4 L2 granule into one CF Dataset.

The PRODUCT group and its three SUPPORT_DATA children are read with :func:xarray.open_datatree and merged on the shared swath dims (time, scanline, ground_pixel) (plus layer / level / corner where a variable carries them). Variables are renamed through the registry's tropomi aliases (methane_mixing_ratio -> xch4, ..._bias_corrected -> xch4_bias_corrected, ..._precision -> xch4_precision, column_averaging_kernel, methane_profile_apriori -> ch4_profile_apriori, dry_air_subcolumns, qa_value, surface_pressure -> sp); pressure_interval keeps its product name. latitude / longitude become the 2-D lat / lon coordinates.

Time is decoded as in the product: the length-1 time dimension keeps the granule reference time (PRODUCT/time, seconds since 2010-01-01) as its index coordinate, and the per-scanline delta_time offset is folded into a scanline_time coordinate on (time, scanline) (datetime64), replacing delta_time. The reference index is what multi-granule concatenation and sel(time=...) use; scanline_time is the measurement time of each scanline.

Parameters:

Name Type Description Default
path str | Path

Path to the S5P_*_L2__CH4____*.nc granule.

 required
qa_min float | None

Pixels with qa_value < qa_min are set to NaN in every data variable (the PUM recommends 0.5). None keeps all pixels.

 0.5
keep_groups bool

When True the flattened Dataset keeps the raw product variable names (no registry renaming, no CF attribute stamping); the time decoding and QA screening still apply.

 False

Returns:

Type Description
Dataset

Flat Dataset on ``(time, scanline, ground_pixel[, layer, level,
Dataset

corner])withlon/lat/scanline_time`` coordinates.

Raises:

Type Description
KeyError

If the file has no PRODUCT group.
Source code in packages/xrtoolz-products/src/xrproducts/_src/local/openers/tropomi.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def open_tropomi_ch4_l2(
    path: str | Path,
    *,
    qa_min: float | None = 0.5,
    keep_groups: bool = False,
) -> xr.Dataset:
    """Flatten a TROPOMI CH4 L2 granule into one CF Dataset.

    The ``PRODUCT`` group and its three ``SUPPORT_DATA`` children are read
    with :func:`xarray.open_datatree` and merged on the shared swath dims
    ``(time, scanline, ground_pixel)`` (plus ``layer`` / ``level`` /
    ``corner`` where a variable carries them). Variables are renamed
    through the registry's ``tropomi`` aliases (``methane_mixing_ratio``
    -> ``xch4``, ``..._bias_corrected`` -> ``xch4_bias_corrected``,
    ``..._precision`` -> ``xch4_precision``, ``column_averaging_kernel``,
    ``methane_profile_apriori`` -> ``ch4_profile_apriori``,
    ``dry_air_subcolumns``, ``qa_value``, ``surface_pressure`` -> ``sp``);
    ``pressure_interval`` keeps its product name. ``latitude`` /
    ``longitude`` become the 2-D ``lat`` / ``lon`` coordinates.

    Time is decoded as in the product: the length-1 ``time`` dimension
    keeps the granule reference time (``PRODUCT/time``, seconds since
    2010-01-01) as its index coordinate, and the per-scanline
    ``delta_time`` offset is folded into a ``scanline_time`` coordinate on
    ``(time, scanline)`` (``datetime64``), replacing ``delta_time``. The
    reference index is what multi-granule concatenation and
    ``sel(time=...)`` use; ``scanline_time`` is the measurement time of
    each scanline.

    Args:
        path: Path to the ``S5P_*_L2__CH4____*.nc`` granule.
        qa_min: Pixels with ``qa_value < qa_min`` are set to NaN in every
            data variable (the PUM recommends ``0.5``). ``None`` keeps
            all pixels.
        keep_groups: When ``True`` the flattened Dataset keeps the raw
            product variable names (no registry renaming, no CF attribute
            stamping); the time decoding and QA screening still apply.

    Returns:
        Flat Dataset on ``(time, scanline, ground_pixel[, layer, level,
        corner])`` with ``lon`` / ``lat`` / ``scanline_time`` coordinates.

    Raises:
        KeyError: If the file has no ``PRODUCT`` group.
    """
    tree = xr.open_datatree(path)
    groups = {g.lstrip("/") for g in tree.groups}
    if _PRODUCT not in groups:
        raise KeyError(f"{path!s} has no {_PRODUCT!r} group")
    parts = [tree[_PRODUCT].to_dataset()]
    parts.extend(tree[g].to_dataset() for g in _SUPPORT_GROUPS if g in groups)
    ds = xr.merge(parts, compat="override", join="exact", combine_attrs="override")
    ds.attrs = dict(tree.attrs)
    ds = _decode_scanline_time(ds)
    if not keep_groups:
        ds = _rename_canonical(ds)
    if qa_min is not None:
        ds = subset_where(
            ds, QA_VALUE.for_source("tropomi"), qa_min, np.inf, drop=False
        )
    return ds

apply_cf_attrs 

apply_cf_attrs(da: DataArray, variable: str | Variable, *, overwrite: bool = False) -> DataArray

Stamp CF attributes from variable onto da.

By default, existing attrs are kept. Set overwrite=True to replace them with the canonical values from the registry.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/validation.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def apply_cf_attrs(
    da: xr.DataArray,
    variable: str | Variable,
    *,
    overwrite: bool = False,
) -> xr.DataArray:
    """Stamp CF attributes from ``variable`` onto ``da``.

    By default, existing attrs are kept. Set ``overwrite=True`` to
    replace them with the canonical values from the registry.
    """
    var = resolve(variable)
    # Shallow copy: we only mutate ``attrs`` below, so there's no reason
    # to duplicate the underlying array buffer.
    out = da.copy(deep=False)
    for k, v in var.cf_attrs().items():
        if overwrite or k not in out.attrs:
            out.attrs[k] = v
    return out

register 

register(variable: Variable) -> Variable

Add variable to :data:REGISTRY (keyed on :attr:Variable.name).

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/variable.py 
1411
1412
1413
1414
def register(variable: Variable) -> Variable:
    """Add ``variable`` to :data:`REGISTRY` (keyed on :attr:`Variable.name`)."""
    REGISTRY[variable.name] = variable
    return variable

resolve 

resolve(v: str | Variable) -> Variable

Return a :class:Variable given an instance or its canonical name.

Raises:

Type Description
KeyError

if v is a string not present in :data:REGISTRY.
Source code in packages/xrtoolz-products/src/xrproducts/types/_src/variable.py 
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
def resolve(v: str | Variable) -> Variable:
    """Return a :class:`Variable` given an instance or its canonical name.

    Raises:
        KeyError: if ``v`` is a string not present in :data:`REGISTRY`.
    """
    if isinstance(v, Variable):
        return v
    try:
        return REGISTRY[v]
    except KeyError as exc:
        raise KeyError(
            f"Unknown variable {v!r}. Known: {sorted(REGISTRY)}. "
            "Pass a Variable instance to use a variable outside the registry."
        ) from exc

subset_bbox 

subset_bbox(ds: Dataset, lon_bnds: tuple[float, float], lat_bnds: tuple[float, float], lon: str = 'lon', lat: str = 'lat') -> Dataset

Keep points whose (lon, lat) fall inside a bounding box.

Works with both 1-D rectilinear grids (where sel with slices would also work) and 2-D lon/lat coordinate arrays, because it masks via where(..., drop=True).

Parameters:

Name Type Description Default
ds Dataset

Input dataset.

 required
lon_bnds tuple[float, float]

(lon_min, lon_max).

 required
lat_bnds tuple[float, float]

(lat_min, lat_max).

 required
lon str

Name of the longitude coordinate.

 'lon'
lat str

Name of the latitude coordinate.

 'lat'

Returns:

Type Description
Dataset

Dataset restricted to the bounding box.
Source code in packages/xrtoolz-products/src/xrproducts/types/_src/subset.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def subset_bbox(
    ds: xr.Dataset,
    lon_bnds: tuple[float, float],
    lat_bnds: tuple[float, float],
    lon: str = "lon",
    lat: str = "lat",
) -> xr.Dataset:
    """Keep points whose ``(lon, lat)`` fall inside a bounding box.

    Works with both 1-D rectilinear grids (where ``sel`` with slices would
    also work) and 2-D lon/lat coordinate arrays, because it masks via
    ``where(..., drop=True)``.

    Args:
        ds: Input dataset.
        lon_bnds: ``(lon_min, lon_max)``.
        lat_bnds: ``(lat_min, lat_max)``.
        lon: Name of the longitude coordinate.
        lat: Name of the latitude coordinate.

    Returns:
        Dataset restricted to the bounding box.
    """
    lon_min, lon_max = lon_bnds
    lat_min, lat_max = lat_bnds
    mask = (
        (ds[lon] >= lon_min)
        & (ds[lon] <= lon_max)
        & (ds[lat] >= lat_min)
        & (ds[lat] <= lat_max)
    )
    return ds.where(mask, drop=True)

subset_time 

subset_time(ds: Dataset, time_min: str | datetime64 | Timestamp, time_max: str | datetime64 | Timestamp, time: str = 'time') -> Dataset

Restrict a dataset to the time range [time_min, time_max].

Parameters:

Name Type Description Default
ds Dataset

Input dataset.

 required
time_min str | datetime64 | Timestamp

Lower bound, any type xarray accepts in a slice (str, numpy.datetime64, pandas.Timestamp, ...).

 required
time_max str | datetime64 | Timestamp

Upper bound, same types as time_min.

 required
time str

Name of the time coordinate.

 'time'

Returns:

Type Description
Dataset

Dataset sliced along the time dimension.
Source code in packages/xrtoolz-products/src/xrproducts/types/_src/subset.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def subset_time(
    ds: xr.Dataset,
    time_min: str | np.datetime64 | pd.Timestamp,
    time_max: str | np.datetime64 | pd.Timestamp,
    time: str = "time",
) -> xr.Dataset:
    """Restrict a dataset to the time range ``[time_min, time_max]``.

    Args:
        ds: Input dataset.
        time_min: Lower bound, any type xarray accepts in a ``slice``
            (``str``, ``numpy.datetime64``, ``pandas.Timestamp``, ...).
        time_max: Upper bound, same types as ``time_min``.
        time: Name of the time coordinate.

    Returns:
        Dataset sliced along the time dimension.
    """
    return ds.sel({time: slice(time_min, time_max)})

subset_where 

subset_where(ds: Dataset, variable: str, min_val: float, max_val: float, drop: bool = True) -> Dataset

Keep points where variable is in [min_val, max_val].

Parameters:

Name Type Description Default
ds Dataset

Input dataset.

 required
variable str

Name of the variable used to build the mask.

 required
min_val float

Lower bound (inclusive).

 required
max_val float

Upper bound (inclusive).

 required
drop bool

Whether to drop cells outside the range (default True); if False, masked cells are replaced with NaN.

 True

Returns:

Type Description
Dataset

Dataset with cells outside the range masked or dropped.
Source code in packages/xrtoolz-products/src/xrproducts/types/_src/subset.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def subset_where(
    ds: xr.Dataset,
    variable: str,
    min_val: float,
    max_val: float,
    drop: bool = True,
) -> xr.Dataset:
    """Keep points where ``variable`` is in ``[min_val, max_val]``.

    Args:
        ds: Input dataset.
        variable: Name of the variable used to build the mask.
        min_val: Lower bound (inclusive).
        max_val: Upper bound (inclusive).
        drop: Whether to drop cells outside the range (default ``True``);
            if ``False``, masked cells are replaced with NaN.

    Returns:
        Dataset with cells outside the range masked or dropped.
    """
    mask = (ds[variable] >= float(min_val)) & (ds[variable] <= float(max_val))
    return ds.where(mask, drop=drop)

validate_dataset 

validate_dataset(ds: Dataset, variables: list[str | Variable], **kwargs: bool) -> ValidationReport

Validate multiple variables in a dataset.

Missing variables are reported as errors; present ones are checked via :func:validate_variable. Keyword arguments are forwarded.

Source code in packages/xrtoolz-products/src/xrproducts/types/_src/validation.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def validate_dataset(
    ds: xr.Dataset,
    variables: list[str | Variable],
    **kwargs: bool,
) -> ValidationReport:
    """Validate multiple variables in a dataset.

    Missing variables are reported as errors; present ones are checked
    via :func:`validate_variable`. Keyword arguments are forwarded.
    """
    report = ValidationReport()
    for spec in variables:
        var = resolve(spec)
        if var.name not in ds.data_vars:
            report.add(
                Issue(
                    variable=var.name,
                    severity=Severity.ERROR,
                    code="missing_variable",
                    message=(
                        f"{var.name!r} not in dataset; have {sorted(ds.data_vars)}"
                    ),
                )
            )
            continue
        sub = validate_variable(ds[var.name], var, **kwargs)
        report.issues.extend(sub.issues)
    return report

validate_variable 

validate_variable(da: DataArray, variable: str | Variable, *, check_range: bool = True, check_units: bool = True, check_standard_name: bool = True, check_dtype: bool = False) -> ValidationReport

Validate a single :class:xarray.DataArray against a :class:Variable.

Parameters:

Name Type Description Default
da DataArray

Array under test.

 required
variable str | Variable

Either a :class:Variable or a name looked up in :data:xrproducts.types.REGISTRY.

 required
check_range bool

If True and the variable defines valid_range, flag finite values falling outside as errors.

 True
check_units bool

Compare da.attrs["units"] to the canonical units. Missing units is a warning; mismatch is an error.

 True
check_standard_name bool

Compare da.attrs["standard_name"] to the canonical standard name. Missing is a warning; mismatch is an error.

 True
check_dtype bool

Require da.dtype == variable.dtype (opt-in).

 False

Returns:

Name Type Description
A ValidationReport

class:ValidationReport aggregating all findings.
Source code in packages/xrtoolz-products/src/xrproducts/types/_src/validation.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def validate_variable(
    da: xr.DataArray,
    variable: str | Variable,
    *,
    check_range: bool = True,
    check_units: bool = True,
    check_standard_name: bool = True,
    check_dtype: bool = False,
) -> ValidationReport:
    """Validate a single :class:`xarray.DataArray` against a :class:`Variable`.

    Args:
        da: Array under test.
        variable: Either a :class:`Variable` or a name looked up in
            :data:`xrproducts.types.REGISTRY`.
        check_range: If ``True`` and the variable defines ``valid_range``,
            flag finite values falling outside as errors.
        check_units: Compare ``da.attrs["units"]`` to the canonical
            units. Missing ``units`` is a warning; mismatch is an error.
        check_standard_name: Compare ``da.attrs["standard_name"]`` to
            the canonical standard name. Missing is a warning; mismatch
            is an error.
        check_dtype: Require ``da.dtype == variable.dtype`` (opt-in).

    Returns:
        A :class:`ValidationReport` aggregating all findings.
    """
    var = resolve(variable)
    report = ValidationReport()

    if check_standard_name and var.standard_name is not None:
        attr = da.attrs.get("standard_name")
        if attr is None:
            report.add(
                Issue(
                    variable=var.name,
                    severity=Severity.WARNING,
                    code="missing_standard_name",
                    message=f"no standard_name; expected {var.standard_name!r}",
                )
            )
        elif attr != var.standard_name:
            report.add(
                Issue(
                    variable=var.name,
                    severity=Severity.ERROR,
                    code="wrong_standard_name",
                    message=f"standard_name={attr!r}, expected {var.standard_name!r}",
                )
            )

    if check_units and var.units is not None:
        attr = da.attrs.get("units")
        if attr is None:
            report.add(
                Issue(
                    variable=var.name,
                    severity=Severity.WARNING,
                    code="missing_units",
                    message=f"no units; expected {var.units!r}",
                )
            )
        elif attr != var.units:
            report.add(
                Issue(
                    variable=var.name,
                    severity=Severity.ERROR,
                    code="wrong_units",
                    message=f"units={attr!r}, expected {var.units!r}",
                )
            )

    if check_range and var.valid_range is not None:
        lo, hi = var.valid_range
        finite_vals = da.where(np.isfinite(da))
        # Reduce once, eagerly. `.item()` is not implemented for dask
        # arrays, and lazily opened reader output is routinely
        # dask-backed, so compute the three reductions together rather
        # than scalarizing a lazy 0-d DataArray.
        counted, lo_val, hi_val = (
            finite_vals.count(),
            finite_vals.min(),
            finite_vals.max(),
        )
        if counted.chunks is not None:  # dask-backed
            counted, lo_val, hi_val = (v.compute() for v in (counted, lo_val, hi_val))
        n_finite = int(counted.values)
        mn = float(lo_val.values) if n_finite > 0 else None
        mx = float(hi_val.values) if n_finite > 0 else None
        if mn is not None and (mn < lo or (mx is not None and mx > hi)):
            report.add(
                Issue(
                    variable=var.name,
                    severity=Severity.ERROR,
                    code="out_of_range",
                    message=(
                        f"values in [{mn:.3g}, {mx:.3g}] fall outside "
                        f"valid_range [{lo}, {hi}]"
                    ),
                )
            )

    if check_dtype and var.dtype is not None and str(da.dtype) != var.dtype:
        report.add(
            Issue(
                variable=var.name,
                severity=Severity.ERROR,
                code="wrong_dtype",
                message=f"dtype={da.dtype}, expected {var.dtype}",
            )
        )

    return report