diff --git a/automated_ingestion/sources/san_acacia/adapter.py b/automated_ingestion/sources/san_acacia/adapter.py index 0d5d89731..d02bedecf 100644 --- a/automated_ingestion/sources/san_acacia/adapter.py +++ b/automated_ingestion/sources/san_acacia/adapter.py @@ -13,6 +13,92 @@ # See the License for the specific language governing permissions and # limitations under the License. # =============================================================================== -"""Van Essen records to Ocotillo structures. Implemented under BDMS task 3.1.""" +""" +Van Essen mapping rules for the San Acacia source. + +The adapter is the only place the vendor's vocabulary meets Ocotillo's. It is +pure enough to test without a database: it takes raw records and returns +structures, and the loader turns those into rows. + +Per-record failure isolation matches the rest of the pipeline. One unparseable +reading costs that reading, not the series -- a diver that logs one bad row +should not lose a month of good ones. +""" + +from collections.abc import Iterable, Iterator +from typing import Any + +from domain.van_essen import ( + GROUND_SURFACE_REFERENCE, + MEASUREMENT_UNIT, + VanEssenMappingError, + depth_to_water_ft, + external_point_key, + parse_reading_timestamp, +) + +from automated_ingestion.ocotillo.adapter import SourceAdapter +from automated_ingestion.ocotillo.structs import ObservationRecord + + +class SanAcaciaAdapter(SourceAdapter): + """Maps Diver-HUB water levels onto Ocotillo observations.""" + + def __init__(self) -> None: + self.failures: list[dict[str, Any]] = [] + + @property + def source_key(self) -> str: + return "san_acacia" + + def to_observations( + self, records: Iterable[dict[str, Any]] + ) -> Iterator[ObservationRecord]: + """Convert raw rows, collecting per-record failures rather than raising. + + Rows whose ``reference`` is not ground surface are refused outright. The + datum is chosen at request time and cannot be recovered from the row, so + accepting one would mean storing a number whose meaning is unknown -- + the single failure this pipeline must not produce quietly. + """ + for record in records: + try: + yield self._to_observation(record) + except VanEssenMappingError as exc: + self.failures.append({"record": _identify(record), "error": str(exc)}) + + def _to_observation(self, record: dict[str, Any]) -> ObservationRecord: + reference = record.get("reference") + if reference != GROUND_SURFACE_REFERENCE: + raise VanEssenMappingError( + f"Reading was fetched with reference={reference!r}, not " + f"{GROUND_SURFACE_REFERENCE} (ground surface). Its datum is not " + "recoverable from the row." + ) + + unit = record.get("unit") + if unit != "cm": + raise VanEssenMappingError( + f"Reading unit is {unit!r}, expected 'cm'. Converting a value " + "whose unit is not what it claims would be wrong by a factor." + ) + + point_id = record.get("monitoring_point_id") + value = depth_to_water_ft(record.get("level")) + if value is None: + raise VanEssenMappingError("Reading has no level; nothing to store.") + + return ObservationRecord( + external_point_id=external_point_key(point_id), + observation_datetime=parse_reading_timestamp(record.get("dateAndTime")), + value=value, + units=MEASUREMENT_UNIT, + ) + + +def _identify(record: dict[str, Any]) -> str: + """A short handle for a failed record, for logs and metadata.""" + return f"{record.get('monitoring_point_id')}@{record.get('dateAndTime')}" + # ============= EOF ============================================= diff --git a/automated_ingestion/tests/test_san_acacia_adapter.py b/automated_ingestion/tests/test_san_acacia_adapter.py new file mode 100644 index 000000000..a4e074a02 --- /dev/null +++ b/automated_ingestion/tests/test_san_acacia_adapter.py @@ -0,0 +1,71 @@ +# =============================================================================== +# Copyright 2026 ross +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =============================================================================== +""" +Adapter behaviour: what it refuses, and that one bad row costs only that row. +""" + +from automated_ingestion.sources.san_acacia.adapter import SanAcaciaAdapter + + +def _row(**overrides): + row = { + "monitoring_point_id": 40, + "dateAndTime": "2024-10-30T20:00:00", + "level": 471.518, + "unit": "cm", + "reference": 3, + } + row.update(overrides) + return row + + +def test_maps_a_good_row(): + [observation] = list(SanAcaciaAdapter().to_observations([_row()])) + assert observation.external_point_id == "sanacaciareach-40" + assert observation.value == 15.469751 + assert observation.units == "ft" + + +def test_wrong_datum_is_refused(): + # The datum is chosen at request time and cannot be recovered from the row, + # so a reading fetched against another reference has unknown meaning. + adapter = SanAcaciaAdapter() + assert list(adapter.to_observations([_row(reference=1)])) == [] + assert "not 3" in adapter.failures[0]["error"] + + +def test_unexpected_unit_is_refused(): + # Converting a value whose unit is not what it claims is wrong by a factor + # of 30.48 and still looks like a plausible depth. + adapter = SanAcaciaAdapter() + assert list(adapter.to_observations([_row(unit="ft")])) == [] + assert "expected 'cm'" in adapter.failures[0]["error"] + + +def test_one_bad_row_does_not_lose_the_others(): + adapter = SanAcaciaAdapter() + rows = [_row(), _row(dateAndTime="broken"), _row(dateAndTime="2024-10-30T21:00:00")] + assert len(list(adapter.to_observations(rows))) == 2 + assert len(adapter.failures) == 1 + + +def test_failures_identify_the_record(): + adapter = SanAcaciaAdapter() + list(adapter.to_observations([_row(level=None)])) + assert adapter.failures[0]["record"] == "40@2024-10-30T20:00:00" + + +# ============= EOF ============================================= diff --git a/docs/automated-ingestion-pipeline-plan.md b/docs/automated-ingestion-pipeline-plan.md index 7a3773081..190625b10 100644 --- a/docs/automated-ingestion-pipeline-plan.md +++ b/docs/automated-ingestion-pipeline-plan.md @@ -243,16 +243,26 @@ Where this stops resembling Aqueduct: the destination is a relational database w ### 3.1 — Domain layer: Van Essen record → Ocotillo model -Per `ADR4.md`, `domain/` imports nothing from `api/`, `db/`, `schemas/`, `services/`, and no fastapi/sqlalchemy/pydantic/httpx. +Built. `domain/van_essen.py` plus `sources/san_acacia/adapter.py`, 28 tests, no database and no network. -`domain/van_essen.py`, pure functions: -- `drillingDepth` cm → ft (÷ 30.48), reusing `domain/units.py` where it fits -- reading timestamp → tz-aware UTC `datetime` -- `gs` reading → DTW below ground surface, feet (datum fixed — see Epic) -- `lat`/`lng` → WGS84 point (SRID 4326) -- deterministic external key per well and per series, so repeat runs resolve to the same records +**Scope is smaller than this section originally claimed.** The draft called for converting `drillingDepth` from centimetres and building a WGS84 point from `lat`/`lng`. The live `MonitoringPoint` payload is `{id, name}` — no depth, no coordinates — so those functions would have had no input. Well geometry and construction come from the Ocotillo records a point reconciles against, which is consistent with ingestion never creating wells. -Plus an adapter in Aqueduct's `BaseAdapter` shape, with the same per-record failure isolation: a bad record is logged and counted, never fatal. Domain errors subclass `ValueError`, matching the CSV importers' per-row contract. Tests need no database and no network. Every value the mapping *invents* rather than reads is listed in the module docstring with its justification. +What the layer actually does: + +- ✅ Reading timestamp → timezone-aware UTC. A naive value is read as UTC, since the API documents UTC and does not always mark it; reading it as local would shift every observation by the machine's offset, and differently on a laptop than in a container. +- ✅ Centimetres → feet via `domain/units.convert_cm_to_ft`. +- ✅ Deterministic external keys, built from the vendor's **numeric** id rather than the name. `SO-0125` is a Bureau point id and can be corrected; the numeric id is what a re-run must resolve to the same record. The series key names the datum, because a point may later carry temperature or conductivity — both already in the vendor's raw payload. +- ✅ Errors subclass `ValueError`, matching the per-row contract the CSV importers expect. +- ✅ ADR4 layering verified by test rather than by inspection: importing `domain.van_essen` pulls in no `fastapi`, `sqlalchemy`, `pydantic`, `httpx`, `db`, `api`, `schemas`, or `services`. + +**The adapter refuses two things outright**, both because accepting them would produce plausible numbers rather than an error: + +- A row whose `reference` is not 3. The datum is chosen at request time and cannot be recovered from the row. +- A row whose `unit` is not `cm`. Converting a value whose unit is not what it claims is wrong by a factor of 30.48 and still reads as a plausible depth. + +Per-record failures are collected, not raised: one unparseable reading costs that reading, not the series. + +The module docstring lists every value the mapping **invents** rather than reads — the datum, the unit, and the timezone — since inventing is where a mapping goes quietly wrong. ### 3.2 — Bootstrap reference data: reconcile wells, seed parameter, sensor, deployments diff --git a/domain/van_essen.py b/domain/van_essen.py new file mode 100644 index 000000000..9278318b2 --- /dev/null +++ b/domain/van_essen.py @@ -0,0 +1,146 @@ +# =============================================================================== +# Copyright 2026 ross +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =============================================================================== +""" +Rules for turning Van Essen diver readings into Ocotillo values. + +The vendor's Diver-HUB API reports water levels in centimetres against a datum +chosen by the request, and Ocotillo stores feet below ground surface. These +functions do that conversion and nothing else: no database, no HTTP, no vendor +client. The caller fetches; these rules decide what a fetched value means. + +**What this mapping invents rather than reads**, since inventing is where a +mapping goes quietly wrong: + +- *The datum.* The API does not say which datum a reading is on -- the request + does. Every value here is assumed to have come from a request with + ``reference=3``, established by measurement (see + ``docs/sources/san_acacia.md``). A reading fetched with any other reference and + passed through here is silently wrong, which is why the client has no default + for that parameter. +- *The unit.* No field states it. Centimetres was inferred from the elevation + reference resolving to San Acacia's ground elevation, and cross-checked + against plausible depths for a riparian piezometer. +- *The timezone.* The API documents UTC but does not always mark it, so a naive + timestamp is read as UTC rather than as local time. + +**What it deliberately does not do.** Earlier drafts had this module converting +``drillingDepth`` from centimetres and building a WGS84 point from ``lat``/ +``lng``. The live ``MonitoringPoint`` payload is ``{id, name}`` -- no depth, no +coordinates -- so those functions would have had no input. Well geometry and +construction come from the Ocotillo records a point reconciles against. + +Errors subclass ``ValueError`` so a bad record is a per-row failure to the +caller, matching what the CSV importers already expect. +""" + +import math +from datetime import datetime, timezone + +from domain.units import convert_cm_to_ft + +PROJECT_SLUG = "sanacaciareach" +"""Prefix for external identifiers. Matches the vendor's own ``uid`` form.""" + +MEASUREMENT_UNIT = "ft" +"""Ocotillo stores depth to water in feet.""" + +GROUND_SURFACE_REFERENCE = 3 +"""The ``WaterLevelReference`` these rules assume a reading was fetched with. + +Duplicated from the client deliberately: a rule that assumes a datum should +state which one, so that reading this module alone is enough to know what its +numbers mean. +""" + + +class VanEssenMappingError(ValueError): + """A record cannot be mapped. Per-row, never fatal to a run.""" + + +def parse_reading_timestamp(value: str) -> datetime: + """Parse a Diver-HUB ``dateAndTime`` into a timezone-aware UTC datetime. + + A naive timestamp is read as UTC. Reading it as local time would shift every + observation by the machine's offset -- and would do so differently on a + developer's laptop and in a container, which is the kind of discrepancy that + survives review. + """ + if not isinstance(value, str) or not value.strip(): + raise VanEssenMappingError(f"Reading timestamp is missing or blank: {value!r}") + + text = value.strip().replace("Z", "+00:00") + try: + parsed = datetime.fromisoformat(text) + except ValueError as exc: + raise VanEssenMappingError( + f"Reading timestamp {value!r} is not an ISO-8601 instant." + ) from exc + + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def depth_to_water_ft(level_cm: float | None) -> float | None: + """Convert a ground-surface reading in centimetres to feet. + + ``None`` passes through: the vendor reports gaps, and a gap is not an error. + + Negative values are kept. Depth below ground surface goes negative when + water stands above ground, which happens in these riparian wells during high + flow and is real data rather than a fault. + """ + if level_cm is None: + return None + if isinstance(level_cm, bool) or not isinstance(level_cm, (int, float)): + raise VanEssenMappingError(f"Reading level is not a number: {level_cm!r}") + if math.isnan(level_cm) or math.isinf(level_cm): + raise VanEssenMappingError(f"Reading level is not finite: {level_cm!r}") + + return convert_cm_to_ft(float(level_cm)) + + +def external_point_key(monitoring_point_id: int) -> str: + """Stable identifier for a monitoring point. + + Built from the vendor's numeric id rather than its name. Names like + ``SO-0125`` are Bureau point ids and can be corrected; the numeric id is what + the vendor's URLs use and is what a re-run has to resolve to the same record. + """ + if isinstance(monitoring_point_id, bool) or not isinstance( + monitoring_point_id, int + ): + raise VanEssenMappingError( + f"Monitoring point id must be an integer: {monitoring_point_id!r}" + ) + if monitoring_point_id <= 0: + raise VanEssenMappingError( + f"Monitoring point id must be positive: {monitoring_point_id!r}" + ) + return f"{PROJECT_SLUG}-{monitoring_point_id}" + + +def external_series_key(monitoring_point_id: int) -> str: + """Stable identifier for one point's depth-to-water series. + + A point could later carry more than one series -- temperature and + conductivity are already in the vendor's raw payload -- so the datum is part + of the key rather than implied by the point. + """ + return f"{external_point_key(monitoring_point_id)}:dtw-gs" + + +# ============= EOF ============================================= diff --git a/tests/test_van_essen_domain.py b/tests/test_van_essen_domain.py new file mode 100644 index 000000000..a7c2003bd --- /dev/null +++ b/tests/test_van_essen_domain.py @@ -0,0 +1,102 @@ +# =============================================================================== +# Copyright 2026 ross +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =============================================================================== +""" +Van Essen mapping rules. + +No database, no network -- these are the rules alone, which is the point of +keeping them in `domain/`. +""" + +from datetime import datetime, timezone + +import pytest + +from domain.van_essen import ( + VanEssenMappingError, + depth_to_water_ft, + external_point_key, + external_series_key, + parse_reading_timestamp, +) + + +class TestTimestamps: + def test_naive_is_read_as_utc(self): + # The API documents UTC and does not always mark it. Reading naive as + # local would shift every observation by the machine's offset, and shift + # it differently on a laptop and in a container. + assert parse_reading_timestamp("2024-10-30T20:00:00") == datetime( + 2024, 10, 30, 20, 0, tzinfo=timezone.utc + ) + + def test_explicit_utc_matches_naive(self): + assert parse_reading_timestamp( + "2024-10-30T20:00:00Z" + ) == parse_reading_timestamp("2024-10-30T20:00:00") + + def test_offset_is_normalized_to_utc(self): + assert parse_reading_timestamp("2024-10-30T14:00:00-06:00") == datetime( + 2024, 10, 30, 20, 0, tzinfo=timezone.utc + ) + + @pytest.mark.parametrize("value", ["", " ", None, "not-a-date", "2024-13-45"]) + def test_unusable_timestamps_raise(self, value): + with pytest.raises(VanEssenMappingError): + parse_reading_timestamp(value) + + +class TestDepthConversion: + def test_centimetres_become_feet(self): + # SO-0125 on 2024-10-30: 471.518 cm below ground surface. + assert depth_to_water_ft(471.518) == 15.469751 + + def test_gap_passes_through(self): + # The vendor reports gaps. A gap is not an error. + assert depth_to_water_ft(None) is None + + def test_negative_depth_is_kept(self): + # Depth below ground goes negative when water stands above ground, which + # happens in these riparian wells at high flow. Clamping would erase + # real data. + assert depth_to_water_ft(-50.0) == pytest.approx(-1.64042, rel=1e-4) + + @pytest.mark.parametrize("value", [float("nan"), float("inf"), "471.518", True]) + def test_unusable_values_raise(self, value): + with pytest.raises(VanEssenMappingError): + depth_to_water_ft(value) + + +class TestExternalKeys: + def test_point_key_uses_the_numeric_id(self): + # Names like SO-0125 are Bureau point ids and can be corrected; the + # numeric id is what a re-run must resolve to the same record. + assert external_point_key(40) == "sanacaciareach-40" + + def test_series_key_names_the_datum(self): + # A point may later carry temperature or conductivity, both already in + # the vendor's raw payload. + assert external_series_key(40) == "sanacaciareach-40:dtw-gs" + + def test_keys_are_stable_across_calls(self): + assert external_point_key(40) == external_point_key(40) + + @pytest.mark.parametrize("value", [0, -1, "40", None, True]) + def test_unusable_ids_raise(self, value): + with pytest.raises(VanEssenMappingError): + external_point_key(value) + + +# ============= EOF =============================================