CSFS — Community Streamflow Service

Live acquisition and harmonization of global streamflow observations.

CSFS connects to open streamflow data providers worldwide — national hydrological agencies, regional networks, research archives, and global model products — harmonizes their observations into one canonical station/observation schema (SI units per variable — discharge m³/s, stage m, water temperature °C — timestamps in UTC), and maintains a near-real-time DuckDB store with scheduled acquisition, health monitoring, a CLI, and a FastAPI read layer.

Documentation: https://darriey.github.io/CSFS/

Why CSFS?

Programmatic access to river discharge data is fragmented: the community relies either on static archives (GRDC, Caravan, GSIM, EStreams, CAMELS) that are frozen at publication time, or on single-agency clients (USGS dataretrieval, hydrofunctions) that each cover one network. Getting current discharge across, say, France, Brazil, and Japan means learning three APIs, three formats, and three unit conventions. CSFS provides a single interface for live, multi-provider acquisition — one connector per agency, every observation normalized to a common schema, re-acquisition scheduled to each provider's update cadence — and keeps its provider roster honest mechanically, with CI-enforced integrity tests.

Provider roster (the honest numbers)

• 98 sources cataloged in inventory/providers.yaml, labeled by readiness: 76 implemented, 8 research, 5 fallback, 5 manual, 1 degraded, 3 deprecated.
• 84 connectors registered in code — the 76 implemented entries plus 8 still labeled research/degraded while their upstream data paths are validated.
• 34 implemented providers are realtime/near-realtime; the rest are recent/archive sources, including roughly a dozen offline research archives (GRDC, Caravan, GSIM, EStreams, LamaH, CAMELS variants, ROBIN, ADHI, SIEREM).

These numbers are generated by scripts/gen_catalog.py and CI fails when they drift from the inventory and registry.

These statuses are CI-enforced: tests/test_connector_integrity.py fails the build if a connector ships without tests, lacks a scheduler tier, or if the inventory claims implemented for a connector that does not exist. See the full provider catalog.

Note: live-provider commands talk to real agency APIs and can hit transient upstream outages — a failed fetch is usually them, not you.

Install

pip install community-streamflow-service            # core
pip install "community-streamflow-service[pandas]"  # + DataFrame store queries
pip install "community-streamflow-service[api]"     # + FastAPI read layer

Requires Python 3.11+.

Quick start (CLI)

csfs providers                          # list registered providers + tiers
csfs fetch -p usgs --lookback 168 -n 50 # fetch a week of USGS data
csfs status                             # what's in the local DuckDB
csfs health                             # per-connector freshness + run health
csfs serve                              # HTTP read layer (needs the api extra)

Quick start (Python)

import asyncio

import csfs


async def main() -> None:
    async with csfs.open_store("csfs.duckdb", read_only=False) as store:
        await csfs.run_acquisition(store, providers=["usgs"], lookback_hours=48, max_stations=20)

        stations = await store.get_stations(provider="usgs", limit=5)
        # pandas DataFrame indexed by timestamp (needs the [pandas] extra);
        # get_observations() / get_observations_arrow() need no extra.
        df = await store.get_observations_df(stations[0].id)
        print(df["value"].describe())


asyncio.run(main())

Or pull one gauge's series straight from a provider, no database involved:

from datetime import UTC, datetime, timedelta

import csfs

end = datetime.now(UTC)
chunk = csfs.fetch_observations_sync("usgs", "usgs:01646500", start=end - timedelta(days=7), end=end)

The store is a plain DuckDB file — any SQL/pandas/Arrow tooling works on it directly. The blessed, stable surface is what import csfs re-exports; see the Python API guide.

SYMFLUENCE integration

CSFS doubles as a streamflow-observation plugin for SYMFLUENCE: install both packages and SYMFLUENCE auto-discovers the handler, so an experiment YAML needs only

ADDITIONAL_OBSERVATIONS: csfs
CSFS_STATION_ID: "usgs:01646500"

to calibrate against any CSFS-reachable gauge (live fetch, or offline from a CSFS store via CSFS_DB_PATH). See the SYMFLUENCE integration guide.

API keys

Most connectors need no credentials. Exceptions: norway_nve (free NVE HydAPI key) and glofas (Copernicus CDS token in ~/.cdsapirc). Keep keys out of tracked config files.

Architecture

connectors/     Provider plugins (one per data source)
core/           Canonical data models, registry, health, exceptions
store/          Persistence layer (DuckDB default)
scheduler/      Acquisition runner, cron tiers, daemon
api/            FastAPI query layer
cli/            Command-line interface
inventory/      Global provider inventory (YAML)

Details — including the roster-integrity guard system and the hermetic test policy — in the architecture docs.

Contributing

The most valuable contribution is a new provider connector. See CONTRIBUTING.md for the walkthrough and the roster-integrity requirements your PR must satisfy.

Automated CI triage

When CI fails on main, a Claude Code agent (.github/workflows/ci-autotriage.yml) reads the failure, posts a triage report as a commit comment, and classifies it:

Safety: the auto-merge workflow (autofix-automerge.yml) merges a PR only if its entire diff is within connectors//tests/ — a misclassified change can never auto-merge, regardless of label. Claude authenticates via the ANTHROPIC_API_KEY_OAUTH repo secret. Pause anytime with gh workflow disable "CI Auto-Triage" -R DarriEy/CSFS.

Labels: claude-autofix (agent-opened) · automerge-on-green (drift fix, self-merges on green) · needs-human-review (needs a human).

Citing

See CITATION.cff.

License

GPL-3.0-or-later. See LICENSE.