From c20e2e39795f161d6ae3405df3cc9f30e506e131 Mon Sep 17 00:00:00 2001 From: jakeross Date: Wed, 19 Aug 2026 08:15:52 -0700 Subject: [PATCH 1/4] feat(ingestion): reconcile San Acacia points against Ocotillo wells Task 3.2's first half: a report saying, per monitoring point, whether a matching well exists. Read-only on both sides. Also corrects the "33 wells" figure that has run through this plan from the start. It came from Aqueduct's docs/sources/san_acacia.md, in a sentence about `/locations/{projectName}` -- an endpoint that does not exist. That same document supplied the doubled /api/api/ path, the claim the source is unauthenticated, and the gs/vrd payload shape, all disproved against the live API. The count has no more standing than the rest of it, so 38 is not a discrepancy to explain but the number to use, and this was never the blocker it was treated as. Coordinate proximity, the third matching signal the plan called for, is not available: MonitoringPoint is {id, name}. That removes the only fuzzy signal and leaves two exact ones, so every match is defensible rather than probabilistic. The module never picks a winner. More than one candidate is ambiguous and escalates; none is unmatched and escalates. Ingestion does not create wells, and choosing between two plausible ones is exactly the judgement that must not be automated -- the duplicate Geographic Area groups in this database are the standing reminder. Names compare on significant characters, so SO-0125, so 0125 and SO0125 are one identifier while SO-0126 stays a different well. report.ready is false unless every point resolved, and false for empty input, because a partial load produces a series that looks complete and is not. Co-Authored-By: Claude Opus 5 --- .../scripts/reconcile_san_acacia.py | 121 +++++++++++ .../sources/san_acacia/reconcile.py | 196 ++++++++++++++++++ automated_ingestion/tests/test_reconcile.py | 136 ++++++++++++ docs/automated-ingestion-pipeline-plan.md | 39 ++-- docs/sources/san_acacia.md | 24 ++- 5 files changed, 494 insertions(+), 22 deletions(-) create mode 100644 automated_ingestion/scripts/reconcile_san_acacia.py create mode 100644 automated_ingestion/sources/san_acacia/reconcile.py create mode 100644 automated_ingestion/tests/test_reconcile.py diff --git a/automated_ingestion/scripts/reconcile_san_acacia.py b/automated_ingestion/scripts/reconcile_san_acacia.py new file mode 100644 index 000000000..2e730ef2d --- /dev/null +++ b/automated_ingestion/scripts/reconcile_san_acacia.py @@ -0,0 +1,121 @@ +# =============================================================================== +# 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. +# =============================================================================== +""" +Produce the San Acacia reconciliation report. + +Task 3.2 calls for this **before** anything is written: for each monitoring +point Diver-HUB returns, whether a matching Ocotillo well exists. Read-only on +both sides -- it fetches the vendor's point list and queries `thing`, and +changes nothing. + + export DIVERHUB_USERNAME=... DIVERHUB_PASSWORD=... + uv run --group ingestion python -m \\ + automated_ingestion.scripts.reconcile_san_acacia + +Exits non-zero when any point needs a human, so it can gate a later step +without anyone having to read the output carefully. +""" + +import sys + +from automated_ingestion.sources.san_acacia.reconcile import ( + ThingCandidate, + VendorPoint, + format_report, + reconcile, +) + + +def _vendor_points() -> list[VendorPoint]: + import requests + + from automated_ingestion.sources.san_acacia.client import DiverHubClient + from automated_ingestion.sources.san_acacia.dlt_pipeline import PROJECT_ID + + client = DiverHubClient(requests.Session()) + return [ + VendorPoint(monitoring_point_id=p["id"], name=p["name"]) + for p in client.monitoring_points(PROJECT_ID) + ] + + +def _candidates(prefix: str) -> list[ThingCandidate]: + """Wells that could plausibly be San Acacia points. + + Narrowed by name prefix rather than loading every well: the point ids are + `SO-####`, and comparing 38 names against the whole inventory would surface + coincidental matches from other prefixes without adding a real one. + """ + from sqlalchemy import select + + from db.engine import session_ctx + from db.thing import Thing + from db.thing_id_link import ThingIDLink + + with session_ctx() as session: + things = session.execute( + select(Thing.id, Thing.name).where(Thing.name.ilike(f"{prefix}%")) + ).all() + links = session.execute( + select(ThingIDLink.thing_id, ThingIDLink.alternate_id) + ).all() + + by_thing: dict[int, list[str]] = {} + for thing_id, alternate_id in links: + if alternate_id: + by_thing.setdefault(thing_id, []).append(alternate_id) + + return [ + ThingCandidate( + thing_id=thing_id, + name=name, + external_ids=tuple(by_thing.get(thing_id, ())), + ) + for thing_id, name in things + ] + + +def main() -> int: + import argparse + + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--prefix", + default="SO-", + help="Well name prefix to consider as candidates (default: SO-).", + ) + args = parser.parse_args() + + try: + points = _vendor_points() + except Exception as exc: # noqa: BLE001 - the message is the useful part + print(f"Could not list monitoring points: {exc}", file=sys.stderr) + return 2 + + candidates = _candidates(args.prefix) + print(f"Vendor points from Diver-HUB : {len(points)}") + print(f"Ocotillo wells named {args.prefix}* : {len(candidates)}\n") + + report = reconcile(points, candidates) + print(format_report(report)) + return 0 if report.ready else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) + + +# ============= EOF ============================================= diff --git a/automated_ingestion/sources/san_acacia/reconcile.py b/automated_ingestion/sources/san_acacia/reconcile.py new file mode 100644 index 000000000..e7726dfd1 --- /dev/null +++ b/automated_ingestion/sources/san_acacia/reconcile.py @@ -0,0 +1,196 @@ +# =============================================================================== +# 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. +# =============================================================================== +""" +Matching Diver-HUB monitoring points to Ocotillo wells. + +Ingestion never creates a well. A vendor point that matches nothing is a +question for a person, not a row to invent -- the duplicate Geographic Area +groups elsewhere in this database are the standing reminder that "looks like a +new record" is not proof. + +So this decides, per point, one of three things: exactly one candidate +(matched), more than one (ambiguous, escalate), or none (unmatched, escalate). +It never picks a winner among candidates. Choosing between two plausible wells +is precisely the judgement that should not be automated. + +**Matching is on identifiers only.** The plan called for coordinate proximity as +a third signal; the live ``MonitoringPoint`` payload is ``{id, name}`` and +carries no coordinates, so there is nothing to compare. That removes the one +fuzzy signal and leaves two exact ones, which is a better position to be in -- +every match here is defensible rather than probabilistic. + +The functions are pure: they take vendor points and candidate rows and return a +report. Loading the candidates is the caller's job, so the decision logic is +testable without a database. +""" + +from collections.abc import Iterable +from dataclasses import dataclass, field +from enum import Enum + + +class MatchKind(str, Enum): + """How a point was matched, or why it was not.""" + + NAME = "matched-by-name" + EXTERNAL_ID = "matched-by-external-id" + AMBIGUOUS = "ambiguous" + UNMATCHED = "unmatched" + + +@dataclass(frozen=True) +class VendorPoint: + """A monitoring point as Diver-HUB reports it.""" + + monitoring_point_id: int + name: str + + +@dataclass(frozen=True) +class ThingCandidate: + """An Ocotillo well that might be the same well.""" + + thing_id: int + name: str + external_ids: tuple[str, ...] = () + + +@dataclass(frozen=True) +class Match: + """What was decided about one vendor point.""" + + point: VendorPoint + kind: MatchKind + thing_id: int | None = None + candidates: tuple[int, ...] = () + + @property + def needs_a_human(self) -> bool: + return self.kind in (MatchKind.AMBIGUOUS, MatchKind.UNMATCHED) + + +@dataclass +class ReconciliationReport: + """The whole picture, for a person to read before anything is written.""" + + matches: list[Match] = field(default_factory=list) + + @property + def matched(self) -> list[Match]: + return [m for m in self.matches if not m.needs_a_human] + + @property + def ambiguous(self) -> list[Match]: + return [m for m in self.matches if m.kind is MatchKind.AMBIGUOUS] + + @property + def unmatched(self) -> list[Match]: + return [m for m in self.matches if m.kind is MatchKind.UNMATCHED] + + @property + def ready(self) -> bool: + """True when every point resolved to exactly one well. + + Deliberately strict. A partial run that ingests the wells it recognised + and quietly skips the rest produces a series that looks complete and is + not. + """ + return bool(self.matches) and not any(m.needs_a_human for m in self.matches) + + +def _normalize(value: str) -> str: + """Reduce a well identifier to its significant characters. + + Case, spacing and punctuation are dropped, so ``SO-0125``, ``so 0125`` and + ``SO0125`` compare equal -- one identifier written three ways. + + This is still exact matching, not similarity: every significant character + must agree, so ``SO-0126`` remains a different well. The distinction matters + because a fuzzy matcher here would eventually merge two real wells, and the + whole point of this module is that it never chooses between candidates. + """ + return "".join(c for c in (value or "") if c.isalnum()).upper() + + +def match_point(point: VendorPoint, candidates: Iterable[ThingCandidate]) -> Match: + """Decide one point against the wells it might be.""" + target = _normalize(point.name) + + by_name = [c for c in candidates if _normalize(c.name) == target] + by_external = [ + c + for c in candidates + if any(_normalize(x) == target for x in c.external_ids) and c not in by_name + ] + + # Name first: it is the identifier the Bureau uses, and an external id link + # is a record of an association someone made, which may be older. + hits = by_name or by_external + kind = MatchKind.NAME if by_name else MatchKind.EXTERNAL_ID + + if len(hits) == 1: + return Match(point=point, kind=kind, thing_id=hits[0].thing_id) + if len(hits) > 1: + return Match( + point=point, + kind=MatchKind.AMBIGUOUS, + candidates=tuple(c.thing_id for c in hits), + ) + return Match(point=point, kind=MatchKind.UNMATCHED) + + +def reconcile( + points: Iterable[VendorPoint], candidates: Iterable[ThingCandidate] +) -> ReconciliationReport: + """Match every vendor point, reporting rather than resolving.""" + candidate_list = list(candidates) + report = ReconciliationReport() + for point in points: + report.matches.append(match_point(point, candidate_list)) + return report + + +def format_report(report: ReconciliationReport) -> str: + """Human-readable summary. This is the deliverable of task 3.2.""" + lines = [ + f"Vendor points : {len(report.matches)}", + f" matched : {len(report.matched)}", + f" ambiguous : {len(report.ambiguous)}", + f" unmatched : {len(report.unmatched)}", + "", + ] + if report.ready: + lines.append("Every point resolved to exactly one well.") + return "\n".join(lines) + + if report.ambiguous: + lines.append("Ambiguous -- more than one well matches. Do not auto-merge:") + for match in report.ambiguous: + ids = ", ".join(str(c) for c in match.candidates) + lines.append(f" {match.point.name:<12} thing ids: {ids}") + lines.append("") + if report.unmatched: + lines.append("Unmatched -- no well found. Ingestion will not create one:") + for match in report.unmatched: + lines.append( + f" {match.point.name:<12} (vendor id {match.point.monitoring_point_id})" + ) + lines.append("") + lines.append("Resolve these before loading; a partial load looks complete.") + return "\n".join(lines) + + +# ============= EOF ============================================= diff --git a/automated_ingestion/tests/test_reconcile.py b/automated_ingestion/tests/test_reconcile.py new file mode 100644 index 000000000..9ea8b8b7e --- /dev/null +++ b/automated_ingestion/tests/test_reconcile.py @@ -0,0 +1,136 @@ +# =============================================================================== +# 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. +# =============================================================================== +""" +Reconciliation decisions. + +The rule that matters: never pick a winner among candidates. Ingestion does not +create wells and must not choose between two plausible ones. +""" + +from automated_ingestion.sources.san_acacia.reconcile import ( + MatchKind, + ThingCandidate, + VendorPoint, + format_report, + match_point, + reconcile, +) + +POINT = VendorPoint(monitoring_point_id=39, name="SO-0125") + + +def test_exact_name_match(): + match = match_point(POINT, [ThingCandidate(thing_id=7, name="SO-0125")]) + assert match.kind is MatchKind.NAME + assert match.thing_id == 7 + assert not match.needs_a_human + + +def test_name_match_ignores_case_spacing_and_punctuation(): + # One identifier written three ways. Still exact on significant characters. + for written in ("so 0125", "SO0125", " so-0125 "): + match = match_point(POINT, [ThingCandidate(thing_id=7, name=written)]) + assert match.thing_id == 7, written + + +def test_adjacent_identifier_is_not_a_match(): + # Normalization must not become fuzziness: SO-0126 is a different well. + match = match_point(POINT, [ThingCandidate(thing_id=7, name="SO-0126")]) + assert match.kind is MatchKind.UNMATCHED + + +def test_external_id_match_when_the_name_differs(): + match = match_point( + POINT, + [ThingCandidate(thing_id=9, name="Renamed Well", external_ids=("SO-0125",))], + ) + assert match.kind is MatchKind.EXTERNAL_ID + assert match.thing_id == 9 + + +def test_name_wins_over_external_id(): + # The name is the identifier the Bureau uses now; a link records an + # association someone made earlier, which may be stale. + match = match_point( + POINT, + [ + ThingCandidate(thing_id=7, name="SO-0125"), + ThingCandidate(thing_id=9, name="Other", external_ids=("SO-0125",)), + ], + ) + assert match.thing_id == 7 + + +def test_two_wells_with_the_same_name_are_ambiguous(): + # Duplicate rows exist in this database. Picking one is exactly the + # judgement that must not be automated. + match = match_point( + POINT, + [ + ThingCandidate(thing_id=7, name="SO-0125"), + ThingCandidate(thing_id=8, name="SO-0125"), + ], + ) + assert match.kind is MatchKind.AMBIGUOUS + assert match.thing_id is None + assert match.candidates == (7, 8) + assert match.needs_a_human + + +def test_no_candidate_is_unmatched_not_created(): + match = match_point(POINT, []) + assert match.kind is MatchKind.UNMATCHED + assert match.thing_id is None + + +class TestReport: + def _report(self): + points = [ + VendorPoint(39, "SO-0125"), + VendorPoint(40, "SO-0131"), + VendorPoint(41, "SO-0140"), + ] + candidates = [ + ThingCandidate(1, "SO-0125"), + ThingCandidate(2, "SO-0131"), + ThingCandidate(3, "SO-0131"), + ] + return reconcile(points, candidates) + + def test_counts_split_by_outcome(self): + report = self._report() + assert len(report.matched) == 1 + assert len(report.ambiguous) == 1 + assert len(report.unmatched) == 1 + + def test_not_ready_while_anything_needs_a_human(self): + # A partial load produces a series that looks complete and is not. + assert self._report().ready is False + + def test_ready_only_when_everything_resolves(self): + report = reconcile([VendorPoint(39, "SO-0125")], [ThingCandidate(1, "SO-0125")]) + assert report.ready is True + + def test_empty_input_is_not_ready(self): + # Nothing to reconcile is not the same as everything reconciled. + assert reconcile([], []).ready is False + + def test_report_names_the_points_needing_attention(self): + text = format_report(self._report()) + assert "SO-0131" in text and "SO-0140" in text + + +# ============= EOF ============================================= diff --git a/docs/automated-ingestion-pipeline-plan.md b/docs/automated-ingestion-pipeline-plan.md index e07edd73b..7bb9e1ced 100644 --- a/docs/automated-ingestion-pipeline-plan.md +++ b/docs/automated-ingestion-pipeline-plan.md @@ -4,7 +4,7 @@ ## TL;DR -Build the Bureau's first automated data ingestion pipeline, in the OcotilloAPI repo, so continuous depth-to-groundwater readings reach Ocotillo on a schedule instead of by hand. San Acacia Reach (33 Van Essen divers) is the pilot source; the structure it establishes is what every later source inherits. +Build the Bureau's first automated data ingestion pipeline, in the OcotilloAPI repo, so continuous depth-to-groundwater readings reach Ocotillo on a schedule instead of by hand. San Acacia Reach (38 Van Essen divers) is the pilot source; the structure it establishes is what every later source inherits. Stack: **Dagster+** code location → **dlt** extraction → **GCS** raw parquet → **`domain/`** mapping → direct **Postgres** load. Watermark and backfill mechanics are ported from Aqueduct, with two deliberate improvements a relational destination allows: the watermark is read from Postgres rather than a GCS sidecar, and an upsert replaces Aqueduct's delete-then-repost (removing its known window where data goes temporarily missing). @@ -30,11 +30,11 @@ Stack: **Dagster+** code location → **dlt** extraction → **GCS** raw parquet | 1.4 | DB connectivity + least-privilege role | Cloud SQL connector from serverless; scoped Postgres role | 1.2 | | **T2** | **Source extraction** | Van Essen API → GCS raw zone | T1 | | 2.1 | Confirm endpoint + finalize mapping | **Unblocked.** Diver-HUB swagger, JWT login, measure the window ceiling | — | -| 2.2 | dlt resource: locations | 33 wells, `replace`, one call, no pagination | 1.3 | +| 2.2 | dlt resource: locations | 38 wells, `replace`, one call, no pagination | 1.3 | | 2.3 | dlt resource: readings, incremental | Windowed per-point fetch, dlt cursor, `append`, token refresh, failure isolation | 2.1 | | **T3** | **Domain mapping + load** | Van Essen records → Ocotillo Postgres | T1 | | 3.1 | Domain layer | Pure functions: units, datum, timestamps, geometry, external keys | — | -| 3.2 | Bootstrap reference data | Reconcile 33 wells; seed parameter, sensor, deployments | 3.1 | +| 3.2 | Bootstrap reference data | Reconcile 38 wells; seed parameter, sensor, deployments | 3.1 | | 3.3 | Represent "public but provisional" | **Schema change.** `release_status` can't hold both axes | — | | 3.4 | Unique constraint + upsert loader | **Schema change.** `ON CONFLICT DO UPDATE`; makes backfill idempotent | 3.2, 3.3 | | 3.5 | Watermark from Postgres | `MAX(observation_datetime)` per series; no GCS sidecar | 3.4 | @@ -51,11 +51,11 @@ Stack: **Dagster+** code location → **dlt** extraction → **GCS** raw parquet **Goal:** continuous depth-to-groundwater data lands in Ocotillo automatically, on a schedule, with no one hand-carrying files — starting with San Acacia Reach. -The Hydrograph Corrector UI exists and works (BDMS-1137 done), but has no automatic supply of raw data. San Acacia Reach's 33 Van Essen divers historically flowed through the retired FROST/`st2` stack and now flow nowhere. This epic builds the supply. Correction, review, and publication workflows are **out of scope** and belong to their own epic. +The Hydrograph Corrector UI exists and works (BDMS-1137 done), but has no automatic supply of raw data. San Acacia Reach's 38 Van Essen divers historically flowed through the retired FROST/`st2` stack and now flow nowhere. This epic builds the supply. Correction, review, and publication workflows are **out of scope** and belong to their own epic. New top-level `automated_ingestion/` package in OcotilloAPI, deployed as its own Dagster+ code location in the existing `nmbgmr-data-services` org. dlt extracts the Van Essen API to a GCS raw zone; a `domain/` layer maps to the Ocotillo model; a loader writes to Ocotillo Postgres over a direct DB connection. Watermark and backfill mechanics come from Aqueduct. -San Acacia first: 33 wells, one DTW series each, and already mapped in `Aqueduct/docs/sources/san_acacia.md`. It authenticates with a short-lived JWT and must be read in bounded time windows — both cheap enough here to establish the pattern before a harder source needs it. What it establishes — source registry, per-source dlt pipeline, adapter, backfill job factory — every later source inherits. +San Acacia first: 38 wells, one DTW series each, and already mapped in `Aqueduct/docs/sources/san_acacia.md`. It authenticates with a short-lived JWT and must be read in bounded time windows — both cheap enough here to establish the pattern before a harder source needs it. What it establishes — source registry, per-source dlt pipeline, adapter, backfill job factory — every later source inherits. **Ownership: OcotilloAPI.** Not a third Aqueduct source writing into Ocotillo. The loader writes over a direct database connection, which wants the `db/` SQLAlchemy models and `domain/` rules in-process rather than a duplicated schema in another repo. Aqueduct stays the FROST/SensorThings pipeline; this is Ocotillo's own. The two share code by porting (see below), not by importing. @@ -89,7 +89,7 @@ San Acacia first: 33 wells, one DTW series each, and already mapped in `Aqueduct - Scheduled job runs end to end: Van Essen API → GCS parquet → domain mapping → Ocotillo Postgres. - Re-running over an already-loaded window: zero duplicates, zero errors. - Both backfill jobs exist, default `dry_run: true`, chunk by month, resume from last completed chunk. -- 33 wells resolve to `Thing` records — matched or created, no duplicates. +- 38 wells resolve to `Thing` records — matched, never created, no duplicates. - Readings are public, marked provisional, stored as DTW below ground surface in feet. - Series render in the Hydrograph Corrector. - Domain mapping unit-tested with no database, per `ADR4.md`. @@ -178,7 +178,7 @@ Dagster+ Serverless is outside the VPC, so Cloud SQL's private IP is unreachable - ✅ Role DDL in `automated_ingestion/sql/ingestion_role.sql`, kept out of Alembic: roles and grants are per-environment infrastructure, not schema, and migrations do not run as a superuser. - ⬜ Run the DDL per environment; set `DB_DRIVER`, `CLOUD_SQL_*` on the code location; materialize the asset from both a branch and prod deployment. -**The grant list is narrower than the draft assumed, and one part of it is non-obvious.** Writable: `transducer_observation`, `transducer_observation_block`, `deployment`, `sensor`, `parameter`. Read-only: `thing`, `thing_id_link`, `location`, and the three `lexicon_*` tables — `thing` and `location` deliberately *not* writable, because reconciling the 33 wells means matching rows that already exist. A well found missing is a decision for a human, not a row the pipeline invents. +**The grant list is narrower than the draft assumed, and one part of it is non-obvious.** Writable: `transducer_observation`, `transducer_observation_block`, `deployment`, `sensor`, `parameter`. Read-only: `thing`, `thing_id_link`, `location`, and the three `lexicon_*` tables — `thing` and `location` deliberately *not* writable, because reconciling the 38 wells means matching rows that already exist. A well found missing is a decision for a human, not a row the pipeline invents. `parameter` is versioned by sqlalchemy-continuum, so inserting one also writes to `parameter_version` and `transaction`. Without those two grants the write fails on a table the code never names — the kind of error that costs an afternoon. (`transducer_observation` itself is not versioned; only `aquifer_system`, `geologic_formation`, `location`, `observation`, `parameter`, `regulatory_limit`, and `thing` are.) Sequence `USAGE` is granted explicitly, and no default privileges are set: a table added later stays invisible until someone grants it deliberately. @@ -219,7 +219,7 @@ Also still open: the window ceiling (three months works, the limit is unmeasured - ✅ Asset `raw_san_acacia_locations` emits the point count, project id, and a sample of names. Tested against a stub, no network. - ✅ `replace` rather than `append`: this is a snapshot of what the vendor currently lists, and a point disappearing is information rather than something to accumulate. -The payload is `{id, name}` only, so this cannot be a source of geometry or construction detail — it enumerates the points a reading fetch walks. **38 points, not the 33 the plan assumes**, still unexplained. +The payload is `{id, name}` only, so this cannot be a source of geometry or construction detail — it enumerates the points a reading fetch walks. **38 points.** Earlier drafts said 33; that figure came from Aqueduct's stale mapping doc, not from a Bureau record — see 3.2. ### 2.3 — dlt resource: readings → GCS, incremental @@ -237,9 +237,9 @@ The payload is `{id, name}` only, so this cannot be a source of geometry or cons # TASK 3 — Domain mapping and load into Ocotillo -Where this stops resembling Aqueduct: the destination is a relational database with constraints and transactions, and mapping rules belong in `domain/` per `ADR4.md`. Three risks — matching 33 wells without duplicating them, representing "public but provisional" when the schema can't, and making the write idempotent so backfill is safe. +Where this stops resembling Aqueduct: the destination is a relational database with constraints and transactions, and mapping rules belong in `domain/` per `ADR4.md`. Three risks — matching 38 wells without duplicating them, representing "public but provisional" when the schema can't, and making the write idempotent so backfill is safe. -**Done when:** mapping rules are pure functions tested without a database; 33 wells resolve with no duplicates; data is public and separately marked provisional; `transducer_observation` has a unique constraint and the loader upserts against it; loading the same window twice leaves the row count unchanged; the watermark comes from Postgres. +**Done when:** mapping rules are pure functions tested without a database; 38 wells resolve with no duplicates; data is public and separately marked provisional; `transducer_observation` has a unique constraint and the loader upserts against it; loading the same window twice leaves the row count unchanged; the watermark comes from Postgres. ### 3.1 — Domain layer: Van Essen record → Ocotillo model @@ -266,14 +266,19 @@ The module docstring lists every value the mapping **invents** rather than reads ### 3.2 — Bootstrap reference data: reconcile wells, seed parameter, sensor, deployments -Some of the 33 may already exist in Ocotillo under Bureau point IDs. Duplicates are the main risk — the `group_type` collision elsewhere in this database is the reminder that "looks new" isn't proof. +Reconciliation report built — `sources/san_acacia/reconcile.py` and `scripts/reconcile_san_acacia.py`, 12 tests. The seeding half is not built. -- Reconciliation report **first**: per well, whether a matching `Thing` exists — on name, on `monitoringPoints[].name` (e.g. `SO-0125`), and on coordinate proximity. Ambiguous matches escalate to a human, never auto-merge. -- Data migration (existing `data_migrations/` runner, already supports dry-run) creates missing `Location`/`Thing`, links existing ones. Idempotent, dry-run-clean before running for real. -- Lexicon terms, a DTW `Parameter`, and a `VanEssenDiver` `Sensor` created if absent. -- One `Deployment` per well (thing → sensor), `recording_interval` ~5 min where known. -- Van Essen `uid` (e.g. `sanacaciareach-40`) persisted as external identifier. -- `DataProvenance` recorded for Van Essen-sourced well attributes: depth, coordinates, installation date. +**The "33 wells" figure was wrong, and was never a blocker.** It came from Aqueduct's `docs/sources/san_acacia.md` — the same document that also supplied the doubled `/api/api/` path, the claim the source is unauthenticated, and the `gs`/`vrd` payload shape, all disproved against the live API. 38 is what the API returns. Whether all 38 are in scope is a question the per-well report answers concretely. + +**Coordinate proximity is not available.** This section called for matching on name, external id, *and* coordinate proximity. `MonitoringPoint` is `{id, name}` — no coordinates. That removes the only fuzzy signal and leaves two exact ones, which is a better position: every match is defensible rather than probabilistic. + +- ✅ Matching on name and on `thing_id_link.alternate_id`, normalized for case, spacing and punctuation so `SO-0125`, `so 0125` and `SO0125` compare equal. Still exact on significant characters — `SO-0126` stays a different well. +- ✅ Name beats external id when both hit. The name is the identifier the Bureau uses now; a link records an association someone made earlier. +- ✅ **Never picks a winner.** More than one candidate is `ambiguous` and escalates; none is `unmatched` and escalates. Ingestion does not create wells, and choosing between two plausible ones is the judgement that must not be automated. +- ✅ `report.ready` is false unless *every* point resolved, and false for empty input. A partial load produces a series that looks complete and is not. +- ✅ The script exits non-zero when anything needs a human, so it can gate a later step without relying on someone reading the output. +- ⬜ Run it against staging and production and act on the result. +- ⬜ The seeding half: data migration creating missing `Location`/`Thing`, lexicon terms, DTW `Parameter`, `VanEssenDiver` `Sensor`, one `Deployment` per well, the vendor `uid` as external identifier, and `DataProvenance` for Van Essen-sourced attributes. ### 3.3 — Represent "public but provisional" diff --git a/docs/sources/san_acacia.md b/docs/sources/san_acacia.md index b184d5636..787670ba2 100644 --- a/docs/sources/san_acacia.md +++ b/docs/sources/san_acacia.md @@ -1,10 +1,24 @@ # Source: San Acacia Reach (Van Essen divers, Diver-HUB) The pilot source for automated ingestion. Project **4317 `SanAcaciaReach`**, -containing **38 monitoring points** named `SO-####` — the plan and the Aqueduct -mapping both say 33, so five are unaccounted for and must be identified before -3.2 reconciles anything. Ingestion never creates wells, so an unexpected point -is a decision, not a row. Historically flowed through the retired +containing **38 monitoring points** named `SO-####`. + +**On the "33 wells" figure.** Earlier drafts of the plan said 33 and treated 38 +as a discrepancy to resolve. It is not one. The number came from Aqueduct's +`docs/sources/san_acacia.md`, in a sentence describing an endpoint that no +longer exists: + +> Pagination: none — `/locations/{projectName}` returns all 33 wells in one response. + +That document is also where the doubled `/api/api/` path, the claim that the +source is unauthenticated, and the `gs`/`vrd` array payload came from — all four +disproved against the live API. The count has no more standing than the rest of +it: a FROST-era snapshot, not a Bureau record of how many wells the reach has. + +**38 is the live count.** Whether all 38 are in scope — some may be +decommissioned, or belong to a neighbouring project — is a question about the +well inventory, and the reconciliation report answers it concretely, per well, +rather than by arguing about a total. Historically flowed through the retired FROST/`st2` stack; now flows nowhere. This document supersedes the mapping in `Aqueduct/docs/sources/san_acacia.md`, @@ -203,7 +217,7 @@ Settled, not to be relitigated per source: |---|---|---| | 1 | ~~Which `reference` value is ground surface?~~ | **Answered: 3.** Corroboration via `ManualMeasurements` still outstanding | | 2 | ~~What is the window ceiling?~~ | **`WaterLevels` took 730 d / 18111 rows. The 500 is a `DiverData` problem** | -| 3 | ~~Which project id, how many points?~~ | **Answered: 4317, 38 points (not 33)** | +| 3 | ~~Which project id, how many points?~~ | **Answered: 4317, 38 points. The 33 was a stale figure, not a discrepancy** | | 4 | Do `approved=true` and `approved=false` partition the series, or overlap? | Fetch both for one window and compare timestamps | | 5 | Is `dateAndTime` UTC in the response, and is it marked as such? | Inspect a live payload | | 6 | ~~Is `level` in feet?~~ | **No — centimetres.** Convert with `convert_cm_to_ft` | From 50963f4e42f275dd9aafd4b041a6cc8701aafcec Mon Sep 17 00:00:00 2001 From: jakeross Date: Wed, 19 Aug 2026 08:35:26 -0700 Subject: [PATCH 2/4] fix(ingestion): import ThingIdLink from where it actually lives I inferred the module path from the table name: thing_id_link became db.thing_id_link, and the class ThingIDLink. Both wrong -- the class is ThingIdLink and it is defined in db.thing. The query is now exercised against a real database rather than only imported, so the SQL is verified and not just the syntax. Every import in both new modules is checked to resolve, which is the class of mistake this was. Co-Authored-By: Claude Opus 5 --- automated_ingestion/scripts/reconcile_san_acacia.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/automated_ingestion/scripts/reconcile_san_acacia.py b/automated_ingestion/scripts/reconcile_san_acacia.py index 2e730ef2d..973cafbd6 100644 --- a/automated_ingestion/scripts/reconcile_san_acacia.py +++ b/automated_ingestion/scripts/reconcile_san_acacia.py @@ -62,15 +62,14 @@ def _candidates(prefix: str) -> list[ThingCandidate]: from sqlalchemy import select from db.engine import session_ctx - from db.thing import Thing - from db.thing_id_link import ThingIDLink + from db.thing import Thing, ThingIdLink with session_ctx() as session: things = session.execute( select(Thing.id, Thing.name).where(Thing.name.ilike(f"{prefix}%")) ).all() links = session.execute( - select(ThingIDLink.thing_id, ThingIDLink.alternate_id) + select(ThingIdLink.thing_id, ThingIdLink.alternate_id) ).all() by_thing: dict[int, list[str]] = {} From d102cb7b4b760bdbc4beaa50ee75101995a88931 Mon Sep 17 00:00:00 2001 From: jakeross Date: Wed, 19 Aug 2026 08:45:41 -0700 Subject: [PATCH 3/4] fix(ingestion): do not match on external ids by default Reconciling against staging answered 3.2: all 38 Diver-HUB points match Ocotillo wells by name, nothing ambiguous, nothing unmatched. The wells already exist, so the seeding half creates none. The same data showed external-id matching is unsafe here. thing_id_link holds 11,148 links from nine organization/relation pairs that disagree with each other. SO-0131 carries NMBGMR "BRN-E04B (shallow)" plus an unattributed "BRN-E04A", while SO-0132 carries NMBGMR "BRN-E04A (deep)" plus an unattributed "BRN-E04B" -- the two sources swap which physical well is A and which is B. Matching BRN-E04A against that returns one confident hit on SO-0131, contradicting NMBGMR, because the parenthetical suffix stops the collision registering as ambiguous. That is worse than the ambiguity the module was built to escalate: a wrong answer delivered with no sign of trouble. So the fallback is opt-in, and a test pins those exact rows. It costs nothing today, since every point matches by name. Co-Authored-By: Claude Opus 5 --- .../sources/san_acacia/reconcile.py | 45 +++++++++++++++---- automated_ingestion/tests/test_reconcile.py | 34 +++++++++++++- docs/automated-ingestion-pipeline-plan.md | 7 ++- 3 files changed, 75 insertions(+), 11 deletions(-) diff --git a/automated_ingestion/sources/san_acacia/reconcile.py b/automated_ingestion/sources/san_acacia/reconcile.py index e7726dfd1..9a555c866 100644 --- a/automated_ingestion/sources/san_acacia/reconcile.py +++ b/automated_ingestion/sources/san_acacia/reconcile.py @@ -125,16 +125,39 @@ def _normalize(value: str) -> str: return "".join(c for c in (value or "") if c.isalnum()).upper() -def match_point(point: VendorPoint, candidates: Iterable[ThingCandidate]) -> Match: - """Decide one point against the wells it might be.""" +def match_point( + point: VendorPoint, + candidates: Iterable[ThingCandidate], + use_external_ids: bool = False, +) -> Match: + """Decide one point against the wells it might be. + + ``use_external_ids`` is off by default, for a specific reason. + ``thing_id_link`` holds identifiers from several organizations that disagree + with each other. In staging, ``SO-0131`` carries NMBGMR ``BRN-E04B + (shallow)`` plus an unattributed ``BRN-E04A``, while ``SO-0132`` carries + NMBGMR ``BRN-E04A (deep)`` plus an unattributed ``BRN-E04B`` -- the two + sources swap which physical well is A and which is B. + + Matching ``BRN-E04A`` against that returns a single confident hit on + SO-0131, contradicting NMBGMR, because the parenthetical suffix stops the + collision registering as ambiguous. A wrong answer delivered confidently is + worse than no answer. + + It costs nothing today: all 38 Diver-HUB points match Ocotillo wells by name. + """ target = _normalize(point.name) by_name = [c for c in candidates if _normalize(c.name) == target] - by_external = [ - c - for c in candidates - if any(_normalize(x) == target for x in c.external_ids) and c not in by_name - ] + by_external = ( + [ + c + for c in candidates + if any(_normalize(x) == target for x in c.external_ids) and c not in by_name + ] + if use_external_ids + else [] + ) # Name first: it is the identifier the Bureau uses, and an external id link # is a record of an association someone made, which may be older. @@ -153,13 +176,17 @@ def match_point(point: VendorPoint, candidates: Iterable[ThingCandidate]) -> Mat def reconcile( - points: Iterable[VendorPoint], candidates: Iterable[ThingCandidate] + points: Iterable[VendorPoint], + candidates: Iterable[ThingCandidate], + use_external_ids: bool = False, ) -> ReconciliationReport: """Match every vendor point, reporting rather than resolving.""" candidate_list = list(candidates) report = ReconciliationReport() for point in points: - report.matches.append(match_point(point, candidate_list)) + report.matches.append( + match_point(point, candidate_list, use_external_ids=use_external_ids) + ) return report diff --git a/automated_ingestion/tests/test_reconcile.py b/automated_ingestion/tests/test_reconcile.py index 9ea8b8b7e..3317d0bc8 100644 --- a/automated_ingestion/tests/test_reconcile.py +++ b/automated_ingestion/tests/test_reconcile.py @@ -52,16 +52,47 @@ def test_adjacent_identifier_is_not_a_match(): assert match.kind is MatchKind.UNMATCHED -def test_external_id_match_when_the_name_differs(): +def test_external_ids_are_ignored_by_default(): match = match_point( POINT, [ThingCandidate(thing_id=9, name="Renamed Well", external_ids=("SO-0125",))], ) + assert match.kind is MatchKind.UNMATCHED + + +def test_external_id_match_when_explicitly_enabled(): + match = match_point( + POINT, + [ThingCandidate(thing_id=9, name="Renamed Well", external_ids=("SO-0125",))], + use_external_ids=True, + ) assert match.kind is MatchKind.EXTERNAL_ID assert match.thing_id == 9 +def test_external_ids_can_produce_a_confident_wrong_answer(): + """Why external id matching is off by default. Real rows from staging. + + SO-0131 and SO-0132 swap which physical well is A and which is B between + NMBGMR and the unattributed source. Matching BRN-E04A returns SO-0131 with + no hint of trouble, while NMBGMR asserts SO-0132 is BRN-E04A -- the + parenthetical suffix stops the collision registering as ambiguous. + """ + candidates = [ + ThingCandidate(2369, "SO-0131", ("BRN-E04B (shallow)", "BRN-E04A")), + ThingCandidate(2373, "SO-0132", ("BRN-E04A (deep)", "BRN-E04B")), + ] + enabled = match_point( + VendorPoint(999, "BRN-E04A"), candidates, use_external_ids=True + ) + assert enabled.thing_id == 2369 # contradicts NMBGMR, and looks certain + + default = match_point(VendorPoint(999, "BRN-E04A"), candidates) + assert default.kind is MatchKind.UNMATCHED # escalates instead + + def test_name_wins_over_external_id(): + # Only relevant when external ids are enabled. # The name is the identifier the Bureau uses now; a link records an # association someone made earlier, which may be stale. match = match_point( @@ -70,6 +101,7 @@ def test_name_wins_over_external_id(): ThingCandidate(thing_id=7, name="SO-0125"), ThingCandidate(thing_id=9, name="Other", external_ids=("SO-0125",)), ], + use_external_ids=True, ) assert match.thing_id == 7 diff --git a/docs/automated-ingestion-pipeline-plan.md b/docs/automated-ingestion-pipeline-plan.md index 7bb9e1ced..abad6bd2e 100644 --- a/docs/automated-ingestion-pipeline-plan.md +++ b/docs/automated-ingestion-pipeline-plan.md @@ -277,7 +277,12 @@ Reconciliation report built — `sources/san_acacia/reconcile.py` and `scripts/r - ✅ **Never picks a winner.** More than one candidate is `ambiguous` and escalates; none is `unmatched` and escalates. Ingestion does not create wells, and choosing between two plausible ones is the judgement that must not be automated. - ✅ `report.ready` is false unless *every* point resolved, and false for empty input. A partial load produces a series that looks complete and is not. - ✅ The script exits non-zero when anything needs a human, so it can gate a later step without relying on someone reading the output. -- ⬜ Run it against staging and production and act on the result. +- ✅ **Run against staging: all 38 points match by name. Nothing ambiguous, nothing unmatched, `ready = True`.** The wells already exist — SO-0125 is thing 2343, SO-0131 is 2369, and so on through 277 `SO-` wells in that database. So the seeding half creates no wells; it only needs the parameter, sensor, deployments and external identifiers. +- ⬜ Run against production and confirm the same. + +**External-id matching is off by default, on evidence.** `thing_id_link` in staging holds 11,148 links from nine organization/relation pairs — NMBGMR (8,603), PLSS (7,052), an unattributed "Unknown" (4,825), NMOSE, USGS, NMED, TWDB — and they disagree with each other. `SO-0131` carries NMBGMR `BRN-E04B (shallow)` plus an unattributed `BRN-E04A`, while `SO-0132` carries NMBGMR `BRN-E04A (deep)` plus an unattributed `BRN-E04B`: the two sources swap which physical well is A and which is B. (`SO-0262`/`SO-0263` disagree more sharply still — NMBGMR calls them NRCS 3A/3B, the other source NRCS 2.) + +Matching `BRN-E04A` against that returns a single confident hit on `SO-0131`, contradicting NMBGMR, because the parenthetical suffix stops the collision registering as ambiguous. A wrong answer delivered confidently is worse than no answer, so the fallback is opt-in and a test pins the real rows. - ⬜ The seeding half: data migration creating missing `Location`/`Thing`, lexicon terms, DTW `Parameter`, `VanEssenDiver` `Sensor`, one `Deployment` per well, the vendor `uid` as external identifier, and `DataProvenance` for Van Essen-sourced attributes. ### 3.3 — Represent "public but provisional" From 6d8cf475128ee69f8ad52ee60ec8bd25717ef7d2 Mon Sep 17 00:00:00 2001 From: jakeross Date: Wed, 19 Aug 2026 08:55:25 -0700 Subject: [PATCH 4/4] docs(ingestion): confirm reconciliation against production All 38 Diver-HUB points match Ocotillo wells by name in production too, with the same thing ids as staging -- SO-0125 is 2343 in both -- so the two agree by construction and 3.2's reconciliation is settled for both environments. The seeding half creates no wells. Also records that the identifier contradictions are production data rather than a staging artifact. SO-0131/SO-0132 and SO-0262/SO-0263 are paired shallow/deep piezometers whose A/B designations disagree between identifier sources, which matters to anyone reasoning about those wells through the BRN- or NRCS names even though ingestion itself is unaffected. Co-Authored-By: Claude Opus 5 --- docs/automated-ingestion-pipeline-plan.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/docs/automated-ingestion-pipeline-plan.md b/docs/automated-ingestion-pipeline-plan.md index abad6bd2e..24201288d 100644 --- a/docs/automated-ingestion-pipeline-plan.md +++ b/docs/automated-ingestion-pipeline-plan.md @@ -278,11 +278,13 @@ Reconciliation report built — `sources/san_acacia/reconcile.py` and `scripts/r - ✅ `report.ready` is false unless *every* point resolved, and false for empty input. A partial load produces a series that looks complete and is not. - ✅ The script exits non-zero when anything needs a human, so it can gate a later step without relying on someone reading the output. - ✅ **Run against staging: all 38 points match by name. Nothing ambiguous, nothing unmatched, `ready = True`.** The wells already exist — SO-0125 is thing 2343, SO-0131 is 2369, and so on through 277 `SO-` wells in that database. So the seeding half creates no wells; it only needs the parameter, sensor, deployments and external identifiers. -- ⬜ Run against production and confirm the same. +- ✅ **Production confirms it**: same 38 matches, same thing ids (SO-0125 is 2343 in both), `ready = True`. Staging is a clone of production for these tables, so the two agree by construction. **External-id matching is off by default, on evidence.** `thing_id_link` in staging holds 11,148 links from nine organization/relation pairs — NMBGMR (8,603), PLSS (7,052), an unattributed "Unknown" (4,825), NMOSE, USGS, NMED, TWDB — and they disagree with each other. `SO-0131` carries NMBGMR `BRN-E04B (shallow)` plus an unattributed `BRN-E04A`, while `SO-0132` carries NMBGMR `BRN-E04A (deep)` plus an unattributed `BRN-E04B`: the two sources swap which physical well is A and which is B. (`SO-0262`/`SO-0263` disagree more sharply still — NMBGMR calls them NRCS 3A/3B, the other source NRCS 2.) Matching `BRN-E04A` against that returns a single confident hit on `SO-0131`, contradicting NMBGMR, because the parenthetical suffix stops the collision registering as ambiguous. A wrong answer delivered confidently is worse than no answer, so the fallback is opt-in and a test pins the real rows. + +**This is production data, not a staging artifact.** The same contradictions are in both. They are worth someone's attention independently of this pipeline: `SO-0131`/`SO-0132` and `SO-0262`/`SO-0263` are paired shallow/deep piezometers whose A/B designations disagree between identifier sources, and a swap there means a shallow series attributed to a deep well. Ingestion is unaffected — the vendor names points `SO-####` and Ocotillo agrees on those — but anyone reasoning about these wells through the `BRN-`/`NRCS` names is working from two incompatible answers. - ⬜ The seeding half: data migration creating missing `Location`/`Thing`, lexicon terms, DTW `Parameter`, `VanEssenDiver` `Sensor`, one `Deployment` per well, the vendor `uid` as external identifier, and `DataProvenance` for Van Essen-sourced attributes. ### 3.3 — Represent "public but provisional"