feat(ingestion): wire the loader end to end

san_acacia_observations joins the pieces that existed separately: reconcile the
vendor point to a well, choose the deployment its transducer hangs from, ask the
database where that series got to, fetch forward, map, upsert, and extend the QC
block.

The seeding half of 3.2 turned out to be nothing. All 38 wells already have
deployments, the parameter exists as `groundwater level` in feet -- the unit the
adapter emits -- and existing observations already use it. So the series is
chosen rather than created.

Choosing it needs a rule, because a well carries several open deployments: a
deployment is equipment, not a measured property. SO-0140 has a DiverLink, a
Pressure Transducer and a Diver Cable, and only the transducer produces a water
level. Picking any other would attribute a reading to a cable.

That resolves cleanly for 35 of the 38 wells. Two have two open transducers and
one has none; those are skipped and reported. Taking the lower id would be a
silent guess about equipment, and a removed transducer is not used as a fallback
-- writing current data against retired kit looks like success while being
wrong.

A well that cannot be resolved costs that well's readings for the run, not the
other thirty-seven's.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: jirhiker

automated_ingestion/defs/assets/__init__.py

@@ -27,6 +27,7 @@
 from automated_ingestion.sources.san_acacia.ingest import (
     raw_san_acacia_locations,
     raw_san_acacia_readings,
+    san_acacia_observations,
 )
 
 
@@ -37,6 +38,7 @@ def all_assets() -> list[AssetsDefinition]:
         database_connectivity,
         raw_san_acacia_locations,
         raw_san_acacia_readings,
+        san_acacia_observations,
     ]
 
 

automated_ingestion/sources/san_acacia/ingest.py

@@ -27,6 +27,7 @@
 
 from dagster import AssetExecutionContext, MetadataValue, Output, asset
 
+from automated_ingestion.defs.resources import OcotilloDatabase
 from automated_ingestion.sources.san_acacia.client import DiverHubClient
 
 
@@ -113,6 +114,189 @@ def raw_san_acacia_readings(context: AssetExecutionContext) -> Output[int]:
     )
 
 
