feat(ingestion): add the Van Essen domain rules and adapter

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 =============================================

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 =============================================

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
 

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 =============================================