From 113fdd6f4d8ef1f786c9bc8dfd27650d2b2f5ac2 Mon Sep 17 00:00:00 2001 From: jakeross Date: Wed, 19 Aug 2026 10:35:37 -0700 Subject: [PATCH 1/2] 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 --- automated_ingestion/defs/assets/__init__.py | 2 + .../sources/san_acacia/ingest.py | 184 ++++++++++++++++++ .../sources/san_acacia/resolve.py | 111 +++++++++++ automated_ingestion/tests/test_resolve.py | 93 +++++++++ docs/automated-ingestion-pipeline-plan.md | 12 ++ 5 files changed, 402 insertions(+) create mode 100644 automated_ingestion/sources/san_acacia/resolve.py create mode 100644 automated_ingestion/tests/test_resolve.py 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/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`. From ac7b5e86a1182316bd65ac3e79a420ce0eb972fe Mon Sep 17 00:00:00 2001 From: jakeross Date: Wed, 19 Aug 2026 11:02:09 -0700 Subject: [PATCH 2/2] fix(ingestion): raise the ingestion floor to 2024 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 being installed 2024-11-25 -- the vendor project was populated then. INITIAL_START was 2015-01-01, chosen before anyone knew what the vendor retains. At a 365-day span that made a first run walk twelve windows per well, ten of them guaranteed empty, against an endpoint that answers 500 when pushed. The floor is now 2024-01-01: three windows per well, 228 requests across the thirty-eight rather than 912. It sits nine months below the earliest observed reading rather than at it, because only six of thirty-eight points were probed and a well with slightly earlier data should not be silently truncated. This also records why the datum comparison could not be completed. Ocotillo's AMPAPI data for these wells ends August 2022 and the vendor starts late 2024, so there are no matching timestamps to compare -- attempted on SO-0125 and SO-0245, zero vendor rows at every reference. The two sources never overlap, which means no datum mixing can occur on a normal run, and roughly twenty-seven months are missing from the record and cannot be recovered from this source. Co-Authored-By: Claude Opus 5 --- automated_ingestion/scripts/compare_datum.py | 217 ++++++++++++++++++ .../scripts/diverhub_retention.py | 121 ++++++++++ .../sources/san_acacia/dlt_pipeline.py | 26 ++- docs/sources/san_acacia.md | 32 +++ 4 files changed, 391 insertions(+), 5 deletions(-) create mode 100644 automated_ingestion/scripts/compare_datum.py create mode 100644 automated_ingestion/scripts/diverhub_retention.py 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/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