+@asset(
+    group_name="san_acacia",
+    deps=[raw_san_acacia_readings],
+    description="Water levels mapped to the Ocotillo model and loaded to Postgres.",
+)
+def san_acacia_observations(
+    context: AssetExecutionContext, database: OcotilloDatabase
+) -> Output[int]:
+    """Load San Acacia water levels into `transducer_observation`.
+
+    Per well: match the vendor point to an Ocotillo well, choose the deployment
+    its transducer hangs from, ask the database where that series got to, fetch
+    forward from there, map, and upsert.
+
+    A well that cannot be resolved is skipped and counted, never guessed at.
+    Ingestion does not create wells or pick between candidate deployments, so an
+    unresolved well is a question for a person -- and skipping it costs that
+    well's readings for this run, not the other thirty-seven's.
+    """
+    from datetime import datetime, timezone
+
+    from automated_ingestion.ocotillo.loader import ensure_block, load_observations
+    from automated_ingestion.shared.watermark import (
+        PostgresWatermarkStore,
+        resolve_start,
+    )
+    from automated_ingestion.sources.san_acacia.adapter import SanAcaciaAdapter
+    from automated_ingestion.sources.san_acacia.client import GROUND_SURFACE_REFERENCE
+    from automated_ingestion.sources.san_acacia.dlt_pipeline import (
+        INITIAL_START,
+        PROJECT_ID,
+        READING_SPAN,
+    )
+    from automated_ingestion.sources.san_acacia.reconcile import (
+        VendorPoint,
+        reconcile,
+    )
+    from automated_ingestion.sources.san_acacia.resolve import (
+        PARAMETER_NAME,
+        resolve_deployment,
+    )
+    from domain.van_essen import parse_reading_timestamp
+
+    client = _client()
+    points = [
+        VendorPoint(monitoring_point_id=p["id"], name=p["name"])
+        for p in client.monitoring_points(PROJECT_ID)
+    ]
+    end = int(datetime.now(tz=timezone.utc).timestamp())
+    floor = parse_reading_timestamp(INITIAL_START)
+
+    rows_loaded = 0
+    skipped: list[dict[str, Any]] = []
+    adapter_failures = 0
+
+    with database.session() as session:
+        parameter_id = _parameter_id(session, PARAMETER_NAME)
+        report = reconcile(points, _well_candidates(session))
+        watermarks = PostgresWatermarkStore(session)
+
+        for match in report.matches:
+            if match.needs_a_human:
+                skipped.append({"point": match.point.name, "reason": match.kind.value})
+                continue
+
+            thing_id = match.thing_id
+            resolution = resolve_deployment(_deployments(session, thing_id))
+            if resolution.needs_a_human:
+                skipped.append(
+                    {"point": match.point.name, "reason": resolution.kind.value}
+                )
+                continue
+
+            start = resolve_start(watermarks, thing_id, parameter_id, floor)
+            adapter = SanAcaciaAdapter()
+            raw = (
+                {
+                    "monitoring_point_id": match.point.monitoring_point_id,
+                    "dateAndTime": row["dateAndTime"],
+                    "level": row["level"],
+                    "unit": "cm",
+                    "reference": GROUND_SURFACE_REFERENCE,
+                }
+                for row in client.water_levels(
+                    match.point.monitoring_point_id,
+                    int(start.timestamp()),
+                    end,
+                    reference=GROUND_SURFACE_REFERENCE,
+                    span=READING_SPAN,
+                )
+            )
+
+            observations = list(adapter.to_observations(raw))
+            adapter_failures += len(adapter.failures)
+            if not observations:
+                continue
+
+            result = load_observations(
+                session,
+                observations,
+                resolution.deployment_id,
+                parameter_id,
+                release_status="public",
+            )
+            rows_loaded += result.rows_written
+            ensure_block(
+                session,
+                thing_id=thing_id,
+                parameter_id=parameter_id,
+                start=min(o.observation_datetime for o in observations),
+                end=max(o.observation_datetime for o in observations),
+                release_status="public",
+            )
+
+    if skipped:
+        context.log.warning(
+            "%s of %s wells skipped: %s",
+            len(skipped),
+            len(points),
+            ", ".join(f"{s['point']} ({s['reason']})" for s in skipped),
+        )
+
+    return Output(
+        rows_loaded,
+        metadata={
+            "rows_loaded": MetadataValue.int(rows_loaded),
+            "wells_attempted": MetadataValue.int(len(points)),
+            "wells_skipped": MetadataValue.int(len(skipped)),
+            "adapter_failures": MetadataValue.int(adapter_failures),
+            "skipped": MetadataValue.json(skipped),
+        },
+    )
+
+
+def _parameter_id(session: Any, name: str) -> int:
+    from sqlalchemy import select
+
+    from db.parameter import Parameter
+
+    parameter_id = session.scalar(
+        select(Parameter.id).where(Parameter.parameter_name == name)
+    )
+    if parameter_id is None:
+        raise RuntimeError(
+            f"No parameter named {name!r}. Ingestion does not create parameters; "
+            "seed it before loading."
+        )
+    return parameter_id
+
+
+def _well_candidates(session: Any) -> list[Any]:
+    """Ocotillo wells the vendor points might be, narrowed by name prefix."""
+    from sqlalchemy import select
+
+    from db.thing import Thing
+
+    from automated_ingestion.sources.san_acacia.reconcile import ThingCandidate
+
+    rows = session.execute(
+        select(Thing.id, Thing.name).where(Thing.name.ilike("SO-%"))
+    ).all()
+    return [ThingCandidate(thing_id=i, name=n) for i, n in rows]
+
+
+def _deployments(session: Any, thing_id: int) -> list[Any]:
+    from sqlalchemy import select
+
+    from db.deployment import Deployment
+    from db.sensor import Sensor
+
+    from automated_ingestion.sources.san_acacia.resolve import DeploymentCandidate
+
+    rows = session.execute(
+        select(Deployment.id, Sensor.sensor_type, Deployment.removal_date)
+        .join(Sensor, Sensor.id == Deployment.sensor_id)
+        .where(Deployment.thing_id == thing_id)
+    ).all()
+    return [
+        DeploymentCandidate(deployment_id=i, sensor_type=t, removal_date=r)
+        for i, t, r in rows
+    ]
+
+
 def _row_count(load_info: Any) -> int:
     """Rows dlt reports as loaded, or 0 when it reports nothing."""
     try:

