diff --git a/automated_ingestion/defs/assets/__init__.py b/automated_ingestion/defs/assets/__init__.py index 3671abfa5..4e7a0494c 100644 --- a/automated_ingestion/defs/assets/__init__.py +++ b/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, ] diff --git a/automated_ingestion/scripts/compare_datum.py b/automated_ingestion/scripts/compare_datum.py new file mode 100644 index 000000000..b513ffdf5 --- /dev/null +++ b/automated_ingestion/scripts/compare_datum.py @@ -0,0 +1,217 @@ +# =============================================================================== +# 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. +# =============================================================================== +""" +Check that ingested readings agree with the observations Ocotillo already holds. + +Fourteen San Acacia wells carry AMPAPI transducer data through August 2022, +loaded under a datum nobody has verified. This pipeline reads Diver-HUB with +``reference=3`` and converts centimetres to feet. If those disagree, the same +series ends up holding two datums -- and the numbers look plausible either way, +which is the failure this source is most prone to. + +Magnitude alone cannot settle it: ``reference=1`` (top of casing) differs from +``reference=3`` (ground surface) by a fixed 45.456 cm -- about 1.49 ft -- which +is well inside the natural range of these wells. Only values at the *same +instant* separate them, so this compares timestamp by timestamp. + +It fetches all four references rather than just the one in use, so the output +also independently confirms which reference Ocotillo's existing data was loaded +against. + + export DIVERHUB_USERNAME=... DIVERHUB_PASSWORD=... + uv run --group ingestion python -m \\ + automated_ingestion.scripts.compare_datum --well SO-0125 + +Read-only on both sides. +""" + +import argparse +import statistics +import sys +from datetime import timedelta + +REFERENCES = (0, 1, 2, 3) + + +def _existing(cursor, well: str, limit: int): + cursor.execute( + """ + SELECT o.observation_datetime, o.value + FROM transducer_observation o + JOIN deployment d ON d.id = o.deployment_id + JOIN thing t ON t.id = d.thing_id + WHERE t.name = %s + ORDER BY o.observation_datetime DESC + LIMIT %s + """, + (well, limit), + ) + return cursor.fetchall() + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--well", default="SO-0125", help="Ocotillo PointID") + parser.add_argument("--point-id", type=int, help="Diver-HUB monitoring point id") + parser.add_argument( + "--instance", default="waterdatainitiative-271000:us-west4:dataservices" + ) + parser.add_argument("--database", default="ocotillo-staging") + parser.add_argument("--samples", type=int, default=200) + parser.add_argument( + "--tolerance-minutes", + type=int, + default=30, + help=( + "How far apart two readings may be and still count as the same " + "instant. Exact equality is too strict: the existing rows are on the " + "hour and the vendor logs at 15-minute offsets." + ), + ) + args = parser.parse_args() + + import requests + from google.cloud.sql.connector import Connector + + from automated_ingestion.sources.san_acacia.client import DiverHubClient + from automated_ingestion.sources.san_acacia.dlt_pipeline import PROJECT_ID + from domain.units import convert_cm_to_ft + from domain.van_essen import parse_reading_timestamp + + client = DiverHubClient(requests.Session()) + + point_id = args.point_id + if point_id is None: + matches = [ + p for p in client.monitoring_points(PROJECT_ID) if p["name"] == args.well + ] + if not matches: + print(f"{args.well} is not a Diver-HUB monitoring point.", file=sys.stderr) + return 2 + point_id = matches[0]["id"] + + connector = Connector() + conn = connector.connect( + args.instance, + "pg8000", + user=_account(), + db=args.database, + enable_iam_auth=True, + ) + try: + rows = _existing(conn.cursor(), args.well, args.samples) + finally: + conn.close() + connector.close() + + if not rows: + print(f"No existing observations for {args.well}.", file=sys.stderr) + return 1 + + existing = {stamp.replace(tzinfo=None): value for stamp, value in rows} + start = min(existing) - timedelta(days=1) + end = max(existing) + timedelta(days=1) + print(f"{args.well} (Diver-HUB point {point_id})") + print( + f" {len(existing)} existing observations, {min(existing)} -> {max(existing)}" + ) + print( + f" Ocotillo values: {min(existing.values()):.2f} .. {max(existing.values()):.2f} ft\n" + ) + + tolerance = timedelta(minutes=args.tolerance_minutes) + print( + f" {'reference':<12}{'vendor rows':>12}{'matched':>9}" + f"{'mean diff ft':>15}{'max diff ft':>14}" + ) + best = None + for reference in REFERENCES: + vendor = {} + for row in client.water_levels( + point_id, + int(start.timestamp()), + int(end.timestamp()), + reference=reference, + ): + if row.get("level") is None: + continue + stamp = parse_reading_timestamp(row["dateAndTime"]).replace(tzinfo=None) + vendor[stamp] = convert_cm_to_ft(row["level"]) + + # Nearest within tolerance rather than exact equality. A reading logged + # at :45 against one recorded on the hour is the same measurement to + # anyone comparing datums; insisting on identical timestamps finds + # nothing and says nothing. + stamps = sorted(vendor) + diffs = [] + for stamp, value in existing.items(): + near = min(stamps, key=lambda s: abs(s - stamp)) if stamps else None + if near is not None and abs(near - stamp) <= tolerance: + diffs.append(abs(value - vendor[near])) + + if not diffs: + print(f" reference={reference:<4}{len(vendor):>10}{'none':>11}") + continue + + mean, worst = statistics.mean(diffs), max(diffs) + print( + f" reference={reference:<4}{len(vendor):>10}{len(diffs):>9}" + f"{mean:>15.3f}{worst:>14.3f}" + ) + if best is None or mean < best[1]: + best = (reference, mean) + + if best is None: + print("\n Nothing to compare.") + print( + " If vendor rows is 0, Diver-HUB does not retain this window for " + "this point -- try a well whose data runs later, or widen --tolerance-minutes." + ) + return 1 + + reference, mean = best + print(f"\n Closest: reference={reference}, mean difference {mean:.3f} ft") + if mean < 0.05: + verdict = ( + f"Ocotillo's existing data matches reference={reference}." + if reference == 3 + else f"Ocotillo's existing data was loaded on reference={reference}, NOT 3." + ) + else: + verdict = ( + "No reference matches closely. The existing data may use a different " + "unit, datum or correction than any raw Diver-HUB series." + ) + print(f" {verdict}") + return 0 + + +def _account() -> str: + import subprocess + + return subprocess.run( + ["gcloud", "config", "get-value", "account"], + capture_output=True, + text=True, + timeout=30, + ).stdout.strip() + + +if __name__ == "__main__": + raise SystemExit(main()) + + +# ============= EOF ============================================= diff --git a/automated_ingestion/scripts/diverhub_retention.py b/automated_ingestion/scripts/diverhub_retention.py new file mode 100644 index 000000000..0f23541de --- /dev/null +++ b/automated_ingestion/scripts/diverhub_retention.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. +# =============================================================================== +""" +Find how far back Diver-HUB actually serves each monitoring point. + +This matters for two reasons. + +``INITIAL_START`` is 2015-01-01, a floor chosen before anyone knew what the +vendor retains. A first run for a well with no history walks from there in +windows, and every window before the vendor's earliest reading is a request that +returns nothing -- against an endpoint that answers 500 when pushed. + +And the fourteen wells that already hold AMPAPI data stop in August 2022, while +the vendor appears to start much later. If so the two datasets never overlap, +which is why the datum comparison found nothing to compare: there is a gap +between them, not a seam. + +Binary search on presence, roughly ten requests per point rather than a walk. + + export DIVERHUB_USERNAME=... DIVERHUB_PASSWORD=... + uv run --group ingestion python -m \\ + automated_ingestion.scripts.diverhub_retention --limit 6 +""" + +import argparse +from datetime import datetime, timedelta, timezone + +PROBE_WINDOW = timedelta(days=30) + + +def _has_data(client, point_id: int, when: datetime, reference: int) -> bool: + """Is there any reading in the month starting at ``when``?""" + rows = client.water_levels( + point_id, + int(when.timestamp()), + int((when + PROBE_WINDOW).timestamp()), + reference=reference, + span=int(PROBE_WINDOW.total_seconds()), + ) + return any(True for _ in rows) + + +def earliest_reading( + client, point_id: int, reference: int, floor: datetime +) -> datetime | None: + """Approximate the first month that holds data, by bisection.""" + now = datetime.now(tz=timezone.utc) + if not _has_data(client, point_id, now - PROBE_WINDOW, reference): + # Nothing recent; the point may be retired. Fall back to a wide check. + if not _has_data(client, point_id, floor, reference): + pass # keep searching regardless -- absence now proves nothing + + low, high = floor, now + if _has_data(client, point_id, low, reference): + return low + + # Invariant: no data at `low`, data somewhere at or before `high`. + for _ in range(12): + if (high - low) <= PROBE_WINDOW: + break + middle = low + (high - low) / 2 + if _has_data(client, point_id, middle, reference): + high = middle + else: + low = middle + return high if _has_data(client, point_id, high - PROBE_WINDOW, reference) else high + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--limit", type=int, default=6, help="How many points to probe") + parser.add_argument("--floor", default="2015-01-01T00:00:00+00:00") + args = parser.parse_args() + + import requests + + from automated_ingestion.sources.san_acacia.client import ( + GROUND_SURFACE_REFERENCE, + DiverHubClient, + ) + from automated_ingestion.sources.san_acacia.dlt_pipeline import PROJECT_ID + from domain.van_essen import parse_reading_timestamp + + client = DiverHubClient(requests.Session()) + floor = parse_reading_timestamp(args.floor) + points = client.monitoring_points(PROJECT_ID)[: args.limit] + + print(f"Probing {len(points)} of {PROJECT_ID}'s monitoring points") + print(f" {'point':<12}{'earliest data (approx)':>26}") + for point in points: + found = earliest_reading(client, point["id"], GROUND_SURFACE_REFERENCE, floor) + shown = found.date().isoformat() if found else "none found" + print(f" {point['name']:<12}{shown:>26}") + + print( + "\nIf these cluster well after August 2022, the vendor and the existing\n" + "AMPAPI records do not overlap, and INITIAL_START can be raised to the\n" + "earliest date actually served -- saving a decade of empty requests on\n" + "every first run." + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) + + +# ============= EOF ============================================= diff --git a/automated_ingestion/sources/san_acacia/dlt_pipeline.py b/automated_ingestion/sources/san_acacia/dlt_pipeline.py index 465c8729f..5caa4c122 100644 --- a/automated_ingestion/sources/san_acacia/dlt_pipeline.py +++ b/automated_ingestion/sources/san_acacia/dlt_pipeline.py @@ -70,13 +70,29 @@ reliably. """ -INITIAL_START = "2015-01-01T00:00:00+00:00" +INITIAL_START = "2024-01-01T00:00:00+00:00" """Floor for a point that has never been ingested. -A floor, never a backfill lever: moving it forward does not delete anything -already landed, and moving it backward does not fetch history for a point whose -cursor has advanced past it. Use a backfill job for that -(``BACKFILL_STRATEGY.md`` section 2). +Diver-HUB serves nothing before late 2024. Probing six points put their earliest +reading at 2024-10-08 and 2024-11-10, which matches the deployments on these +wells being installed 2024-11-25 -- the vendor project was populated then. + +The floor sits at 2024-01-01 rather than at the earliest observed reading, +because only six of the thirty-eight points were probed and a well with slightly +earlier data should not be silently truncated. Nine months of margin costs one +extra empty window; guessing too late loses real readings. + +It was 2015-01-01, chosen before anyone knew what the vendor retains. At a +365-day span that made a first run walk about twelve windows per well, ten of +them guaranteed empty, against an endpoint that answers 500 when pushed. + +Still a floor, never a backfill lever: lowering it will not re-fetch history for +a series whose watermark has advanced past it (`shared/watermark.py`), and there +is no history before 2024 to fetch. + +**The record has a gap.** The fourteen wells carrying AMPAPI data stop in August +2022 and the vendor starts in late 2024, so roughly twenty-seven months are +missing and cannot be recovered from this source. """ SOURCE = register( diff --git a/automated_ingestion/sources/san_acacia/ingest.py b/automated_ingestion/sources/san_acacia/ingest.py index ff318d9fc..52228c4da 100644 --- a/automated_ingestion/sources/san_acacia/ingest.py +++ b/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: diff --git a/automated_ingestion/sources/san_acacia/resolve.py b/automated_ingestion/sources/san_acacia/resolve.py new file mode 100644 index 000000000..68638976d --- /dev/null +++ b/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 ============================================= diff --git a/automated_ingestion/tests/test_resolve.py b/automated_ingestion/tests/test_resolve.py new file mode 100644 index 000000000..f5330ba09 --- /dev/null +++ b/automated_ingestion/tests/test_resolve.py @@ -0,0 +1,93 @@ +# =============================================================================== +# 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 the deployment a water level belongs to. + +A well carries several open deployments because a deployment is equipment, not a +measured property. Picking the wrong one attributes a water level to a cable. +""" + +from datetime import date + +from automated_ingestion.sources.san_acacia.resolve import ( + DeploymentCandidate, + ResolutionKind, + resolve_deployment, +) + +# The real equipment on SO-0140 in staging. +DIVERLINK = DeploymentCandidate(436, "DiverLink") +TRANSDUCER = DeploymentCandidate(437, "Pressure Transducer") +CABLE = DeploymentCandidate(438, "Diver Cable") + + +def test_the_transducer_is_chosen_from_a_full_nest(): + resolution = resolve_deployment([DIVERLINK, TRANSDUCER, CABLE]) + assert resolution.kind is ResolutionKind.RESOLVED + assert resolution.deployment_id == 437 + + +def test_a_barometer_is_not_a_water_level(): + # Barometers are deployed on these wells too, and measure air pressure. + resolution = resolve_deployment([DeploymentCandidate(500, "Barometer"), TRANSDUCER]) + assert resolution.deployment_id == 437 + + +def test_two_open_transducers_are_ambiguous(): + # Two of the 38 wells are in this state. Taking the lower id would be a + # guess about equipment, made silently. + resolution = resolve_deployment( + [TRANSDUCER, DeploymentCandidate(600, "Pressure Transducer")] + ) + assert resolution.kind is ResolutionKind.AMBIGUOUS + assert resolution.deployment_id is None + assert resolution.candidates == (437, 600) + assert resolution.needs_a_human + + +def test_no_transducer_is_missing_not_invented(): + # SO-0246 has no open transducer deployment at all. + resolution = resolve_deployment([DIVERLINK, CABLE]) + assert resolution.kind is ResolutionKind.MISSING + assert resolution.deployment_id is None + + +def test_a_removed_transducer_is_not_a_fallback(): + # Writing today's readings against retired equipment would be worse than + # skipping the well, because it would look like it worked. + resolution = resolve_deployment( + [DeploymentCandidate(700, "Pressure Transducer", removal_date=date(2024, 1, 1))] + ) + assert resolution.kind is ResolutionKind.MISSING + + +def test_a_removed_transducer_does_not_make_a_live_one_ambiguous(): + resolution = resolve_deployment( + [ + DeploymentCandidate( + 700, "Pressure Transducer", removal_date=date(2024, 1, 1) + ), + TRANSDUCER, + ] + ) + assert resolution.deployment_id == 437 + + +def test_no_deployments_at_all(): + assert resolve_deployment([]).kind is ResolutionKind.MISSING + + +# ============= EOF ============================================= diff --git a/docs/automated-ingestion-pipeline-plan.md b/docs/automated-ingestion-pipeline-plan.md index 028321bd7..87a1349ab 100644 --- a/docs/automated-ingestion-pipeline-plan.md +++ b/docs/automated-ingestion-pipeline-plan.md @@ -287,6 +287,18 @@ Matching `BRN-E04A` against that returns a single confident hit on `SO-0131`, co **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.2 seeding — nothing needed, measured 2026-08-19 + +The plan expected to create wells, a parameter, a sensor and deployments. Checked against staging: **all of it already exists.** + +- All 38 wells have deployments — 108 open ones between them, because a deployment is a piece of equipment rather than a measured property. SO-0140 carries three: a `DiverLink` (telemetry), a `Pressure Transducer` (the reading), and a `Diver Cable`. `Barometer` appears elsewhere. +- The parameter exists: id 1, `groundwater level`, `default_unit = ft` — which is what the adapter emits, so the centimetre conversion in `domain/van_essen.py` lands in the right unit. +- Existing observations for these wells already use that parameter. + +**So the series is chosen, not created.** `sources/san_acacia/resolve.py` picks the open `Pressure Transducer` deployment. Across the 38 wells that resolves cleanly for **35**; **2** have two open transducers and **1** (SO-0246) has none. Those three are skipped and reported rather than guessed at — taking the lower id would be a silent decision about equipment. + +A *removed* transducer is not used as a fallback. Writing current readings against retired equipment would look like success while being wrong. + ### 3.3 — Represent "public but provisional" Built. Migration `b2c3d4e5f6a7` adds `data_maturity` to `transducer_observation`. diff --git a/docs/sources/san_acacia.md b/docs/sources/san_acacia.md index 787670ba2..943a61018 100644 --- a/docs/sources/san_acacia.md +++ b/docs/sources/san_acacia.md @@ -107,6 +107,38 @@ lets it read a window without decompressing and parsing every record. Objects written before this change are `.jsonl.gz`. dlt reads both, so they do not need migrating, but a replay spanning that boundary reads two formats. + +## Retention and the gap in the record + +Diver-HUB serves nothing before late 2024. Probing six points put their earliest +reading at **2024-10-08** and **2024-11-10** — matching the deployments on these +wells, installed **2024-11-25**. The vendor project was populated then. + +Ocotillo already holds AMPAPI transducer data for fourteen of these wells, +ending **2022-08-03**. + +**So the two sources never overlap, and roughly twenty-seven months are missing +from the record.** That gap cannot be filled from Diver-HUB. If the divers were +logging through it, the readings are somewhere else. + +Two consequences: + +- **The datum comparison is impossible.** Comparing the vendor's readings + against Ocotillo's existing values at matching timestamps was the plan for + confirming `reference=3` against real data. There are no matching timestamps. + Attempted on SO-0125 (Feb 2022) and SO-0245 (Jul–Aug 2022); the vendor + returned zero rows for both windows at every reference. The case for + `reference=3` therefore rests on the probe evidence — the elevation + cross-check and the 1.49 ft stickup — not on agreement with what is stored. +- **No datum mixing can occur on a normal run.** Each series resumes from its + own watermark, and the vendor has nothing to return before 2024, so the two + bodies of data stay separate by construction rather than by care. + +`INITIAL_START` is 2024-01-01 as a result: nine months of margin below the +earliest observed reading, since only six of thirty-eight points were probed. +That takes a first run from twelve windows per well to three — 228 requests +across all thirty-eight instead of 912. + ## Field mapping ### Water levels — the ingested series