diff --git a/api/chemisty.py b/api/chemisty.py index 5519c3f02..0fe0ed150 100644 --- a/api/chemisty.py +++ b/api/chemisty.py @@ -13,7 +13,17 @@ # See the License for the specific language governing permissions and # limitations under the License. # =============================================================================== +from datetime import datetime + from fastapi import APIRouter +from fastapi_pagination.ext.sqlalchemy import paginate +from sqlalchemy import asc, desc, select + +from api.pagination import CustomPage +from core.dependencies import amp_viewer_dependency, session_dependency +from db.chemistry_views import WaterChemistryResultsView +from schemas.chemistry import WaterChemistryResultResponse +from services.legacy_chemistry import canonical_parameter_name, result_kind # from services.validation.chemistry import validate_analyte @@ -25,6 +35,83 @@ ) +# Only columns that mean something to a client of this endpoint. A whitelist +# rather than getattr on the view: the latter would expose every column, +# including the ones carrying release state, as a public sort key. +_RESULT_SORT_COLUMNS = { + "observation_datetime": WaterChemistryResultsView.observation_datetime, + "parameter_name": WaterChemistryResultsView.parameter_name, + "value": WaterChemistryResultsView.value, + "id": WaterChemistryResultsView.id, +} + + +@router.get("/results", summary="Get water chemistry results", tags=["chemistry"]) +def get_water_chemistry_results( + session: session_dependency, + user: amp_viewer_dependency, + thing_id: int | None = None, + start_time: datetime | None = None, + end_time: datetime | None = None, + sort: str | None = None, + order: str | None = None, +) -> CustomPage[WaterChemistryResultResponse]: + """ + Retrieve water chemistry results, one row per analyte. + + Reads the legacy NMA chemistry tables, which is where the water chemistry + actually is -- the refactored `observation` table holds none of it. Rows + come from the public view, so an unreleased thing or a sample flagged + `PublicRelease = false` is not served here regardless of who is asking. + + `start_time` is inclusive and `end_time` exclusive, so a calendar year is + `start_time=YYYY-01-01&end_time=YYYY+1-01-01` with no risk of picking up a + result recorded at midnight on New Year's Day of the following year. + + `sort` accepts `observation_datetime`, `parameter_name`, `value`, or `id`; + `order` accepts `asc` or `desc`. The default is newest first, so a client + that wants a well's most recent analysis can ask for size 1. + """ + query = select(WaterChemistryResultsView) + + if thing_id is not None: + query = query.where(WaterChemistryResultsView.thing_id == thing_id) + + if start_time is not None: + query = query.where( + WaterChemistryResultsView.observation_datetime >= start_time + ) + + if end_time is not None: + query = query.where(WaterChemistryResultsView.observation_datetime < end_time) + + sort_column = _RESULT_SORT_COLUMNS.get( + sort or "observation_datetime", + WaterChemistryResultsView.observation_datetime, + ) + direction = asc if (order or "desc").lower() == "asc" else desc + + # id is the tiebreaker so paging is stable: without it two analytes sharing + # a timestamp can swap pages between requests and be served twice or never. + query = query.order_by(direction(sort_column), WaterChemistryResultsView.id) + + def transformer(rows): + # Analytes come out of the legacy tables as symbols; the response + # speaks the lexicon's names so a consumer can match a result to a + # drinking water standard without knowing the legacy vocabulary. + return [ + WaterChemistryResultResponse.model_validate(row).model_copy( + update={ + "parameter_name": canonical_parameter_name(row.parameter_name), + "result_kind": result_kind(row.id), + } + ) + for row in rows + ] + + return paginate(query=query, conn=session, transformer=transformer) + + # @router.get( # "/analysis_set", # response_model=CustomPage[WaterChemistryAnalysisSetResponse], diff --git a/core/initializers.py b/core/initializers.py index 25420615d..01ef37230 100644 --- a/core/initializers.py +++ b/core/initializers.py @@ -225,8 +225,10 @@ def register_api_routes(app): from api.feedback import router as feedback_router from api.disclaimer import router as disclaimer_router from api.geothermal import router as geothermal_router + from api.chemisty import router as chemistry_router app.include_router(asset_router) + app.include_router(chemistry_router) app.include_router(author_router) app.include_router(contact_router) app.include_router(disclaimer_router) diff --git a/db/chemistry_views.py b/db/chemistry_views.py new file mode 100644 index 000000000..925a75ee6 --- /dev/null +++ b/db/chemistry_views.py @@ -0,0 +1,77 @@ +# =============================================================================== +# 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. +# =============================================================================== +"""Read-only mappings over the legacy water-chemistry views. + +`ogc_water_chemistry` and `ogc_internal_water_chemistry` are materialized views +built in d9e0f1a2b3c4 by unioning the four legacy NMA chemistry tables +(NMA_MajorChemistry, NMA_MinorTraceChemistry, NMA_Radionuclides, +NMA_FieldParameters) into one analyte-per-row shape. They were added for the OGC +EDR mount; these mappings let the REST API serve the same rows, which is where +the chemistry data actually lives -- the refactored `observation` table holds no +water chemistry. + +Views only. Like db/ngwmn_views.py these use their own declarative base so +Alembic never tries to autogenerate a table for them, and the underlying +relations are refreshed by the migration that owns them, not from here. +""" + +from datetime import datetime + +from sqlalchemy import DateTime, Float, Integer, String +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column + + +class ChemistryViewBase(DeclarativeBase): + """Declarative base for chemistry view mappings, excluded from Alembic.""" + + +class _WaterChemistryResultColumns: + """Columns shared by the public and internal chemistry views. + + `id` is a text key (``maj-1``, ``min-2``, ``rad-3``, ``fld-4``) rather than + an integer: a row's identity is which legacy table it came from plus that + table's own id, and the four id sequences overlap. + """ + + id: Mapped[str] = mapped_column("id", String, primary_key=True) + thing_id: Mapped[int] = mapped_column("thing_id", Integer) + station_name: Mapped[str | None] = mapped_column("station_name", String) + thing_type: Mapped[str | None] = mapped_column("thing_type", String) + sample_id: Mapped[int | None] = mapped_column("sample_id", Integer) + parameter_name: Mapped[str] = mapped_column("parameter_name", String) + value: Mapped[float | None] = mapped_column("value", Float) + unit: Mapped[str | None] = mapped_column("unit", String) + # Named `datetime` in the view; exposed under the name the observation + # endpoints already use so clients do not need a second field name. + observation_datetime: Mapped[datetime] = mapped_column("datetime", DateTime) + release_status: Mapped[str | None] = mapped_column("release_status", String) + + +class WaterChemistryResultsView(_WaterChemistryResultColumns, ChemistryViewBase): + """Public chemistry analyses: released things, released samples.""" + + __tablename__ = "ogc_water_chemistry" + + +class InternalWaterChemistryResultsView( + _WaterChemistryResultColumns, ChemistryViewBase +): + """Every chemistry analysis, including unreleased things and samples.""" + + __tablename__ = "ogc_internal_water_chemistry" + + +# ============= EOF ============================================= diff --git a/schemas/chemistry.py b/schemas/chemistry.py new file mode 100644 index 000000000..10d94a0de --- /dev/null +++ b/schemas/chemistry.py @@ -0,0 +1,69 @@ +# =============================================================================== +# 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. +# =============================================================================== +from datetime import datetime, timezone +from typing import Literal + +from pydantic import BaseModel, ConfigDict, field_serializer, field_validator + + +class WaterChemistryResultResponse(BaseModel): + """One legacy chemistry analyte result. + + Not a `BaseResponseModel`: the row comes from a view over the legacy NMA + tables, so it has a text id rather than an integer one and carries no + `created_at` of its own. + """ + + id: str + thing_id: int + station_name: str | None = None + sample_id: int | None = None + parameter_name: str + value: float | None = None + unit: str | None = None + observation_datetime: datetime + # Which legacy table the result came from. A field measurement was read at + # the wellhead and a lab one was not, which is the distinction an + # owner-facing report has to draw -- and the legacy tables are the only + # place that distinction is recorded. + result_kind: Literal["major", "minor", "radionuclide", "field", "unknown"] = ( + "unknown" + ) + + model_config = ConfigDict(from_attributes=True) + + @field_validator("observation_datetime") + @classmethod + def assume_utc(cls, value: datetime) -> datetime: + """Stamp naive legacy timestamps as UTC. + + The legacy tables store collection and analysis dates without a zone -- + they are calendar dates, not instants. Attaching UTC keeps them stable: + `astimezone` on a naive value would read it in the server's local zone, + which would move a sample collected Jan 01 into the previous year for + any server west of Greenwich, and a report for that year would then come + back empty. + """ + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + @field_serializer("observation_datetime") + def serialize_observation_datetime(self, value: datetime) -> str: + return value.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +# ============= EOF ============================================= diff --git a/schemas/location.py b/schemas/location.py index e96a2474f..11139c84a 100644 --- a/schemas/location.py +++ b/schemas/location.py @@ -135,6 +135,14 @@ class LocationGeoJSONResponse(BaseModel): @model_validator(mode="before") @classmethod def populate_fields(cls, data: Any) -> Any: + # A thing can have no current location -- it is associated with one + # over an effective period, and that period can be closed or never + # opened. Hand None straight back so the optional annotation resolves + # it, rather than reaching for __table__ on it and turning a well with + # no location into a 500 for the whole page it appears on. + if data is None: + return None + # convert row to dictionary if not isinstance(data, dict): data_dict = {c.name: getattr(data, c.name) for c in data.__table__.columns} diff --git a/schemas/thing.py b/schemas/thing.py index bb2b051eb..c2798b5fd 100644 --- a/schemas/thing.py +++ b/schemas/thing.py @@ -207,7 +207,7 @@ class BaseThingResponse(BaseResponseModel): name: str site_name: str | None = None thing_type: str - current_location: LocationGeoJSONResponse + current_location: LocationGeoJSONResponse | None = None first_visit_date: PastOrTodayDate | None groups: list[GroupResponse] = [] monitoring_status: str | None diff --git a/scripts/seed_nma_chemistry.py b/scripts/seed_nma_chemistry.py new file mode 100644 index 000000000..79ea18270 --- /dev/null +++ b/scripts/seed_nma_chemistry.py @@ -0,0 +1,546 @@ +#!/usr/bin/env python3 +# =============================================================================== +# 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. +# =============================================================================== +"""Seed the test database with NMA legacy major/minor chemistry data. + +Copies a small subset of real chemistry out of a local clone of another database +(by default the ``ocotillo_prod`` clone) into ``ocotilloapi_test``, so that +chemistry endpoints, the normalized-chemistry views and the LIMS ingestion code +have realistic analytes, units, censored ("<") symbols and detection limits to +read without a SQL Server connection. + +Copied per selected ``NMA_Chemistry_SampleInfo``: + + thing (parent of the sample info; thing_id is NOT NULL) + -> location + location_thing_association (the live thing->location link) + -> NMA_Chemistry_SampleInfo + -> NMA_MajorChemistry rows + -> NMA_MinorTraceChemistry rows + +Primary keys are *not* preserved. The target already holds unrelated rows at low +ids, so every row is inserted without its id and children are repointed at the +new parent id. Legacy uuid/OBJECTID columns are copied verbatim -- they are the +natural keys this script reconciles on, which is what makes re-runs idempotent: + + NMA_Chemistry_SampleInfo."nma_SamplePtID" already present -> candidate skipped + location.nma_pk_location / thing.nma_pk_welldata already present -> reused + +Lexicon-backed columns are validated against the target ``lexicon_term`` table. +Nullable ones are nulled out when the term is missing; ``thing.thing_type`` is +NOT NULL, so a thing whose type is absent from the target lexicon disqualifies +its sample infos instead. + +The seed is transient. ``tests/conftest.py`` has a session-scoped autouse +fixture that drops and re-migrates the schema, so any ``pytest`` run wipes these +rows -- re-run this script afterwards. + +Usage: + python -m scripts.seed_nma_chemistry # 60 sample infos + python -m scripts.seed_nma_chemistry --samples 200 + python -m scripts.seed_nma_chemistry --dry-run +""" + +from __future__ import annotations + +import argparse +import getpass +import os +import sys +from collections import defaultdict +from typing import Any + +from dotenv import load_dotenv +from sqlalchemy import create_engine, text +from sqlalchemy.engine import Connection, Engine + +# Columns never copied: surrogate keys and the trigger-maintained search vector. +SKIP_COLUMNS = {"id", "search_vector"} + +# Geometry columns are read as EWKT and re-parsed on insert; pg8000 has no +# geometry codec of its own. +GEOMETRY_COLUMNS = {("location", "point")} + +# Columns whose value must exist in the target lexicon_term table. +LEXICON_COLUMNS = { + "location": {"release_status", "nma_data_reliability"}, + "thing": { + "thing_type", + "release_status", + "formation_completion_code", + "spring_type", + "well_construction_method", + "well_pump_type", + }, +} + +CHEMISTRY_TABLES = ("NMA_MajorChemistry", "NMA_MinorTraceChemistry") + + +def build_engine(database: str) -> Engine: + """Engine for `database` on the host configured in .env. + + Deliberately does not import db.engine: that module binds one engine to + POSTGRES_DB at import time, and this script needs two databases at once. + """ + password = os.environ.get("POSTGRES_PASSWORD", "") + host = os.environ.get("POSTGRES_HOST", "localhost") + port = os.environ.get("POSTGRES_PORT", "5432") + user = os.environ.get("POSTGRES_USER", "").strip() or getpass.getuser() + + auth = f"{user}:{password}@" if user and password else "" + port_part = f":{port}" if port else "" + url = f"postgresql+pg8000://{auth}{host}{port_part}/{database}" + return create_engine(url, future=True) + + +def copyable_columns(src: Connection, dst: Connection, table: str) -> list[str]: + """Columns present in both databases and safe to insert explicitly.""" + sql = text( + "select column_name from information_schema.columns " + "where table_schema = 'public' and table_name = :t" + ) + src_cols = {r[0] for r in src.execute(sql, {"t": table})} + dst_cols = {r[0] for r in dst.execute(sql, {"t": table})} + if not src_cols: + raise SystemExit(f"Source database has no table {table!r}") + if not dst_cols: + raise SystemExit(f"Target database has no table {table!r}") + + missing = (src_cols - dst_cols) | (dst_cols - src_cols) + if missing: + print(f" {table}: skipping columns absent on one side: {sorted(missing)}") + + return sorted((src_cols & dst_cols) - SKIP_COLUMNS) + + +def select_clause(table: str, columns: list[str]) -> str: + parts = [] + for col in columns: + if (table, col) in GEOMETRY_COLUMNS: + parts.append(f'ST_AsEWKT("{col}") as "{col}"') + else: + parts.append(f'"{col}"') + return ", ".join(parts) + + +def insert_returning_id( + dst: Connection, table: str, columns: list[str], row: dict[str, Any] +) -> int: + placeholders = [] + for col in columns: + if (table, col) in GEOMETRY_COLUMNS: + placeholders.append(f"ST_GeomFromEWKT(:{col})") + else: + placeholders.append(f":{col}") + + col_list = ", ".join(f'"{c}"' for c in columns) + sql = text( + f'insert into "{table}" ({col_list}) values ({", ".join(placeholders)}) ' + "returning id" + ) + return dst.execute(sql, {c: row[c] for c in columns}).scalar_one() + + +def scrub_lexicon( + table: str, row: dict[str, Any], terms: set[str], nulled: dict[str, int] +) -> None: + """Null out nullable lexicon-backed values the target lexicon lacks.""" + for col in LEXICON_COLUMNS.get(table, ()): + if col == "thing_type": # NOT NULL; handled by candidate filtering + continue + value = row.get(col) + if value is not None and value not in terms: + row[col] = None + nulled[f"{table}.{col}"] += 1 + + +def load_lexicon_terms(dst: Connection) -> set[str]: + return {r[0] for r in dst.execute(text("select term from lexicon_term"))} + + +def select_candidates( + src: Connection, dst: Connection, limit: int, terms: set[str] +) -> list[dict[str, Any]]: + """Sample infos worth copying, oldest id first for a stable subset. + + Requires both a major and a minor/trace row so the seed always exercises + both tables, and a thing whose type the target lexicon already knows. + """ + seeded = { + r[0] + for r in dst.execute( + text( + 'select "nma_SamplePtID" from "NMA_Chemistry_SampleInfo" ' + 'where "nma_SamplePtID" is not null' + ) + ) + } + + rows = src.execute(text(""" + select si.id, si."nma_SamplePtID", si.thing_id, t.thing_type + from "NMA_Chemistry_SampleInfo" si + join thing t on t.id = si.thing_id + where si."nma_SamplePtID" is not null + and exists (select 1 from "NMA_MajorChemistry" mc + where mc.chemistry_sample_info_id = si.id) + and exists (select 1 from "NMA_MinorTraceChemistry" mt + where mt.chemistry_sample_info_id = si.id) + order by si.id + """)).mappings() + + candidates = [] + skipped_seeded = 0 + skipped_type = 0 + for row in rows: + if row["nma_SamplePtID"] in seeded: + skipped_seeded += 1 + continue + if row["thing_type"] not in terms: + skipped_type += 1 + continue + candidates.append(dict(row)) + if len(candidates) >= limit: + break + + if skipped_seeded: + print(f" {skipped_seeded} sample info(s) already seeded, skipped") + if skipped_type: + print( + f" {skipped_type} sample info(s) skipped: thing_type not in target lexicon" + ) + return candidates + + +def copy_location( + src: Connection, + dst: Connection, + columns: list[str], + source_location_id: int, + location_map: dict[int, int], + terms: set[str], + nulled: dict[str, int], +) -> int | None: + """Copy one source location, reusing a target row when already present.""" + if source_location_id in location_map: + return location_map[source_location_id] + + row = ( + src.execute( + text( + f"select {select_clause('location', columns)} from location " + "where id = :i" + ), + {"i": source_location_id}, + ) + .mappings() + .first() + ) + if row is None: + return None + row = dict(row) + + legacy_key = row.get("nma_pk_location") + if legacy_key is not None: + existing = dst.execute( + text("select id from location where nma_pk_location = :k limit 1"), + {"k": legacy_key}, + ).scalar() + if existing is not None: + location_map[source_location_id] = existing + return existing + + scrub_lexicon("location", row, terms, nulled) + target_id = insert_returning_id(dst, "location", columns, row) + location_map[source_location_id] = target_id + return target_id + + +def copy_location_associations( + src: Connection, + dst: Connection, + column_sets: dict[str, list[str]], + source_thing_id: int, + target_thing_id: int, + location_map: dict[int, int], + terms: set[str], + nulled: dict[str, int], +) -> tuple[int, int]: + """Copy a thing's locations and the association rows that link them. + + thing.nma_pk_location is a legacy audit column; the live model reaches a + location through location_thing_association (Thing.location_associations), + so a seeded thing without association rows reads as a well with no location. + """ + assoc_columns = column_sets["location_thing_association"] + rows = ( + src.execute( + text( + f"select {select_clause('location_thing_association', assoc_columns)} " + "from location_thing_association where thing_id = :i order by id" + ), + {"i": source_thing_id}, + ) + .mappings() + .all() + ) + + locations = 0 + associations = 0 + for row in rows: + payload = dict(row) + source_location_id = payload["location_id"] + before = len(location_map) + target_location_id = copy_location( + src, + dst, + column_sets["location"], + source_location_id, + location_map, + terms, + nulled, + ) + if target_location_id is None: + continue + if len(location_map) > before: + locations += 1 + + payload["location_id"] = target_location_id + payload["thing_id"] = target_thing_id + insert_returning_id(dst, "location_thing_association", assoc_columns, payload) + associations += 1 + + return locations, associations + + +def copy_thing( + src: Connection, + dst: Connection, + columns: list[str], + thing_id: int, + terms: set[str], + nulled: dict[str, int], +) -> tuple[int, bool]: + """Return (target thing id, created) for a source thing id.""" + row = ( + src.execute( + text(f"select {select_clause('thing', columns)} from thing where id = :i"), + {"i": thing_id}, + ) + .mappings() + .first() + ) + if row is None: + raise SystemExit(f"Source thing {thing_id} vanished mid-run") + row = dict(row) + + legacy_key = row.get("nma_pk_welldata") + if legacy_key is not None: + existing = dst.execute( + text("select id from thing where nma_pk_welldata = :k limit 1"), + {"k": legacy_key}, + ).scalar() + if existing is not None: + return existing, False + + scrub_lexicon("thing", row, terms, nulled) + return insert_returning_id(dst, "thing", columns, row), True + + +def copy_chemistry( + src: Connection, + dst: Connection, + table: str, + columns: list[str], + source_sample_info_id: int, + target_sample_info_id: int, +) -> int: + rows = ( + src.execute( + text( + f'select {select_clause(table, columns)} from "{table}" ' + "where chemistry_sample_info_id = :i order by id" + ), + {"i": source_sample_info_id}, + ) + .mappings() + .all() + ) + + count = 0 + for row in rows: + payload = dict(row) + payload["chemistry_sample_info_id"] = target_sample_info_id + insert_returning_id(dst, table, columns, payload) + count += 1 + return count + + +def seed(source_db: str, target_db: str, samples: int, dry_run: bool) -> int: + source_engine = build_engine(source_db) + target_engine = build_engine(target_db) + + nulled: dict[str, int] = defaultdict(int) + totals: dict[str, int] = defaultdict(int) + + with source_engine.connect() as src, target_engine.begin() as dst: + print(f"Reading {source_db!r}, writing {target_db!r}") + + column_sets = { + table: copyable_columns(src, dst, table) + for table in ( + "location", + "thing", + "location_thing_association", + "NMA_Chemistry_SampleInfo", + *CHEMISTRY_TABLES, + ) + } + terms = load_lexicon_terms(dst) + + candidates = select_candidates(src, dst, samples, terms) + if not candidates: + print("Nothing to seed: no unseeded sample infos matched.") + return 0 + print(f"Selected {len(candidates)} sample info(s) to copy") + + if dry_run: + for candidate in candidates[:10]: + print( + f" would copy sample_info id={candidate['id']} " + f"thing_id={candidate['thing_id']}" + ) + if len(candidates) > 10: + print(f" ... and {len(candidates) - 10} more") + dst.rollback() + return 0 + + thing_map: dict[int, int] = {} + location_map: dict[int, int] = {} + for candidate in candidates: + source_thing_id = candidate["thing_id"] + if source_thing_id not in thing_map: + target_thing_id, created = copy_thing( + src, dst, column_sets["thing"], source_thing_id, terms, nulled + ) + thing_map[source_thing_id] = target_thing_id + if created: + totals["thing"] += 1 + + if created: + locations, associations = copy_location_associations( + src, + dst, + column_sets, + source_thing_id, + target_thing_id, + location_map, + terms, + nulled, + ) + totals["location"] += locations + totals["location_thing_association"] += associations + + info_columns = column_sets["NMA_Chemistry_SampleInfo"] + info_row = ( + src.execute( + text( + f"select {select_clause('NMA_Chemistry_SampleInfo', info_columns)} " + 'from "NMA_Chemistry_SampleInfo" where id = :i' + ), + {"i": candidate["id"]}, + ) + .mappings() + .first() + ) + payload = dict(info_row) + payload["thing_id"] = thing_map[source_thing_id] + + target_info_id = insert_returning_id( + dst, "NMA_Chemistry_SampleInfo", info_columns, payload + ) + totals["NMA_Chemistry_SampleInfo"] += 1 + + for table in CHEMISTRY_TABLES: + totals[table] += copy_chemistry( + src, + dst, + table, + column_sets[table], + candidate["id"], + target_info_id, + ) + + print("\nSeeded:") + for table in ( + "location", + "thing", + "location_thing_association", + "NMA_Chemistry_SampleInfo", + *CHEMISTRY_TABLES, + ): + print(f" {table}: {totals[table]}") + if nulled: + print("\nNulled lexicon-backed values missing from the target lexicon:") + for key, count in sorted(nulled.items()): + print(f" {key}: {count}") + return 0 + + +def main() -> int: + load_dotenv(override=False) + + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--source-db", + default="ocotillo_prod", + help="database to read from (default: ocotillo_prod)", + ) + parser.add_argument( + "--target-db", + default="ocotilloapi_test", + help="database to write to (default: ocotilloapi_test)", + ) + parser.add_argument( + "--samples", + type=int, + default=60, + help="number of sample infos to copy (default: 60)", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="report what would be copied, write nothing", + ) + parser.add_argument( + "--force", + action="store_true", + help="allow a target database whose name lacks 'test'", + ) + args = parser.parse_args() + + if "test" not in args.target_db and not args.force: + parser.error( + f"refusing to write to {args.target_db!r}: name does not contain " + "'test'. Pass --force if this is really intended." + ) + if args.source_db == args.target_db: + parser.error("--source-db and --target-db must differ") + + return seed(args.source_db, args.target_db, args.samples, args.dry_run) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/services/legacy_chemistry.py b/services/legacy_chemistry.py new file mode 100644 index 000000000..a387a76b4 --- /dev/null +++ b/services/legacy_chemistry.py @@ -0,0 +1,204 @@ +# =============================================================================== +# 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. +# =============================================================================== +"""Legacy analyte symbols to the lexicon's parameter names. + +The legacy NMA chemistry tables record analytes as symbols (`As`, `SO4`, +`pHf`). Every consumer that wants to say something about a result -- compare it +to a drinking water standard, group it, print it for a well owner -- needs the +name, because that is what the rest of the system keys on. Doing that mapping +per consumer means each one gets to be wrong on its own; doing it here means a +symbol resolves the same way everywhere. + +An unrecognized symbol passes through unchanged rather than being dropped: the +result is still real and still worth printing, it just carries no name anything +can look up. + +Deliberate omissions +-------------------- +Ambiguous symbols are left unmapped so nothing downstream can act on a guess. +A parameter with no recognized name is reported without a standards comparison, +which is the safe outcome -- inventing a name is what would let a limit be +applied to the wrong quantity: + +- ``NO3``/``NO2`` map to the as-NO3/as-NO2 names, not the as-N ones. The + nitrate MCL is 10 mg/L *as N*, which is about 45 mg/L as NO3; mapping the + wrong one flags every moderately nitrated well in the state. ``NO3(N)`` and + ``NO2(N)`` are the as-N measurements and do map. +- ``CN6``, ``DO``, ``ORP``, ``C14_years``, ``CF``, ``CFC*``, ``GA``, ``GB``, + ``Ra226``, ``Sr90``: no unambiguous lexicon term, so no mapping. +""" + +# Symbol -> lexicon `parameter_name` term. Keys are matched case-sensitively +# first, then case-insensitively, since the legacy tables are inconsistent +# about capitalizing symbols. +LEGACY_ANALYTE_NAMES: dict[str, str] = { + # --- Major ions and whole-water measures --- + "Ca": "Calcium", + "Ca(total)": "Calcium, total, unfiltered", + "Mg": "Magnesium", + "Mg(total)": "Magnesium, total, unfiltered", + "Na": "Sodium", + "Na(total)": "Sodium, total, unfiltered", + "K": "Potassium", + "K(total)": "Potassium, total, unfiltered", + "HCO3": "Bicarbonate", + "CO3": "Carbonate", + "SO4": "Sulfate", + "Cl": "Chloride", + "F": "Fluoride", + "Br": "Bromide", + "TDS": "Total Dissolved Solids", + "HRD": "Hardness (CaCO3)", + "ALK": "Alkalinity, Total", + "IONBAL": "Ion Balance", + "TAn": "Total Anions", + "TCat": "Total Cations", + "PO4": "Phosphate", + "NO3": "Nitrate (as NO3)", + "NO3(N)": "Nitrate (as N)", + "NO2": "Nitrite (as NO2)", + "NO2(N)": "Nitrite (as N)", + "NH4": "Ammonium", + "H2S": "Hydrogen sulfide", + "DOC": "Dissolved organic carbon", + "TOC": "Total organic carbon", + "TKN": "Total Kjeldahl nitrogen", + "TN": "Total nitrogen", + "SiO2": "Silica", + "Si": "Silicon", + "Si(total)": "Silicon, total, unfiltered", + # --- Metals and trace elements --- + "Ag": "Silver", + "Ag(total)": "Silver, total, unfiltered", + "Al": "Aluminum", + "Al(total)": "Aluminum, total, unfiltered", + "As": "Arsenic", + "As(total)": "Arsenic, total, unfiltered", + "B": "Boron", + "B(total)": "Boron, total, unfiltered", + "Ba": "Barium", + "Ba(total)": "Barium, total, unfiltered", + "Be": "Beryllium", + "Be(total)": "Beryllium, total, unfiltered", + "Cd": "Cadmium", + "Cd(total)": "Cadmium, total, unfiltered", + "Co": "Cobalt", + "Co(total)": "Cobalt, total, unfiltered", + "Cr": "Chromium", + "Cr(total)": "Chromium, total, unfiltered", + "Cu": "Copper", + "Cu(total)": "Copper, total, unfiltered", + "Fe": "Iron", + "Fe(total)": "Iron, total, unfiltered", + "Hg": "Mercury", + "Hg(total)": "Mercury, total, unfiltered", + "Li": "Lithium", + "Li(total)": "Lithium, total, unfiltered", + "Mn": "Manganese", + "Mn(total)": "Manganese, total, unfiltered", + "Mo": "Molybdenum", + "Mo(total)": "Molybdenum, total, unfiltered", + "Ni": "Nickel", + "Ni(total)": "Nickel, total, unfiltered", + "Pb": "Lead", + "Pb(total)": "Lead, total, unfiltered", + "Sb": "Antimony", + "Sb(total)": "Antimony, total, unfiltered", + "Se": "Selenium", + "Se(total)": "Selenium, total, unfiltered", + "Sn": "Tin", + "Sn(total)": "Tin, total, unfiltered", + "Sr": "Strontium", + "Sr(total)": "Strontium, total, unfiltered", + "Th": "Thorium", + "Th(total)": "Thorium, total, unfiltered", + "Ti": "Titanium", + "Ti(total)": "Titanium, total, unfiltered", + "Tl": "Thallium", + "Tl(total)": "Thallium, total, unfiltered", + # The uranium MCL (0.03 mg/L) is for total uranium; the lexicon spells the + # measurement it belongs to with the method it is usually run by. + "U": "Uranium (total, by ICP-MS)", + "U(total)": "Uranium, total, unfiltered", + "V": "Vanadium", + "V(total)": "Vanadium, total, unfiltered", + "Zn": "Zinc", + "Zn(total)": "Zinc, total, unfiltered", + # --- Field and laboratory measurements --- + # Field and lab pH are the same quantity to the lexicon; which instrument + # read it is carried by the source table, not by the parameter name. + "pHf": "pH", + "pHL": "pH", + "T": "temperature", + "CONDLAB": "Conductivity, laboratory", + # --- Isotopes --- + "3H": "Tritium", + "H2r": "Deuterium:Hydrogen ratio", + "O18r": "18O:16O ratio", + "C13r": "13C:12C ratio", + "C14": "14C content, pmc", + "d18O-SO4": "delta O18 sulfate", + "d34S-SO4": "Sulfate 34 isotope ratio", +} + +_LEGACY_ANALYTE_NAMES_LOWER = { + symbol.lower(): name for symbol, name in LEGACY_ANALYTE_NAMES.items() +} + + +def canonical_parameter_name(symbol: str | None) -> str | None: + """The lexicon parameter name for a legacy analyte symbol. + + Returns the symbol unchanged when it is not one this module knows about. + """ + if symbol is None: + return None + + trimmed = symbol.strip() + if not trimmed: + return trimmed + + if trimmed in LEGACY_ANALYTE_NAMES: + return LEGACY_ANALYTE_NAMES[trimmed] + + return _LEGACY_ANALYTE_NAMES_LOWER.get(trimmed.lower(), trimmed) + + +# The view's text ids are prefixed with the legacy table they came from. That +# prefix is the only record of whether a result was read in the field or by a +# lab, so it is translated into something a client can read rather than being +# left for each client to parse out of an id. +_RESULT_KINDS = { + "maj": "major", + "min": "minor", + "rad": "radionuclide", + "fld": "field", +} + + +def result_kind(result_id: str | None) -> str: + """Which legacy chemistry table a view row came from.""" + if not result_id: + return "unknown" + + prefix, _, remainder = result_id.partition("-") + if not remainder: + return "unknown" + + return _RESULT_KINDS.get(prefix, "unknown") + + +# ============= EOF ============================================= diff --git a/tests/test_legacy_chemistry_names.py b/tests/test_legacy_chemistry_names.py new file mode 100644 index 000000000..3715a811a --- /dev/null +++ b/tests/test_legacy_chemistry_names.py @@ -0,0 +1,97 @@ +# =============================================================================== +# 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. +# =============================================================================== +"""Legacy analyte symbol to lexicon parameter name mapping.""" + +import pytest + +from services.legacy_chemistry import canonical_parameter_name, result_kind + + +@pytest.mark.parametrize( + "symbol, expected", + [ + ("As", "Arsenic"), + ("Pb", "Lead"), + ("SO4", "Sulfate"), + ("TDS", "Total Dissolved Solids"), + ("HRD", "Hardness (CaCO3)"), + ("pHf", "pH"), + ("pHL", "pH"), + ("As(total)", "Arsenic, total, unfiltered"), + ("H2r", "Deuterium:Hydrogen ratio"), + ], +) +def test_maps_legacy_symbols_to_lexicon_names(symbol, expected): + assert canonical_parameter_name(symbol) == expected + + +def test_distinguishes_nitrate_as_n_from_nitrate_as_no3(): + """The nitrate MCL is 10 mg/L *as N*, roughly 45 mg/L as NO3. + + Collapsing the two would apply the as-N limit to an as-NO3 number and flag + wells that are nowhere near it, so only the as-N measurement gets the name + the standard is keyed to. + """ + assert canonical_parameter_name("NO3(N)") == "Nitrate (as N)" + assert canonical_parameter_name("NO3") == "Nitrate (as NO3)" + assert canonical_parameter_name("NO2(N)") == "Nitrite (as N)" + assert canonical_parameter_name("NO2") == "Nitrite (as NO2)" + + +@pytest.mark.parametrize("symbol", ["CN6", "DO", "ORP", "C14_years", "GA", "Ra226"]) +def test_leaves_ambiguous_symbols_alone(symbol): + """An unmapped symbol is reported as-is and compared to nothing. + + Guessing a name is what would let a limit be applied to the wrong quantity. + """ + assert canonical_parameter_name(symbol) == symbol + + +def test_tolerates_legacy_capitalization_and_padding(): + assert canonical_parameter_name(" as ") == "Arsenic" + assert canonical_parameter_name("TDS ") == "Total Dissolved Solids" + + +def test_passes_through_unknown_and_empty_values(): + assert canonical_parameter_name("NotAnAnalyte") == "NotAnAnalyte" + assert canonical_parameter_name("") == "" + assert canonical_parameter_name(None) is None + + +@pytest.mark.parametrize( + "result_id, expected", + [ + ("maj-1", "major"), + ("min-19198", "minor"), + ("rad-7", "radionuclide"), + ("fld-42", "field"), + ], +) +def test_reads_the_source_table_off_the_id(result_id, expected): + """A field measurement was read at the wellhead and a lab one was not. + + The view's id prefix is the only place that survives, so a client is told + which it is rather than being left to parse an id. + """ + assert result_kind(result_id) == expected + + +@pytest.mark.parametrize("result_id", ["", None, "1234", "unprefixed-", "zzz-1"]) +def test_unrecognized_ids_report_an_unknown_source(result_id): + assert result_kind(result_id) == "unknown" + + +# ============= EOF ============================================= diff --git a/tests/test_thing_without_location.py b/tests/test_thing_without_location.py new file mode 100644 index 000000000..9031ad854 --- /dev/null +++ b/tests/test_thing_without_location.py @@ -0,0 +1,61 @@ +# =============================================================================== +# 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. +# =============================================================================== +"""A thing with no current location must not take down the page it is on. + +A thing is associated with a location over an effective period, and that period +can be closed or never opened. When `current_location` was a required field the +GeoJSON validator was handed None, reached for `__table__` on it, and the whole +listing came back 500 -- one unlocated well made every well unreadable. +""" + +from db.engine import session_ctx +from db.thing import Thing +from main import app +from schemas.location import LocationGeoJSONResponse +from starlette.testclient import TestClient + +client = TestClient(app) + + +def test_geojson_validator_passes_none_through(): + assert LocationGeoJSONResponse.populate_fields(None) is None + + +def test_listing_wells_survives_one_with_no_location(water_well_thing): + unlocated = Thing( + name="TEST-NOLOC-1", + thing_type="water well", + release_status="public", + ) + with session_ctx() as session: + session.add(unlocated) + session.commit() + unlocated_id = unlocated.id + + try: + response = client.get("/thing/water-well", params={"size": 100}) + assert response.status_code == 200, response.text + + items = {item["id"]: item for item in response.json()["items"]} + assert unlocated_id in items + assert items[unlocated_id]["current_location"] is None + finally: + with session_ctx() as session: + session.delete(session.get(Thing, unlocated_id)) + session.commit() + + +# ============= EOF =============================================