automated_ingestion/sources/san_acacia/resolve.py

@@ -0,0 +1,111 @@
+# ===============================================================================
+# 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.
+# ===============================================================================
+"""
+Choosing which deployment a water level belongs to.
+
+A San Acacia well carries several open deployments at once, because a deployment
+is a piece of equipment rather than a measured property. SO-0140 has three:
+
+    DiverLink            DN431-1ch    telemetry
+    Pressure Transducer  DI801 10m    measures the water level
+    Diver Cable          AS2006-6m    the cable
+
+Only the pressure transducer produces the reading being ingested, so that is the
+deployment an observation hangs from. Picking any of the others would attribute
+a water level to a cable.
+
+Like the reconciler, this never chooses between equally good candidates. Two
+open transducers on one well is a question about the equipment record, not
+something to resolve by taking the lower id.
+"""
+
+from collections.abc import Iterable
+from dataclasses import dataclass
+from datetime import date
+from enum import Enum
+
+WATER_LEVEL_SENSOR_TYPE = "Pressure Transducer"
+"""The sensor type whose deployment carries a water level.
+
+Checked against staging: of the 38 San Acacia wells, 35 have exactly one open
+deployment of this type, 2 have two, and 1 has none. The other types present are
+`DiverLink`, `Diver Cable` and `Barometer`, none of which measure depth to
+water.
+"""
+
+PARAMETER_NAME = "groundwater level"
+"""The Ocotillo parameter these readings are. Its `default_unit` is `ft`, which
+is what the adapter emits -- the conversion from the vendor's centimetres
+happens in `domain/van_essen.py`."""
+
+
+class ResolutionKind(str, Enum):
+    RESOLVED = "resolved"
+    AMBIGUOUS = "ambiguous"
+    MISSING = "missing"
+
+
+@dataclass(frozen=True)
+class DeploymentCandidate:
+    """A deployment on the well, with the bit needed to judge it."""
+
+    deployment_id: int
+    sensor_type: str
+    removal_date: date | None = None
+
+    @property
+    def is_open(self) -> bool:
+        return self.removal_date is None
+
+
+@dataclass(frozen=True)
+class Resolution:
+    """Which deployment to load into, or why none was chosen."""
+
+    kind: ResolutionKind
+    deployment_id: int | None = None
+    candidates: tuple[int, ...] = ()
+
+    @property
+    def needs_a_human(self) -> bool:
+        return self.kind is not ResolutionKind.RESOLVED
+
+
+def resolve_deployment(candidates: Iterable[DeploymentCandidate]) -> Resolution:
+    """Pick the open pressure-transducer deployment, or refuse.
+
+    Closed deployments are excluded rather than preferred-against: a removed
+    transducer is not where today's readings belong, and treating it as a
+    fallback would quietly write current data against retired equipment.
+    """
+    open_transducers = [
+        c for c in candidates if c.is_open and c.sensor_type == WATER_LEVEL_SENSOR_TYPE
+    ]
+
+    if len(open_transducers) == 1:
+        return Resolution(
+            kind=ResolutionKind.RESOLVED,
+            deployment_id=open_transducers[0].deployment_id,
+        )
+    if len(open_transducers) > 1:
+        return Resolution(
+            kind=ResolutionKind.AMBIGUOUS,
+            candidates=tuple(c.deployment_id for c in open_transducers),
+        )
+    return Resolution(kind=ResolutionKind.MISSING)
+
+
+# ============= EOF =============================================