Water Quality Analysis Utils

wqautils is a batteries-included toolkit for evaluating water-quality observations, generating indices, and shipping analysis workflows to both the command line and web services. It combines configurable regulatory standards, Pydantic data contracts, calibration and unit normalization helpers, and analytics utilities that cover time-series inspection, notifications, visualization, and reporting.

Highlights

• Configurable standards — load WHO/EPA defaults or bring your own JSON/SQLite tables and switch profiles at runtime.
• Rich data models — strongly typed WaterQualityParameters, metadata capture, and a WaterQualityStatus enum that surfaces OK vs. Alert states.
• Calibrations & unit handling — normalize mixed units (°C/°F/K, µS/mS) and apply per-sensor calibration curves before evaluation.
• Indices & analytics — compute NSF and CCME Water Quality Indices, derive moving averages, z-score anomalies, trend slopes, and sustained alert streaks.
• Ingestion & export — helpers for CSV/JSON ingestion, serialization, and visualization hooks via Matplotlib.
• Interfaces included — Typer-powered CLI for batch analysis and a FastAPI service for programmatic integration, plus a notification system with cooldown management.

Installation

Clone the repository and install dependencies with Poetry (recommended for development):

poetry install

To use the library from another project you can also install it locally:

pip install .

Quick start

from datetime import datetime

from wqautils.calibration import CalibrationSetting
from wqautils.models import WaterQualityParameters
from wqautils.utils import check_water_quality

sample = WaterQualityParameters(
	temperature=20.5,
	dissolved_oxygen=8.1,
	conductivity=1.2,
	turbidity=1.8,
	ph=7.1,
	metadata={
		"sample_id": "station-7",
		"collected_at": datetime.utcnow().isoformat(),
		"latitude": 45.1,
		"longitude": -122.7,
	},
)

results = check_water_quality(
	sample,
	standard="default",
	calibrations={"conductivity": CalibrationSetting(scale=950)},
	units={"conductivity": "ms_cm"},
)

print(results.overall_status())        # -> WaterQualityStatus.OK
print(results.nsf_wqi, results.ccme_wqi)
print(results.conductivity.explanation)

WaterQualityResults.as_dict() returns a JSON-ready payload containing metadata, per-parameter explanations, computed indices, and timestamps.

Configurable standards

Built-in standards (default, freshwater_aquatic) live under wqautils/data/standards. You can list them via the CLI:

poetry run wqautils standards

Load a bundled configuration or point to your own JSON/SQLite definition:

from wqautils.standards import load_standard, load_standard_from_sqlite

who = load_standard("default")
custom = load_standard("/path/to/custom_standard.json")
lake_rules = load_standard_from_sqlite("standards.db", name="lake-alpha")

JSON parameters accept bounds, messaging, optional labels, and allow_negative flags. See wqautils/data/standards/default.json for a template.

Indices & time-series analytics

• wqautils.indices.compute_indices returns NSF & CCME WQI scores for compatible samples.
• wqautils.timeseries provides to_dataframe, moving_average, rolling_zscore, detect_anomalies, trend_slope, and sustained_alerts helpers for pandas DataFrames.
• Visualization helpers in wqautils.visualization generate Matplotlib figures for parameter trends and anomaly overlays.

Notifications & automation

Use wqautils.notifications.AlertNotifier to emit callbacks when parameters transition into alert states. Cooldowns prevent duplicate alerts, and results are passed through a structured AlertEvent payload.

Command line interface

The Typer CLI supports batch analysis of CSV/JSON files:

poetry run wqautils analyze data/samples.csv --standard default --output results.json

Inputs can be CSV, JSON, or newline-delimited JSON. Results are printed to stdout or written to disk as pretty-printed JSON.

FastAPI service

Serve the analyzer over HTTP using Uvicorn:

poetry run uvicorn wqautils.service:app --reload

Key endpoints:

• GET /standards – list bundled standard names.
• POST /analyze – submit a list of WaterQualityParameters payloads and receive structured results.
• GET /health – lightweight health check for deployment probes.

Data ingestion & export

Utilities in wqautils.io cover CSV/JSON ingestion with schema validation and lossless export of analysis results. Pair them with wqautils.utils.check_water_quality for end-to-end pipelines.

Development & testing

Run the test suite and static checks from the project root:

poetry run pytest
poetry run ruff check

Feel free to extend the library with new standards, analytics, or notification sinks—comprehensive unit tests cover the existing surface area to help you iterate safely.