From e13c5c3a23f49a93163973f610a7ed59243314df Mon Sep 17 00:00:00 2001 From: jross Date: Wed, 8 Jul 2026 10:08:13 -0600 Subject: [PATCH 1/4] feat(cli): add chemistry LIMS ingestion with Google Drive sync Add `oco water-chemistry` commands to ingest lab LIMS .xlsx workbooks into the legacy NMA chemistry tables (NMA_MajorChemistry / NMA_MinorTraceChemistry), porting the analyte mapping, non-detect handling, and EPA-200.7 / low-bromide dedup from the AMPAPI chemfile.py script. - bulk-upload: parse a LIMS .xlsx and load a single file - sync-drive: on-demand, engineer-triggered ingest of new/changed files from a Google Drive folder, with an ingested-file manifest stored in GCS - append semantics: each distinct lab sample (WCLab_ID) becomes a new sample point using the next letter incrementor on the base PointID (MG-030A, MG-030B, ... Z, AA); a lab sample already recorded for the well is skipped, so re-runs are idempotent - add a `cli` dependency group (openpyxl, google-api-python-client); CI syncs it explicitly, the production requirements export excludes it - add docs/chemistry-ingestion-runbook.md Refs BDMS-1034 Co-Authored-By: Claude Opus 4.8 --- .env.example | 7 + .github/workflows/jira_codex_pr.yml | 2 +- .github/workflows/tests.yml | 4 +- cli/cli.py | 217 ++++++++++ cli/service_adapter.py | 12 + docs/chemistry-ingestion-runbook.md | 212 ++++++++++ pyproject.toml | 8 + services/chemistry_drive.py | 310 ++++++++++++++ services/chemistry_lims.py | 634 ++++++++++++++++++++++++++++ tests/test_chemistry_drive.py | 219 ++++++++++ tests/test_chemistry_lims.py | 287 +++++++++++++ uv.lock | 79 ++++ 12 files changed, 1988 insertions(+), 3 deletions(-) create mode 100644 docs/chemistry-ingestion-runbook.md create mode 100644 services/chemistry_drive.py create mode 100644 services/chemistry_lims.py create mode 100644 tests/test_chemistry_drive.py create mode 100644 tests/test_chemistry_lims.py diff --git a/.env.example b/.env.example index 3b53b9ae7..dfdc98844 100644 --- a/.env.example +++ b/.env.example @@ -55,6 +55,13 @@ TRANSFER_NMW_MIRROR=True # load the NMW_* 1:1 staging mirror GCS_BUCKET_NAME= GOOGLE_APPLICATION_CREDENTIALS=/path/to/gcs_credentials.json +# chemistry ingestion: Google Drive folder an engineer ingests LIMS .xlsx +# workbooks from, on demand (no polling/scheduling; service account / ADC must +# have read access to this folder). The ingested-file manifest is stored in +# GCS_BUCKET_NAME at CHEMISTRY_INGEST_MANIFEST_PATH. +CHEMISTRY_DRIVE_FOLDER_ID= +CHEMISTRY_INGEST_MANIFEST_PATH=chemistry-ingest/manifest.json + # set to development for lexicon and parameter to be populated and enable the enums to work MODE=development diff --git a/.github/workflows/jira_codex_pr.yml b/.github/workflows/jira_codex_pr.yml index 344177723..110e0ce79 100644 --- a/.github/workflows/jira_codex_pr.yml +++ b/.github/workflows/jira_codex_pr.yml @@ -71,7 +71,7 @@ jobs: - name: Sync dependencies (pyproject/uv.lock) run: | set -euo pipefail - uv sync --all-extras --dev + uv sync --all-extras --dev --group cli - name: Verify tooling exists run: | diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 72f35451e..272b4bcc8 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -82,7 +82,7 @@ jobs: key: venv-${{ runner.os }}-${{ steps.setup-python.outputs.python-version }}-${{ hashFiles('uv.lock') }} - name: Install the project - run: uv sync --locked --all-extras --dev + run: uv sync --locked --all-extras --dev --group cli - name: Show Alembic heads run: uv run alembic heads @@ -174,7 +174,7 @@ jobs: key: venv-${{ runner.os }}-${{ steps.setup-python.outputs.python-version }}-${{ hashFiles('uv.lock') }} - name: Install the project - run: uv sync --locked --all-extras --dev + run: uv sync --locked --all-extras --dev --group cli - name: Show Alembic heads run: uv run alembic heads diff --git a/cli/cli.py b/cli/cli.py index 14c8f9470..8a6432e4f 100644 --- a/cli/cli.py +++ b/cli/cli.py @@ -32,8 +32,10 @@ cli = typer.Typer(help="Command line interface for managing the application.") water_levels = typer.Typer(help="Water-level utilities") +water_chemistry = typer.Typer(help="Water-chemistry utilities") data_migrations = typer.Typer(help="Data migration utilities") cli.add_typer(water_levels, name="water-levels") +cli.add_typer(water_chemistry, name="water-chemistry") cli.add_typer(data_migrations, name="data-migrations") @@ -959,6 +961,221 @@ def water_levels_bulk_upload( raise typer.Exit(result.exit_code) +@water_chemistry.command("bulk-upload") +def water_chemistry_bulk_upload( + file_path: str = typer.Option( + ..., + "--file", + exists=True, + file_okay=True, + dir_okay=False, + readable=True, + help="Path to LIMS .xlsx workbook containing chemistry results.", + ), + output_format: OutputFormat | None = typer.Option( + None, + "--output", + help="Optional output format", + ), + theme: ThemeMode = typer.Option( + ThemeMode.auto, "--theme", help="Color theme: auto, light, dark." + ), +): + """ + parse a LIMS chemistry workbook and load it into the NMA Major/Minor + chemistry tables. All-or-nothing: if any analyte already exists, or any row + fails to map, nothing is written. + """ + from cli.service_adapter import chemistry_lims_xlsx + + colors = _palette(theme) + result = chemistry_lims_xlsx( + file_path, pretty_json=output_format == OutputFormat.json + ) + + if output_format == OutputFormat.json: + typer.echo(result.stdout) + raise typer.Exit(result.exit_code) + + payload = result.payload if isinstance(result.payload, dict) else {} + summary = payload.get("summary", {}) + validation_errors = payload.get("validation_errors", []) + created_samples = payload.get("created_samples", []) + skipped_duplicates = payload.get("skipped_duplicates", []) + + if result.exit_code == 0: + typer.secho("[WATER CHEMISTRY IMPORT] SUCCESS", fg=colors["ok"], bold=True) + else: + typer.secho( + "[WATER CHEMISTRY IMPORT] COMPLETED WITH ISSUES", + fg=colors["issue"], + bold=True, + ) + typer.secho("=" * 72, fg=colors["accent"]) + + if summary: + processed = summary.get("total_rows_processed", 0) + imported = summary.get("total_rows_imported", 0) + rows_with_issues = summary.get("validation_errors_or_warnings", 0) + typer.secho("SUMMARY", fg=colors["accent"], bold=True) + label_width = 16 + value_width = 8 + typer.secho(" " + "-" * (label_width + 3 + value_width), fg=colors["muted"]) + typer.secho( + f" {'processed':<{label_width}} | {processed:>{value_width}}", + fg=colors["accent"], + ) + typer.secho( + f" {'imported':<{label_width}} | {imported:>{value_width}}", + fg=colors["ok"], + ) + issue_color = colors["issue"] if rows_with_issues else colors["ok"] + typer.secho( + f" {'rows_with_issues':<{label_width}} | {rows_with_issues:>{value_width}}", + fg=issue_color, + ) + typer.echo() + + if created_samples: + typer.secho("CREATED SAMPLES", fg=colors["ok"], bold=True) + for sample in created_samples: + typer.secho( + f" - {sample['sample_point_id']} " + f"(WCLab_ID {sample.get('wclab_id')}): {sample.get('rows', 0)} row(s)", + fg=colors["ok"], + ) + typer.echo() + + if skipped_duplicates: + typer.secho("SKIPPED (already ingested)", fg=colors["muted"], bold=True) + for dupe in skipped_duplicates: + typer.secho( + f" - {dupe['pointid']} (WCLab_ID {dupe.get('wclab_id')})", + fg=colors["field"], + ) + typer.echo() + + if validation_errors: + typer.secho("VALIDATION", fg=colors["accent"], bold=True) + typer.secho( + f"Validation errors: {len(validation_errors)}", + fg=colors["issue"], + bold=True, + ) + for entry in validation_errors[:25]: + typer.secho(f" - {entry}", fg=colors["issue"]) + if len(validation_errors) > 25: + typer.secho( + f"... and {len(validation_errors) - 25} more validation errors", + fg=colors["issue"], + ) + + typer.secho("=" * 72, fg=colors["accent"]) + raise typer.Exit(result.exit_code) + + +@water_chemistry.command("sync-drive") +def water_chemistry_sync_drive( + folder_id: str = typer.Option( + None, + "--folder-id", + help="Google Drive folder id to scan. Defaults to $CHEMISTRY_DRIVE_FOLDER_ID.", + ), + dry_run: bool = typer.Option( + False, + "--dry-run", + help="List new files without downloading, ingesting, or updating the manifest.", + ), + output_format: OutputFormat | None = typer.Option( + None, + "--output", + help="Optional output format", + ), + theme: ThemeMode = typer.Option( + ThemeMode.auto, "--theme", help="Color theme: auto, light, dark." + ), +): + """ + ingest LIMS chemistry workbooks from a Google Drive folder. Run on demand + by an engineer -- this does no polling or scheduling. New or changed files + are ingested; a manifest of ingested files is kept in GCS so + already-processed files are skipped. + """ + import json as _json + + from services.chemistry_drive import ChemistryDriveConfigError, sync_and_ingest + + colors = _palette(theme) + try: + result = sync_and_ingest(folder_id=folder_id, dry_run=dry_run) + except ChemistryDriveConfigError as exc: + typer.secho(str(exc), fg=colors["issue"], bold=True, err=True) + raise typer.Exit(1) from exc + + if output_format == OutputFormat.json: + typer.echo(_json.dumps(result.to_payload())) + raise typer.Exit(result.exit_code) + + summary = result.to_payload()["summary"] + header = ( + "[CHEMISTRY DRIVE SYNC] DRY RUN" if result.dry_run else "[CHEMISTRY DRIVE SYNC]" + ) + header_color = colors["ok"] if result.exit_code == 0 else colors["issue"] + typer.secho(header, fg=header_color, bold=True) + typer.secho("=" * 72, fg=colors["accent"]) + typer.secho(f"Folder: {result.folder_id}", fg=colors["accent"]) + typer.echo() + + typer.secho("SUMMARY", fg=colors["accent"], bold=True) + for label, value, color in ( + ("files_seen", summary["files_seen"], colors["accent"]), + ("new_files", summary["new_files"], colors["accent"]), + ("ingested", summary["ingested"], colors["ok"]), + ("skipped", summary["skipped"], colors["muted"]), + ( + "failed", + summary["failed"], + colors["issue"] if summary["failed"] else colors["ok"], + ), + ): + typer.secho(f" {label:<12} | {value:>6}", fg=color) + typer.echo() + + if result.dry_run and result.new_files: + typer.secho("NEW FILES (not ingested)", fg=colors["accent"], bold=True) + for name in result.new_files: + typer.secho(f" - {name}", fg=colors["field"]) + typer.echo() + + if result.ingested: + typer.secho("INGESTED", fg=colors["ok"], bold=True) + for record in result.ingested: + typer.secho( + f" - {record['name']}: {record.get('rows_imported', 0)} row(s)", + fg=colors["ok"], + ) + typer.echo() + + if result.failed: + typer.secho("FAILED", fg=colors["issue"], bold=True) + for record in result.failed: + detail = record.get("error") + if not detail: + payload = record.get("payload", {}) + errors = payload.get("validation_errors") or [] + if errors: + detail = errors[0] + if len(errors) > 1: + detail += f" (+{len(errors) - 1} more)" + else: + detail = "ingestion aborted" + typer.secho(f" - {record['name']}: {detail}", fg=colors["issue"]) + typer.echo() + + typer.secho("=" * 72, fg=colors["accent"]) + raise typer.Exit(result.exit_code) + + @data_migrations.command("list") def data_migrations_list( theme: ThemeMode = typer.Option( diff --git a/cli/service_adapter.py b/cli/service_adapter.py index bc7bb6ccf..9b7c4393e 100644 --- a/cli/service_adapter.py +++ b/cli/service_adapter.py @@ -89,6 +89,18 @@ def water_levels_csv(source_file: Path | str, *, pretty_json: bool = False): return result +def chemistry_lims_xlsx(source_file: Path | str, *, pretty_json: bool = False): + from services.chemistry_lims import bulk_upload_chemistry + + if isinstance(source_file, str): + source_file = Path(source_file) + + result = bulk_upload_chemistry(source_file, pretty_json=pretty_json) + if result.stderr: + print(result.stderr, file=sys.stderr) + return result + + def associate_assets(source_directory: Path | str) -> list[str]: """ given a directory diff --git a/docs/chemistry-ingestion-runbook.md b/docs/chemistry-ingestion-runbook.md new file mode 100644 index 000000000..d978edc95 --- /dev/null +++ b/docs/chemistry-ingestion-runbook.md @@ -0,0 +1,212 @@ +# Chemistry Ingestion — Interim Workaround Runbook + +Purpose: capture the known gaps, assumptions, and step-by-step process for +Data Services (Ocotillo) chemistry ingestion, so the manual workaround can be +run consistently while the fuller solution is pursued. + +- Jira: [BDMS-1034](https://nmbgmr.atlassian.net/browse/BDMS-1034). +- Code (Data Services path): + - `services/chemistry_lims.py` — parse a LIMS `.xlsx` and load the legacy + NMA chemistry tables (analyte mapping ported from AMPAPI `chemfile.py`). + - `services/chemistry_drive.py` — on-demand ingest of new files from the + shared Drive folder; updates the manifest. Engineer-triggered; no polling. + - `cli/cli.py` — `oco water-chemistry bulk-upload` and `oco water-chemistry + sync-drive`. +- Target tables: `NMA_MajorChemistry`, `NMA_MinorTraceChemistry`, + `NMA_Chemistry_SampleInfo` (`db/nma_legacy.py`). +- Legacy source: AMPAPI `chemfile.py` (`MajorChemistry` / + `MinorandTraceChemistry` in SQL Server). + +--- + +## 0. The workaround at a glance + +```mermaid +flowchart LR + S["Sianin
export lab LIMS batch as .xlsx"] --> D["Shared Google Drive folder
CHEMISTRY_DRIVE_FOLDER_ID"] + E["Engineer
oco water-chemistry sync-drive (on demand)"] --> D + E --> T["Ingest new/changed files →
Ocotillo NMA_* chemistry tables"] + E --> M["manifest.json in GCS
updated with per-file results"] +``` + +Chemistry is ingested into a single destination: the Ocotillo (Data Services) +Postgres database — the legacy `NMA_MajorChemistry`, `NMA_MinorTraceChemistry`, +and `NMA_Chemistry_SampleInfo` tables. + +--- + +## 1. Roles + +- **Sianin** — exports each lab chemistry batch from the LIMS as an `.xlsx` + workbook and drops it in the shared Drive folder. One workbook per batch. +- **Engineer** (e.g. Kelsey) — explicitly triggers the Data Services ingestion + CLI against the folder when a run is wanted, reviews the printed results, and + confirms the manifest updated. Acts on any files reported as `failed`. There + is no polling or scheduling — ingestion only happens when an engineer runs it. +- **Engineering** — owns the CLI, the analyte map, and the fuller solution. + +--- + +## 2. Prerequisites (Kelsey's machine) + +- [ ] Repo checked out; env installed **with the CLI group**: + `uv sync --locked --group cli` (installs `openpyxl` + + `google-api-python-client`, which are not part of the API runtime). +- [ ] Postgres (Ocotillo) reachable; `.env` has `POSTGRES_*` (or Cloud SQL) + creds pointing at the target Data Services database. +- [ ] `GCS_BUCKET_NAME` set (holds the manifest). +- [ ] `CHEMISTRY_DRIVE_FOLDER_ID` set to the shared folder id + (see `.env.example`). Optional `CHEMISTRY_INGEST_MANIFEST_PATH` + (default `chemistry-ingest/manifest.json`). +- [ ] Google credentials available with: + - **read** access to the shared Drive folder, + - **read/write** on the GCS bucket. + Locally this is application-default credentials + (`gcloud auth application-default login`) for an account added to the + folder; in production it is the base64 `GCS_SERVICE_ACCOUNT_KEY` service + account (which must be a member of the folder). + +--- + +## 3. Process — Sianin (drop files) + +1. Export the lab batch from the LIMS as an `.xlsx` workbook. It must carry the + standard LIMS columns: `Param`, `Results_Units`, `Dilution`, `AnalysisTime`, + `SampleNumber`, `CustomerSampleNumber`, `SamplePointID`, `Method`, `Test`, + `ReportedND`, `LowerLimit`, `SampleDate`. +2. Ensure `SamplePointID` matches the well's PointID / Ocotillo Thing name. +3. Drop the workbook in the shared Drive folder. Do not edit a file in place + after it has been ingested — a content change re-ingests it (by md5). + +--- + +## 4. Process — Engineer (run the ingest) + +Dry run first to see what is new without writing anything: + +```bash +oco water-chemistry sync-drive --dry-run +``` + +Then ingest: + +```bash +oco water-chemistry sync-drive +# or point at a specific folder: +oco water-chemistry sync-drive --folder-id +# machine-readable: +oco water-chemistry sync-drive --output json +``` + +Read the summary. Buckets: + +- **ingested** — file loaded; shows rows imported. +- **skipped** — the whole file was already ingested (manifest has a `success` + entry and the file's md5 is unchanged). +- **ingested with skipped samples** — a file loads, but any lab sample + (`WCLab_ID` / SampleNumber) already recorded for the well is skipped and + listed under `skipped_duplicates`; this is normal and not a failure. New lab + samples for the same well are appended as a new lettered sample point + (`MG-030A`, `MG-030B`, ...). +- **failed** — nothing loaded for that file (a data-quality abort). Causes: + +| Reported cause | Meaning | Action | +|----------------|---------|--------| +| `Unmapped analyte Param=...` | A LIMS `Param` name is not in the analyte map. | Send the Param name to engineering to add to `FMapper`. | +| `no matching Thing (well) found` | `SamplePointID` has no Ocotillo well. | Verify the PointID; ensure the well was transferred to Data Services first. | + +Exit code is non-zero if any file failed. + +A single file (bypassing Drive) can be loaded directly: + +```bash +oco water-chemistry bulk-upload --file /path/to/batch.xlsx +``` + +--- + +## 5. The manifest + +- Location: `gs://$GCS_BUCKET_NAME/` + (default `chemistry-ingest/manifest.json`). +- Keyed by **Drive file id**. Each entry records: + `name`, `md5`, `modified_time`, `status` (`success` / `failed`), + `rows_imported`, `validation_errors_or_warnings`, `ingested_at` + (and `error` for hard failures). +- Semantics: + - A file is **skipped** only when its manifest entry is `success` **and** the + Drive md5 is unchanged. + - **failed** or **content-changed** files are retried on the next run. +- The manifest is rewritten after **every** file, so an interrupted run keeps + its progress. +- Inspect it: `gsutil cat gs://$GCS_BUCKET_NAME/chemistry-ingest/manifest.json`. + +--- + +## 6. What the ingest does (summary) + +For each workbook: map each `Param` to an analyte code + target table (major vs +minor) via `FMapper`; compute the value (non-detects become +`LowerLimit × Dilution` with a `<` symbol); collapse duplicate +(SamplePointID, WCLab_ID, analyte) rows (prefer EPA 200.7, or "low bromide" for +Br); resolve the base `SamplePointID → Thing`. Then, per distinct lab sample +(`WCLab_ID`): if that lab sample is already recorded for the well, skip it; +otherwise create a new `NMA_Chemistry_SampleInfo` whose `nma_sample_point_id` +is the base PointID with the **next letter incrementor** appended +(`A`, `B`, ... `Z`, `AA`, ...), and insert the analyte rows under it. +A data-quality problem (a row that fails to map, or a `SamplePointID` with no +matching well) aborts the whole file — nothing is written. + +--- + +## 7. Known gaps + +- **Duplicate detection is WCLab_ID-only.** A re-ingest is recognized by the lab + `WCLab_ID` (SampleNumber). A genuinely new lab sample with a reused SampleNumber + would be treated as a duplicate and skipped; a re-run of the same sample under a + new SampleNumber would append a spurious extra lettered sample. +- **`.xlsx` only.** Legacy `.xls` LIMS exports are not read; the file must be + a modern `.xlsx`. +- **Fixed analyte map.** Unknown `Param` names fail until engineering adds them + to `FMapper`. Only major + minor analytes are handled — field parameters and + radionuclides are out of scope. +- **Well must exist first.** `SamplePointID` must already match an Ocotillo + `Thing.name`; otherwise the file fails. +- **Failed files retry loudly.** A file that fails (e.g. unmapped analyte or a + missing well) is retried on every run and keeps reporting `failed` until + resolved. +- **Concurrent runs race the manifest.** Ingestion is engineer-triggered on + demand by design (no polling/scheduling). But two engineers running + `sync-drive` at the same time race the manifest object (last write wins) — + coordinate so only one run is in flight. +- **No alerting.** Failures surface only in the CLI output the engineer reads. +- **Prod excludes the CLI deps.** The `cli` dependency group + (`openpyxl`, `google-api-python-client`) is not in the production requirements + export, so the ingest runs from an engineer's machine, not the deployed app. + +--- + +## 8. Assumptions + +- One lab batch per `.xlsx`, with the standard LIMS column set (section 3). +- `SamplePointID` == the well PointID == the Ocotillo `Thing.name`. +- All files in the shared folder are chemistry LIMS workbooks (the sync filters + to `.xlsx` by MIME type). +- Analyses agency is NMBGMR; non-detects and units follow the AMPAPI + `chemfile.py` conventions. +- The account running the CLI can read the Drive folder, read/write the GCS + bucket, and reach the Data Services database. + +--- + +## 9. Toward the fuller solution + +Candidate improvements, roughly in priority order: + +- **Stronger duplicate detection** than WCLab_ID alone (e.g. also compare + collection date / analyte set) so a reused or missing SampleNumber can't cause + a wrong skip or a spurious appended sample. +- **Alerting** on failed files (email/Slack) rather than relying on reading CLI + output. +- Make the **analyte map** data-driven (lexicon-backed) so new params don't + require a code change. diff --git a/pyproject.toml b/pyproject.toml index 11ff4d3f7..7759f710e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -143,6 +143,14 @@ dev = [ "python-dotenv>=1.1.1", "requests>=2.34.2", ] +# CLI-only dependencies (chemistry ingestion / Google Drive sync). These are +# imported lazily by `oco` commands and are not needed by the API runtime, so +# they are excluded from the production requirements export (`uv export +# --no-dev`). CI installs them explicitly with `uv sync --group cli`. +cli = [ + "openpyxl==3.1.5", + "google-api-python-client==2.184.0", +] [tool.pytest.ini_options] filterwarnings = [ diff --git a/services/chemistry_drive.py b/services/chemistry_drive.py new file mode 100644 index 000000000..b6f5f2ea8 --- /dev/null +++ b/services/chemistry_drive.py @@ -0,0 +1,310 @@ +# =============================================================================== +# 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. +# =============================================================================== +""" +Ingest LIMS chemistry workbooks from a Google Drive folder, on demand. + +This is explicitly engineer-triggered: it runs once per invocation and does no +polling or scheduling. New/changed ``.xlsx`` files under +``CHEMISTRY_DRIVE_FOLDER_ID`` are downloaded and handed to +:func:`services.chemistry_lims.bulk_upload_chemistry`. A manifest of +already-ingested files is kept as a JSON object in the GCS bucket +(``CHEMISTRY_INGEST_MANIFEST_PATH``, default ``chemistry-ingest/manifest.json``) +so re-runs only process files that are new or whose contents changed. + +Configuration (environment variables): +* ``CHEMISTRY_DRIVE_FOLDER_ID`` - Drive folder id to scan (shared-drive or + My-Drive folder shared with the service account). +* ``CHEMISTRY_INGEST_MANIFEST_PATH`` - GCS object key for the manifest. +* ``GCS_BUCKET_NAME`` - bucket that holds the manifest (shared with gcs_helper). + +Authentication mirrors ``services.gcs_helper``: in production the base64 +service-account key in ``GCS_SERVICE_ACCOUNT_KEY`` is used (with a Drive scope); +otherwise application-default credentials are used. The service account must be +granted at least read access to the target Drive folder. +""" + +from __future__ import annotations + +import base64 +import io +import json +import logging +import os +from dataclasses import dataclass, field +from datetime import datetime, timezone +from functools import lru_cache +from typing import Any + +from core.settings import settings +from services.chemistry_lims import bulk_upload_chemistry +from services.gcs_helper import get_storage_bucket + +logger = logging.getLogger(__name__) + +DRIVE_SCOPES = ["https://www.googleapis.com/auth/drive.readonly"] +XLSX_MIME = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" +DEFAULT_MANIFEST_PATH = "chemistry-ingest/manifest.json" + + +class ChemistryDriveConfigError(Exception): + """The Drive folder is not configured (missing folder id).""" + + +# --- Google Drive access ------------------------------------------------------- + + +def _drive_credentials(): + from google.oauth2 import service_account + + if settings.mode == "production": + key_base64 = os.environ.get("GCS_SERVICE_ACCOUNT_KEY") + if not key_base64: + raise ChemistryDriveConfigError( + "GCS_SERVICE_ACCOUNT_KEY is required for Drive access in production." + ) + decoded = base64.b64decode(key_base64).decode("utf-8") + return service_account.Credentials.from_service_account_info( + json.loads(decoded), scopes=DRIVE_SCOPES + ) + + import google.auth + + creds, _ = google.auth.default(scopes=DRIVE_SCOPES) + return creds + + +@lru_cache(maxsize=1) +def get_drive_service(): + from googleapiclient.discovery import build + + return build("drive", "v3", credentials=_drive_credentials(), cache_discovery=False) + + +def list_drive_xlsx(folder_id: str, service=None) -> list[dict]: + """List non-trashed ``.xlsx`` files directly under ``folder_id``.""" + service = service or get_drive_service() + query = ( + f"'{folder_id}' in parents " + "and trashed = false " + f"and mimeType = '{XLSX_MIME}'" + ) + files: list[dict] = [] + page_token = None + while True: + response = ( + service.files() + .list( + q=query, + spaces="drive", + corpora="allDrives", + includeItemsFromAllDrives=True, + supportsAllDrives=True, + fields="nextPageToken, files(id, name, md5Checksum, modifiedTime, size)", + pageToken=page_token, + pageSize=100, + ) + .execute() + ) + files.extend(response.get("files", [])) + page_token = response.get("nextPageToken") + if not page_token: + break + return files + + +def download_drive_file(file_id: str, service=None) -> bytes: + """Download a Drive file's bytes.""" + from googleapiclient.http import MediaIoBaseDownload + + service = service or get_drive_service() + request = service.files().get_media(fileId=file_id, supportsAllDrives=True) + buffer = io.BytesIO() + downloader = MediaIoBaseDownload(buffer, request) + done = False + while not done: + _status, done = downloader.next_chunk() + return buffer.getvalue() + + +# --- manifest (GCS JSON object) ------------------------------------------------ + + +def _manifest_path() -> str: + return os.environ.get("CHEMISTRY_INGEST_MANIFEST_PATH", DEFAULT_MANIFEST_PATH) + + +def load_manifest(bucket=None) -> dict[str, dict]: + bucket = bucket or get_storage_bucket() + blob = bucket.blob(_manifest_path()) + if not blob.exists(): + return {} + try: + return json.loads(blob.download_as_text()) + except (ValueError, json.JSONDecodeError): + logger.warning("Chemistry ingest manifest is corrupt; starting fresh.") + return {} + + +def save_manifest(manifest: dict[str, dict], bucket=None) -> None: + bucket = bucket or get_storage_bucket() + blob = bucket.blob(_manifest_path()) + blob.upload_from_string( + json.dumps(manifest, indent=2, sort_keys=True), + content_type="application/json", + ) + + +# --- orchestration ------------------------------------------------------------- + + +@dataclass +class DriveSyncResult: + folder_id: str + files_seen: int = 0 + new_files: list[str] = field(default_factory=list) + ingested: list[dict] = field(default_factory=list) + skipped: list[str] = field(default_factory=list) + failed: list[dict] = field(default_factory=list) + dry_run: bool = False + + @property + def exit_code(self) -> int: + return 1 if self.failed else 0 + + def to_payload(self) -> dict[str, Any]: + return { + "folder_id": self.folder_id, + "dry_run": self.dry_run, + "summary": { + "files_seen": self.files_seen, + "new_files": len(self.new_files), + "ingested": len(self.ingested), + "skipped": len(self.skipped), + "failed": len(self.failed), + }, + "new_files": self.new_files, + "ingested": self.ingested, + "skipped": self.skipped, + "failed": self.failed, + } + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def sync_and_ingest( + folder_id: str | None = None, + *, + dry_run: bool = False, + bucket=None, + drive_service=None, +) -> DriveSyncResult: + """Scan the Drive folder and ingest any new or changed workbooks. + + A file is considered already ingested when the manifest holds a + ``success`` entry whose ``md5`` matches the current Drive checksum. Failed + or changed files are retried on the next run. + """ + folder_id = folder_id or os.environ.get("CHEMISTRY_DRIVE_FOLDER_ID") + if not folder_id: + raise ChemistryDriveConfigError( + "No Drive folder configured. Set CHEMISTRY_DRIVE_FOLDER_ID or pass --folder-id." + ) + + bucket = bucket or get_storage_bucket() + manifest = load_manifest(bucket) + files = list_drive_xlsx(folder_id, service=drive_service) + + result = DriveSyncResult( + folder_id=folder_id, files_seen=len(files), dry_run=dry_run + ) + + for meta in files: + file_id = meta["id"] + name = meta.get("name", file_id) + md5 = meta.get("md5Checksum") + + entry = manifest.get(file_id) + already_ingested = ( + entry is not None + and entry.get("status") == "success" + and entry.get("md5") == md5 + ) + if already_ingested: + result.skipped.append(name) + continue + + result.new_files.append(name) + if dry_run: + continue + + logger.info("Ingesting chemistry workbook from Drive: %s (%s)", name, file_id) + try: + data = download_drive_file(file_id, service=drive_service) + upload = bulk_upload_chemistry(data) + except Exception as exc: # network / parse / DB failure for this file + logger.exception("Failed to ingest Drive file %s (%s)", name, file_id) + record = { + "name": name, + "file_id": file_id, + "status": "failed", + "error": str(exc), + } + manifest[file_id] = { + "name": name, + "md5": md5, + "modified_time": meta.get("modifiedTime"), + "status": "failed", + "error": str(exc), + "ingested_at": _now_iso(), + } + save_manifest(manifest, bucket) + result.failed.append(record) + continue + + summary = upload.payload.get("summary", {}) + status = "success" if upload.exit_code == 0 else "failed" + manifest[file_id] = { + "name": name, + "md5": md5, + "modified_time": meta.get("modifiedTime"), + "status": status, + "rows_imported": summary.get("total_rows_imported", 0), + "validation_errors_or_warnings": summary.get( + "validation_errors_or_warnings", 0 + ), + "ingested_at": _now_iso(), + } + # Persist after every file so an interrupted run keeps its progress. + save_manifest(manifest, bucket) + + record = { + "name": name, + "file_id": file_id, + "status": status, + "rows_imported": summary.get("total_rows_imported", 0), + "payload": upload.payload, + } + if status == "success": + result.ingested.append(record) + else: + result.failed.append(record) + + return result + + +# ============= EOF ============================================= diff --git a/services/chemistry_lims.py b/services/chemistry_lims.py new file mode 100644 index 000000000..25bb6f553 --- /dev/null +++ b/services/chemistry_lims.py @@ -0,0 +1,634 @@ +# =============================================================================== +# 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. +# =============================================================================== +""" +Ingest a laboratory LIMS chemistry workbook into the legacy NMA chemistry +tables (NMA_MajorChemistry / NMA_MinorTraceChemistry). + +Ported from the AMPAPI ``chemfile.py`` ingestion script. The AMPAPI original +read an ``.xls`` LIMS export with ``xlrd`` and inserted rows into the SQL Server +``MajorChemistry`` / ``MinorandTraceChemistry`` tables keyed off the +``Chemistry SampleInfo`` table. This adaptation: + +* reads an ``.xlsx`` workbook with ``openpyxl``, +* maps each LIMS ``Param`` to an analyte code + target table via ``FMapper``, +* resolves each ``SamplePointID`` (the base well PointID) to a ``Thing`` by + name, +* appends each distinct lab sample (``WCLab_ID``) as a new + ``NMA_Chemistry_SampleInfo`` row whose ``nma_sample_point_id`` is the base + PointID with the next letter incrementor appended (``A``, ``B``, ... ``Z``, + ``AA``, ...), skipping a lab sample already recorded for that well. + +The public entrypoint is :func:`bulk_upload_chemistry`. +""" + +from __future__ import annotations + +import io +import json +import re +import uuid +from dataclasses import dataclass +from datetime import date, datetime +from itertools import groupby +from pathlib import Path +from typing import Any, BinaryIO + +from openpyxl import load_workbook +from sqlalchemy import select +from sqlalchemy.orm import Session + +from db import ( + NMA_Chemistry_SampleInfo, + NMA_MajorChemistry, + NMA_MinorTraceChemistry, + Thing, +) +from db.engine import session_ctx + +# --- analyte mapping (ported verbatim from AMPAPI chemfile.py) ----------------- + +MINOR = "MinorandTraceChemistry" +MAJOR = "MajorChemistry" +EPM = "epm" +MGL = "mg/L" +PDIFF = "%Diff" +PH = "pH" +COND = "µS/cm" + +ANALYSES_AGENCY = "NMBGMR" + + +class AnalyteField: + def __init__(self, xlsfield, dbanalyte, table, units=None, method=None): + self.xlsfield = xlsfield + self.dbanalyte = dbanalyte + self.table = table + self.units = units + self.method = method + + +class FMapper: + def __init__(self): + self._map = [ + AnalyteField("alkalinity as caco3", "ALK", MAJOR, method="As CaCO3"), + AnalyteField("aluminum", "Al", MINOR), + AnalyteField("anions total", "TAn", MAJOR, EPM), + AnalyteField("antimony 121", "Sb", MINOR), + AnalyteField("antimony 123", "Sb", MINOR), + AnalyteField("antimony", "Sb", MINOR), + AnalyteField("arsenic", "As", MINOR), + AnalyteField("barium", "Ba", MINOR), + AnalyteField("beryllium", "Be", MINOR), + AnalyteField( + "bicarbonate (hco3)", "HCO3", MAJOR, method="Alkalinity as HC03" + ), + AnalyteField("boron 11", "B", MINOR), + AnalyteField("boron", "B", MINOR), + AnalyteField("bromide", "Br", MINOR), + AnalyteField("cadmium 111", "Cd", MINOR), + AnalyteField("cadmium", "Cd", MINOR), + AnalyteField("calcium", "Ca", MAJOR), + AnalyteField("carbonate (co3)", "CO3", MAJOR), + AnalyteField("cations total", "TCat", MAJOR, EPM), + AnalyteField("chloride", "Cl", MAJOR), + AnalyteField("chromium", "Cr", MINOR), + AnalyteField("cobalt", "Co", MINOR), + AnalyteField("copper 65", "Cu", MINOR), + AnalyteField("copper", "Cu", MINOR), + AnalyteField("fluoride", "F", MINOR), + AnalyteField("hardness", "HRD", MAJOR, MGL, method="As CaCO3"), + AnalyteField("iron", "Fe", MINOR), + AnalyteField("lead", "Pb", MINOR), + AnalyteField("lithium", "Li", MINOR), + AnalyteField("magnesium", "Mg", MAJOR), + AnalyteField("manganese", "Mn", MINOR), + AnalyteField("mercury", "Hg", MINOR), + AnalyteField("molybdenum 95", "Mo", MINOR), + AnalyteField("molybdenum", "Mo", MINOR), + AnalyteField("nickel", "Ni", MINOR), + AnalyteField("nitrate", "NO3", MINOR), + AnalyteField("nitrite", "NO2", MINOR), + AnalyteField("phosphate", "PO4", MINOR), + AnalyteField("percent difference", "IONBAL", MAJOR, PDIFF), + AnalyteField("potassium", "K", MAJOR), + AnalyteField("selenium", "Se", MINOR), + AnalyteField("siliconDioxide", "SiO2", MINOR), + AnalyteField("sio2", "SiO2", MINOR), + AnalyteField("silicon", "Si", MINOR), + AnalyteField("silver 107", "Ag", MINOR), + AnalyteField("silver", "Ag", MINOR), + AnalyteField("sodium", "Na", MAJOR), + AnalyteField("specific conductance", "CONDLAB", MAJOR, COND), + AnalyteField("strontium", "Sr", MINOR), + AnalyteField("sulfate", "SO4", MAJOR), + AnalyteField("tds calc", "TDS", MAJOR, method="Calculation"), + AnalyteField("thallium", "Tl", MINOR), + AnalyteField("thorium", "Th", MINOR), + AnalyteField("tin", "Sn", MINOR), + AnalyteField("titanium", "Ti", MINOR), + AnalyteField("uranium", "U", MINOR), + AnalyteField("vanadium", "V", MINOR), + AnalyteField("zinc 66", "Zn", MINOR), + AnalyteField("zinc", "Zn", MINOR), + AnalyteField("pH", "pHL", MAJOR, PH), + AnalyteField("ortho phosphate", "PO4", MINOR), + ] + + def values(self): + return self._map + + def get(self, key, attr="xlsfield"): + if key is None: + return None + for p in self._map: + value = getattr(p, attr) + if not isinstance(value, (list, tuple)): + value = (value,) + for vi in value: + if str(vi).lower() == str(key).lower(): + return p + return None + + +FM = FMapper() + +# Target ORM model per FMapper table bucket. +_TABLE_MODEL = {MAJOR: NMA_MajorChemistry, MINOR: NMA_MinorTraceChemistry} + + +class ChemistryMappingError(Exception): + """A LIMS row could not be normalized into an analyte measurement.""" + + +@dataclass +class ChemistryUploadResult: + exit_code: int + stdout: str + stderr: str + payload: dict[str, Any] + + +# --- workbook parsing ---------------------------------------------------------- + +# Columns the LIMS export is expected to carry. Extra columns are ignored; +# missing columns simply read back as ``None``. +LIMS_COLUMNS = ( + "Param", + "Results_Units", + "Dilution", + "AnalysisTime", + "SampleNumber", + "CustomerSampleNumber", + "SamplePointID", + "Method", + "Test", + "ReportedND", + "LowerLimit", + "SampleDate", +) + + +def read_lims_xlsx( + source: Path | str | bytes | BinaryIO, sheet_index: int = 0 +) -> list[dict]: + """Read a LIMS ``.xlsx`` workbook into a list of header->value dicts.""" + if isinstance(source, (bytes, bytearray)): + handle: Any = io.BytesIO(source) + elif isinstance(source, (str, Path)): + handle = source + else: + handle = source + + wb = load_workbook(filename=handle, read_only=True, data_only=True) + try: + sheet = wb.worksheets[sheet_index] + rows = sheet.iter_rows(values_only=True) + try: + header = [str(h).strip() if h is not None else "" for h in next(rows)] + except StopIteration: + return [] + records = [] + for row in rows: + if all(v is None for v in row): + continue + records.append(dict(zip(header, row))) + return records + finally: + wb.close() + + +# --- record normalization ------------------------------------------------------ + + +def _get(record: dict, key: str) -> Any: + value = record.get(key) + if isinstance(value, str): + value = value.strip() + if value == "": + return None + return value + + +def _to_float(value: Any) -> float | None: + if value is None or value == "": + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + +def _to_datetime(value: Any) -> datetime | None: + if value is None or value == "": + return None + if isinstance(value, datetime): + return value + if isinstance(value, date): + return datetime(value.year, value.month, value.day) + for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d", "%m/%d/%Y", "%m/%d/%Y %H:%M:%S"): + try: + return datetime.strptime(str(value), fmt) + except ValueError: + continue + return None + + +def prep_record(record: dict) -> dict: + """Normalize one raw LIMS row into an analyte-measurement dict. + + Raises :class:`ChemistryMappingError` when the row cannot be mapped. + """ + param = _get(record, "Param") + pm = FM.get(param) + if pm is None: + raise ChemistryMappingError(f"Unmapped analyte Param={param!r}") + + pointid = _get(record, "SamplePointID") or _get(record, "CustomerSampleNumber") + if not pointid: + raise ChemistryMappingError("Missing SamplePointID") + + units = pm.units or _get(record, "Results_Units") + + reported = _get(record, "ReportedND") + if reported is not None and str(reported).upper() == "ND": + lower = _to_float(_get(record, "LowerLimit")) or 0.0 + dilution = _to_float(_get(record, "Dilution")) + dilution = dilution if dilution else 1.0 + sample_value = lower * dilution + symbol = "<" + else: + sample_value = _to_float(reported) + symbol = None + + analysis_method = _get(record, "Method") + if pm.method: + analysis_method = ( + f"{analysis_method}, {pm.method}" if analysis_method else pm.method + ) + + analysis_date = _to_datetime(_get(record, "AnalysisTime")) + sample_date = _to_datetime(_get(record, "SampleDate")) or analysis_date + wclab_id = _get(record, "SampleNumber") + + return { + "analyte": pm.dbanalyte, + "table": pm.table, + "units": str(units) if units is not None else None, + "symbol": symbol, + "sample_value": sample_value, + "analysis_method": str(analysis_method) if analysis_method else None, + "analysis_date": analysis_date, + "sample_date": sample_date, + "wclab_id": str(wclab_id) if wclab_id is not None else None, + "samplepointid": str(pointid), + "test": _get(record, "Test"), + } + + +def dedupe_records(records: list[dict]) -> list[dict]: + """Collapse duplicate (SamplePointID, WCLab_ID, Analyte) rows. + + Mirrors AMPAPI chemfile.dbprep_records: when the same analyte is reported + more than once for the *same lab sample*, keep the ``low bromide`` test for + Br and the ``EPA 200.7`` method for everything else. Falls back to the first + row when no preferred method is present. Keyed on ``WCLab_ID`` too so two + distinct lab samples for one well keep their own analyte values. + """ + + def keyf(r: dict) -> tuple[str, str, str]: + return (r["samplepointid"], r["wclab_id"] or "", r["analyte"]) + + out: list[dict] = [] + for (_pid, _wclab, analyte), group in groupby(sorted(records, key=keyf), key=keyf): + group = list(group) + if len(group) < 2: + out.extend(group) + continue + + if analyte == "Br": + picked = next( + (r for r in group if (r.get("test") or "").casefold() == "low bromide"), + None, + ) + else: + picked = next( + ( + r + for r in group + if (r.get("analysis_method") or "") + .casefold() + .startswith("epa 200.7") + ), + None, + ) + out.append(picked or group[0]) + return out + + +# --- persistence --------------------------------------------------------------- + + +_SUFFIX_RE_TEMPLATE = r"^{base}([A-Z]+)$" + + +def _resolve_thing_id(session: Session, pointid: str) -> int | None: + things = session.scalars(select(Thing).where(Thing.name == pointid)).all() + if not things: + return None + # Thing.name is not guaranteed unique; take the lowest id deterministically. + return min(t.id for t in things) + + +def _suffix_to_int(suffix: str) -> int: + """Bijective base-26: A->1, B->2, ..., Z->26, AA->27, AB->28, ...""" + n = 0 + for ch in suffix: + n = n * 26 + (ord(ch) - ord("A") + 1) + return n + + +def _int_to_suffix(n: int) -> str: + """Inverse of :func:`_suffix_to_int` (``n`` >= 1).""" + letters: list[str] = [] + while n > 0: + n, rem = divmod(n - 1, 26) + letters.append(chr(ord("A") + rem)) + return "".join(reversed(letters)) + + +def _existing_suffix_ints(session: Session, thing_id: int, base: str) -> set[int]: + """Suffix numbers already used for ``base`` under this Thing. + + Chemistry sample points are the well PointID (``base``) with an appended + letter incrementor (``A``, ``B``, ... ``Z``, ``AA``, ...). Returns the set + of used incrementors, as bijective-base-26 integers, so the next one can be + computed. + """ + values = session.scalars( + select(NMA_Chemistry_SampleInfo.nma_sample_point_id).where( + NMA_Chemistry_SampleInfo.thing_id == thing_id + ) + ).all() + pattern = re.compile(_SUFFIX_RE_TEMPLATE.format(base=re.escape(base))) + used: set[int] = set() + for value in values: + if not value: + continue + match = pattern.match(value) + if match: + used.add(_suffix_to_int(match.group(1))) + return used + + +def _sample_exists_for_wclab( + session: Session, thing_id: int, wclab_id: str | None +) -> bool: + """True if this lab sample (WCLab_ID) is already recorded for the Thing.""" + if wclab_id is None: + return False + return ( + session.scalars( + select(NMA_Chemistry_SampleInfo.id).where( + NMA_Chemistry_SampleInfo.thing_id == thing_id, + NMA_Chemistry_SampleInfo.nma_wclab_id == wclab_id, + ) + ).first() + is not None + ) + + +def _build_measurement( + model, chemistry_sample_info_id: int, rec: dict, sample_point_id: str +): + analysis_date = rec["analysis_date"] + if model is NMA_MinorTraceChemistry and isinstance(analysis_date, datetime): + # NMA_MinorTraceChemistry.analysis_date is a DATE column. + analysis_date = analysis_date.date() + return model( + chemistry_sample_info_id=chemistry_sample_info_id, + nma_global_id=uuid.uuid4(), + nma_sample_point_id=sample_point_id, + nma_wclab_id=rec["wclab_id"], + analyte=rec["analyte"], + symbol=rec["symbol"], + sample_value=rec["sample_value"], + units=rec["units"], + analysis_method=rec["analysis_method"], + analysis_date=analysis_date, + analyses_agency=ANALYSES_AGENCY, + ) + + +def bulk_upload_chemistry( + source: Path | str | bytes, *, pretty_json: bool = False +) -> ChemistryUploadResult: + """Ingest a LIMS ``.xlsx`` workbook into the NMA chemistry tables. + + ``source`` may be a filesystem path or the raw ``.xlsx`` bytes (e.g. a file + downloaded from Google Drive). + + The workbook's ``SamplePointID`` is the base well PointID. Each distinct lab + sample (``WCLab_ID`` / SampleNumber) for a well becomes a new + ``NMA_Chemistry_SampleInfo`` row whose ``nma_sample_point_id`` is the base + with the next letter incrementor appended (``A``, ``B``, ... ``Z``, ``AA``, + ...). A lab sample already recorded for the well (same ``WCLab_ID``) is + skipped, so re-running is idempotent. + + A data-quality problem (a row that fails to map, or a ``SamplePointID`` with + no matching Thing) aborts the whole file -- nothing is written. + """ + if isinstance(source, str): + source = Path(source) + + try: + raw_records = read_lims_xlsx(source) + except Exception as exc: # openpyxl raises a variety of parse errors + return _result( + processed=0, + imported=0, + validation_errors=[f"Could not read workbook: {exc}"], + skipped_duplicates=[], + created=[], + pretty_json=pretty_json, + ) + + processed = len(raw_records) + validation_errors: list[str] = [] + prepped: list[dict] = [] + for offset, raw in enumerate(raw_records): + # +2: worksheet row 1 is the header, enumerate is 0-based. + row_number = offset + 2 + try: + prepped.append(prep_record(raw)) + except ChemistryMappingError as exc: + validation_errors.append(f"Row {row_number}: {exc}") + + prepped = dedupe_records(prepped) + + with session_ctx() as session: + # Resolve every distinct (base) sample point to a Thing up front. + base_pointids = sorted({r["samplepointid"] for r in prepped}) + thing_ids: dict[str, int | None] = { + pid: _resolve_thing_id(session, pid) for pid in base_pointids + } + for pid in base_pointids: + if thing_ids[pid] is None: + validation_errors.append( + f"SamplePointID {pid}: no matching Thing (well) found" + ) + + # Abort the whole file on any data-quality problem before writing. + if validation_errors: + return _result( + processed=processed, + imported=0, + validation_errors=validation_errors, + skipped_duplicates=[], + created=[], + pretty_json=pretty_json, + ) + + # One sample = one lab sample (WCLab_ID) for a well. + def bucket_key(r: dict) -> tuple[str, str | None]: + return (r["samplepointid"], r["wclab_id"]) + + buckets: dict[tuple[str, str | None], list[dict]] = {} + for rec in sorted( + prepped, key=lambda r: (r["samplepointid"], r["wclab_id"] or "") + ): + buckets.setdefault(bucket_key(rec), []).append(rec) + + # Per-Thing set of used suffix incrementors, seeded from the DB and + # extended as we assign new ones within this run. + used_suffixes: dict[int, set[int]] = {} + skipped_duplicates: list[dict] = [] + created: list[dict] = [] + imported = 0 + + for (base, wclab_id), recs in buckets.items(): + thing_id = thing_ids[base] + + # Already ingested this lab sample -> skip (idempotent), keep going. + if _sample_exists_for_wclab(session, thing_id, wclab_id): + skipped_duplicates.append({"pointid": base, "wclab_id": wclab_id}) + continue + + if thing_id not in used_suffixes: + used_suffixes[thing_id] = _existing_suffix_ints(session, thing_id, base) + next_int = ( + max(used_suffixes[thing_id]) + 1 if used_suffixes[thing_id] else 1 + ) + used_suffixes[thing_id].add(next_int) + sample_point_id = f"{base}{_int_to_suffix(next_int)}" + + collection_date = next( + (r["sample_date"] for r in recs if r["sample_date"]), None + ) + info = NMA_Chemistry_SampleInfo( + thing_id=thing_id, + nma_sample_pt_id=uuid.uuid4(), + nma_sample_point_id=sample_point_id, + nma_wclab_id=wclab_id, + analyses_agency=ANALYSES_AGENCY, + collection_date=collection_date, + ) + session.add(info) + session.flush() # assign info.id for the FK below + + for rec in recs: + model = _TABLE_MODEL[rec["table"]] + session.add(_build_measurement(model, info.id, rec, sample_point_id)) + imported += 1 + created.append( + { + "sample_point_id": sample_point_id, + "wclab_id": wclab_id, + "rows": len(recs), + } + ) + + session.commit() + + return _result( + processed=processed, + imported=imported, + validation_errors=validation_errors, + skipped_duplicates=skipped_duplicates, + created=created, + pretty_json=pretty_json, + ) + + +def _result( + *, + processed: int, + imported: int, + validation_errors: list[str], + skipped_duplicates: list[dict], + created: list[dict], + pretty_json: bool, +) -> ChemistryUploadResult: + rows_with_issues = len(validation_errors) + len(skipped_duplicates) + payload = { + "summary": { + "total_rows_processed": processed, + "total_rows_imported": imported, + "validation_errors_or_warnings": rows_with_issues, + "samples_created": len(created), + "samples_skipped": len(skipped_duplicates), + }, + "validation_errors": validation_errors, + "skipped_duplicates": skipped_duplicates, + "created_samples": created, + } + stdout = json.dumps(payload, indent=2 if pretty_json else None) + stderr_parts: list[str] = [] + if validation_errors: + stderr_parts.append("\n".join(validation_errors)) + if skipped_duplicates: + dupes = ", ".join( + f"{d['pointid']} (WCLab_ID {d['wclab_id']})" for d in skipped_duplicates + ) + stderr_parts.append(f"Skipped already-ingested lab sample(s): {dupes}") + stderr = "\n".join(stderr_parts) + # Only a data-quality abort is a failure; skipped duplicates are idempotent. + exit_code = 1 if validation_errors else 0 + return ChemistryUploadResult( + exit_code=exit_code, stdout=stdout, stderr=stderr, payload=payload + ) + + +# ============= EOF ============================================= diff --git a/tests/test_chemistry_drive.py b/tests/test_chemistry_drive.py new file mode 100644 index 000000000..2ad2769fb --- /dev/null +++ b/tests/test_chemistry_drive.py @@ -0,0 +1,219 @@ +# =============================================================================== +# 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. +# =============================================================================== +"""Tests for the Drive-polling chemistry sync (services/chemistry_drive.py).""" + +import io + +import pytest +from openpyxl import Workbook +from sqlalchemy import delete + +from db.engine import session_ctx +from db.nma_legacy import NMA_Chemistry_SampleInfo +from services import chemistry_drive +from services.chemistry_drive import ( + ChemistryDriveConfigError, + load_manifest, + save_manifest, + sync_and_ingest, +) + +LIMS_HEADER = [ + "Param", + "Results_Units", + "Dilution", + "AnalysisTime", + "SampleNumber", + "CustomerSampleNumber", + "SamplePointID", + "Method", + "Test", + "ReportedND", + "LowerLimit", + "SampleDate", +] + + +def _workbook_bytes(param="calcium", value="12.5", pointid="Test Well"): + wb = Workbook() + ws = wb.active + ws.append(LIMS_HEADER) + ws.append( + [ + param, + "mg/L", + 1, + "2024-06-15", + "LAB-1", + pointid, + pointid, + "EPA 200.7", + "Major", + value, + 0.01, + "2024-06-01", + ] + ) + buffer = io.BytesIO() + wb.save(buffer) + return buffer.getvalue() + + +class FakeBlob: + def __init__(self, store, name): + self._store = store + self._name = name + + def exists(self): + return self._name in self._store + + def download_as_text(self): + return self._store[self._name] + + def upload_from_string(self, data, content_type=None): + self._store[self._name] = data + + +class FakeBucket: + def __init__(self): + self.store = {} + + def blob(self, name): + return FakeBlob(self.store, name) + + +@pytest.fixture() +def fake_bucket(): + return FakeBucket() + + +@pytest.fixture() +def _cleanup_chemistry(): + yield + with session_ctx() as session: + session.execute( + delete(NMA_Chemistry_SampleInfo).where( + NMA_Chemistry_SampleInfo.nma_wclab_id.like("LAB-%") + ) + ) + session.commit() + + +def _stub_drive(monkeypatch, files: list[dict], contents: dict[str, bytes]): + monkeypatch.setattr( + chemistry_drive, "list_drive_xlsx", lambda folder_id, service=None: files + ) + + def _download(file_id, service=None): + return contents[file_id] + + monkeypatch.setattr(chemistry_drive, "download_drive_file", _download) + + +# ------------------------- manifest tests ------------------------------------ + + +def test_manifest_roundtrip(fake_bucket): + assert load_manifest(fake_bucket) == {} + save_manifest({"F1": {"status": "success"}}, fake_bucket) + assert load_manifest(fake_bucket) == {"F1": {"status": "success"}} + + +def test_missing_folder_raises(monkeypatch): + monkeypatch.delenv("CHEMISTRY_DRIVE_FOLDER_ID", raising=False) + with pytest.raises(ChemistryDriveConfigError): + sync_and_ingest(folder_id=None) + + +# ------------------------- sync tests ---------------------------------------- + + +def test_sync_ingests_new_file_and_records_manifest( + monkeypatch, fake_bucket, water_well_thing, _cleanup_chemistry +): + files = [ + {"id": "F1", "name": "batch1.xlsx", "md5Checksum": "abc", "modifiedTime": "t0"} + ] + _stub_drive(monkeypatch, files, {"F1": _workbook_bytes()}) + + result = sync_and_ingest(folder_id="folder", bucket=fake_bucket) + + assert result.exit_code == 0 + assert len(result.ingested) == 1 + assert result.ingested[0]["rows_imported"] == 1 + manifest = load_manifest(fake_bucket) + assert manifest["F1"]["status"] == "success" + assert manifest["F1"]["md5"] == "abc" + + +def test_sync_skips_already_ingested_file( + monkeypatch, fake_bucket, water_well_thing, _cleanup_chemistry +): + files = [ + {"id": "F1", "name": "batch1.xlsx", "md5Checksum": "abc", "modifiedTime": "t0"} + ] + _stub_drive(monkeypatch, files, {"F1": _workbook_bytes()}) + + first = sync_and_ingest(folder_id="folder", bucket=fake_bucket) + assert len(first.ingested) == 1 + + second = sync_and_ingest(folder_id="folder", bucket=fake_bucket) + assert second.ingested == [] + assert second.skipped == ["batch1.xlsx"] + + +def test_dry_run_does_not_download_or_write_manifest( + monkeypatch, fake_bucket, water_well_thing, _cleanup_chemistry +): + files = [ + {"id": "F1", "name": "batch1.xlsx", "md5Checksum": "abc", "modifiedTime": "t0"} + ] + monkeypatch.setattr( + chemistry_drive, "list_drive_xlsx", lambda folder_id, service=None: files + ) + + def _boom(file_id, service=None): + raise AssertionError("download must not be called during a dry run") + + monkeypatch.setattr(chemistry_drive, "download_drive_file", _boom) + + result = sync_and_ingest(folder_id="folder", bucket=fake_bucket, dry_run=True) + + assert result.dry_run is True + assert result.new_files == ["batch1.xlsx"] + assert result.ingested == [] + assert load_manifest(fake_bucket) == {} + + +def test_sync_marks_failed_when_ingestion_aborts( + monkeypatch, fake_bucket, water_well_thing, _cleanup_chemistry +): + # A workbook whose SamplePointID has no matching Thing aborts (validation + # error) -> the file is recorded as failed. + _stub_drive( + monkeypatch, + [{"id": "F1", "name": "bad.xlsx", "md5Checksum": "abc", "modifiedTime": "t0"}], + {"F1": _workbook_bytes(pointid="NO-SUCH-WELL")}, + ) + result = sync_and_ingest(folder_id="folder", bucket=fake_bucket) + + assert result.exit_code == 1 + assert len(result.failed) == 1 + assert result.failed[0]["name"] == "bad.xlsx" + assert load_manifest(fake_bucket)["F1"]["status"] == "failed" + + +# ============= EOF ============================================= diff --git a/tests/test_chemistry_lims.py b/tests/test_chemistry_lims.py new file mode 100644 index 000000000..c95d9575f --- /dev/null +++ b/tests/test_chemistry_lims.py @@ -0,0 +1,287 @@ +# =============================================================================== +# 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. +# =============================================================================== +"""Tests for the LIMS chemistry ingestion service (services/chemistry_lims.py).""" + +from pathlib import Path + +import pytest +from openpyxl import Workbook +from sqlalchemy import delete, select + +from db.engine import session_ctx +from db.nma_legacy import ( + NMA_Chemistry_SampleInfo, + NMA_MajorChemistry, + NMA_MinorTraceChemistry, +) +from services.chemistry_lims import ( + _int_to_suffix, + _suffix_to_int, + bulk_upload_chemistry, + dedupe_records, + prep_record, +) + +LIMS_HEADER = [ + "Param", + "Results_Units", + "Dilution", + "AnalysisTime", + "SampleNumber", + "CustomerSampleNumber", + "SamplePointID", + "Method", + "Test", + "ReportedND", + "LowerLimit", + "SampleDate", +] + + +def _write_workbook(path: Path, rows: list[dict]) -> Path: + wb = Workbook() + ws = wb.active + ws.append(LIMS_HEADER) + for row in rows: + ws.append([row.get(col) for col in LIMS_HEADER]) + wb.save(path) + return path + + +def _lims_row(param, value, *, pointid="Test Well", method="EPA 200.7", **overrides): + row = { + "Param": param, + "Results_Units": "mg/L", + "Dilution": 1, + "AnalysisTime": "2024-06-15", + "SampleNumber": "LAB-1", + "CustomerSampleNumber": pointid, + "SamplePointID": pointid, + "Method": method, + "Test": "Trace Metals", + "ReportedND": value, + "LowerLimit": 0.01, + "SampleDate": "2024-06-01", + } + row.update(overrides) + return row + + +@pytest.fixture() +def _cleanup_chemistry(): + """Remove any sample-info (and cascaded analytes) created during a test.""" + yield + with session_ctx() as session: + session.execute( + delete(NMA_Chemistry_SampleInfo).where( + NMA_Chemistry_SampleInfo.nma_wclab_id.like("LAB-%") + ) + ) + session.commit() + + +# ------------------------- pure-function tests ------------------------------- + + +def test_prep_record_maps_analyte_and_table(): + rec = prep_record(_lims_row("calcium", "12.5")) + assert rec["analyte"] == "Ca" + assert rec["table"] == "MajorChemistry" + assert rec["sample_value"] == 12.5 + assert rec["symbol"] is None + + +def test_prep_record_non_detect_uses_lower_limit_times_dilution(): + rec = prep_record(_lims_row("lead", "ND", Dilution=2, LowerLimit=0.01)) + assert rec["analyte"] == "Pb" + assert rec["table"] == "MinorandTraceChemistry" + assert rec["symbol"] == "<" + assert rec["sample_value"] == pytest.approx(0.02) + + +def test_prep_record_unmapped_analyte_raises(): + from services.chemistry_lims import ChemistryMappingError + + with pytest.raises(ChemistryMappingError): + prep_record(_lims_row("unobtanium", "1.0")) + + +def test_dedupe_prefers_epa_200_7(): + rows = [ + prep_record(_lims_row("calcium", "10", method="EPA 6010")), + prep_record(_lims_row("calcium", "11", method="EPA 200.7")), + ] + deduped = dedupe_records(rows) + assert len(deduped) == 1 + assert deduped[0]["sample_value"] == 11.0 + + +@pytest.mark.parametrize( + "suffix,number", + [("A", 1), ("B", 2), ("Z", 26), ("AA", 27), ("AB", 28), ("AZ", 52), ("BA", 53)], +) +def test_suffix_bijective_base26_roundtrip(suffix, number): + assert _suffix_to_int(suffix) == number + assert _int_to_suffix(number) == suffix + + +# ------------------------- ingestion tests ----------------------------------- + + +def test_bulk_upload_inserts_major_and_minor( + tmp_path, water_well_thing, _cleanup_chemistry +): + path = _write_workbook( + tmp_path / "lims.xlsx", + [_lims_row("calcium", "12.5"), _lims_row("arsenic", "0.3")], + ) + + result = bulk_upload_chemistry(path) + + assert result.exit_code == 0, result.stderr + assert result.payload["summary"]["total_rows_imported"] == 2 + + with session_ctx() as session: + info = session.scalars( + select(NMA_Chemistry_SampleInfo).where( + NMA_Chemistry_SampleInfo.thing_id == water_well_thing.id + ) + ).one() + major = session.scalars( + select(NMA_MajorChemistry).where( + NMA_MajorChemistry.chemistry_sample_info_id == info.id + ) + ).all() + minor = session.scalars( + select(NMA_MinorTraceChemistry).where( + NMA_MinorTraceChemistry.chemistry_sample_info_id == info.id + ) + ).all() + + assert {m.analyte for m in major} == {"Ca"} + assert {m.analyte for m in minor} == {"As"} + # First sample for the well -> base PointID + "A". + assert info.nma_sample_point_id == "Test WellA" + assert {m.nma_sample_point_id for m in major} == {"Test WellA"} + + +def test_bulk_upload_skips_duplicate_lab_sample( + tmp_path, water_well_thing, _cleanup_chemistry +): + # Same WCLab_ID (SampleNumber) uploaded twice -> second run is idempotent. + rows = [_lims_row("calcium", "12.5", SampleNumber="LAB-1")] + _write_workbook(tmp_path / "first.xlsx", rows) + first = bulk_upload_chemistry(tmp_path / "first.xlsx") + assert first.exit_code == 0, first.stderr + assert first.payload["summary"]["samples_created"] == 1 + + _write_workbook(tmp_path / "second.xlsx", rows) + second = bulk_upload_chemistry(tmp_path / "second.xlsx") + + # Idempotent: no failure, nothing imported, reported as skipped. + assert second.exit_code == 0 + assert second.payload["summary"]["total_rows_imported"] == 0 + assert second.payload["summary"]["samples_skipped"] == 1 + assert second.payload["skipped_duplicates"][0]["wclab_id"] == "LAB-1" + + with session_ctx() as session: + rows_ca = session.scalars( + select(NMA_MajorChemistry).where(NMA_MajorChemistry.analyte == "Ca") + ).all() + assert len(rows_ca) == 1 # not duplicated + + +def test_bulk_upload_appends_new_lab_sample_with_next_suffix( + tmp_path, water_well_thing, _cleanup_chemistry +): + # A different WCLab_ID for the same well -> a new lettered sample point. + _write_workbook( + tmp_path / "first.xlsx", [_lims_row("calcium", "12.5", SampleNumber="LAB-1")] + ) + first = bulk_upload_chemistry(tmp_path / "first.xlsx") + assert first.exit_code == 0, first.stderr + assert first.payload["created_samples"][0]["sample_point_id"] == "Test WellA" + + _write_workbook( + tmp_path / "second.xlsx", [_lims_row("calcium", "9.9", SampleNumber="LAB-2")] + ) + second = bulk_upload_chemistry(tmp_path / "second.xlsx") + assert second.exit_code == 0, second.stderr + assert second.payload["created_samples"][0]["sample_point_id"] == "Test WellB" + + with session_ctx() as session: + infos = session.scalars( + select(NMA_Chemistry_SampleInfo).where( + NMA_Chemistry_SampleInfo.thing_id == water_well_thing.id + ) + ).all() + assert {i.nma_sample_point_id for i in infos} == {"Test WellA", "Test WellB"} + + +def test_bulk_upload_two_lab_samples_in_one_file_get_a_and_b( + tmp_path, water_well_thing, _cleanup_chemistry +): + # Two distinct lab samples in a single workbook -> A and B in one run. + _write_workbook( + tmp_path / "lims.xlsx", + [ + _lims_row("calcium", "12.5", SampleNumber="LAB-1"), + _lims_row("calcium", "9.9", SampleNumber="LAB-2"), + ], + ) + result = bulk_upload_chemistry(tmp_path / "lims.xlsx") + assert result.exit_code == 0, result.stderr + assert result.payload["summary"]["samples_created"] == 2 + + with session_ctx() as session: + infos = session.scalars( + select(NMA_Chemistry_SampleInfo).where( + NMA_Chemistry_SampleInfo.thing_id == water_well_thing.id + ) + ).all() + assert {i.nma_sample_point_id for i in infos} == {"Test WellA", "Test WellB"} + + +def test_bulk_upload_reports_missing_thing(tmp_path, _cleanup_chemistry): + path = _write_workbook( + tmp_path / "lims.xlsx", + [_lims_row("calcium", "12.5", pointid="NO-SUCH-WELL")], + ) + + result = bulk_upload_chemistry(path) + + assert result.exit_code == 1 + assert result.payload["summary"]["total_rows_imported"] == 0 + assert any("no matching Thing" in e for e in result.payload["validation_errors"]) + + +def test_bulk_upload_reports_unmapped_analyte( + tmp_path, water_well_thing, _cleanup_chemistry +): + path = _write_workbook( + tmp_path / "lims.xlsx", + [_lims_row("calcium", "12.5"), _lims_row("unobtanium", "9.9")], + ) + + result = bulk_upload_chemistry(path) + + # Unmapped analyte is a validation error -> whole file aborts, nothing imported. + assert result.exit_code == 1 + assert result.payload["summary"]["total_rows_imported"] == 0 + assert any("Unmapped analyte" in e for e in result.payload["validation_errors"]) + + +# ============= EOF ============================================= diff --git a/uv.lock b/uv.lock index 791d52f03..ac9968137 100644 --- a/uv.lock +++ b/uv.lock @@ -761,6 +761,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/de/15/545e2b6cf2e3be84bc1ed85613edd75b8aea69807a71c26f4ca6a9258e82/email_validator-2.3.0-py3-none-any.whl", hash = "sha256:80f13f623413e6b197ae73bb10bf4eb0908faf509ad8362c5edeb0be7fd450b4", size = 35604, upload-time = "2025-08-26T13:09:05.858Z" }, ] +[[package]] +name = "et-xmlfile" +version = "2.0.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d3/38/af70d7ab1ae9d4da450eeec1fa3918940a5fafb9055e934af8d6eb0c2313/et_xmlfile-2.0.0.tar.gz", hash = "sha256:dab3f4764309081ce75662649be815c4c9081e88f0837825f90fd28317d4da54", size = 17234, upload-time = "2024-10-25T17:25:40.039Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c1/8b/5fe2cc11fee489817272089c4203e679c63b570a5aaeb18d852ae3cbba6a/et_xmlfile-2.0.0-py3-none-any.whl", hash = "sha256:7a91720bc756843502c3b7504c77b8fe44217c85c537d85037f0f536151b2caa", size = 18059, upload-time = "2024-10-25T17:25:39.051Z" }, +] + [[package]] name = "faker" version = "37.12.0" @@ -945,6 +954,22 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/86/40/9bdbb60b03a332bd45acb8703da08bbc27d991d35286b62e42acc86d243a/google_api_core-2.31.0-py3-none-any.whl", hash = "sha256:ef79fb3784c71cbac89cbd03301ba0c8fb8ad2aa95d7f9204dd9628f7adf59ab", size = 173102, upload-time = "2026-06-03T14:51:26.729Z" }, ] +[[package]] +name = "google-api-python-client" +version = "2.184.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "google-api-core" }, + { name = "google-auth" }, + { name = "google-auth-httplib2" }, + { name = "httplib2" }, + { name = "uritemplate" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/7c/30/8b3a626ccf84ca43da62d77e2d40d70bedc6387951cc5104011cddce34e0/google_api_python_client-2.184.0.tar.gz", hash = "sha256:ef2a3330ad058cdfc8a558d199c051c3356f6ed012436c3ad3d08b67891b039f", size = 13694120, upload-time = "2025-10-01T21:13:48.961Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f4/38/d25ae1565103a545cf18207a5dec09a6d39ad88e5b0399a2430e9edb0550/google_api_python_client-2.184.0-py3-none-any.whl", hash = "sha256:15a18d02f42de99416921c77be235d12ead474e474a1abc348b01a2b92633fa4", size = 14260480, upload-time = "2025-10-01T21:13:46.037Z" }, +] + [[package]] name = "google-auth" version = "2.55.1" @@ -958,6 +983,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e8/1d/f6d3ca1ad0725f2e08a1c6915640748a52de2e66596160a4d53b010cccf0/google_auth-2.55.1-py3-none-any.whl", hash = "sha256:eada68dfd52b3b81191827601e2a0c3fa12540c818534b630ddc5355769c3995", size = 252349, upload-time = "2026-06-25T23:38:52.946Z" }, ] +[[package]] +name = "google-auth-httplib2" +version = "0.4.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "google-auth" }, + { name = "httplib2" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/1c/b3/f192c8bc7e41e0ebdbd95afcae4783417a34b6a6af62d22daf22c3fd38fc/google_auth_httplib2-0.4.0.tar.gz", hash = "sha256:d5b030a204b7a4b4d553ba9ca701b62481ee2b74419325580be70f7d85ffed35", size = 11161, upload-time = "2026-05-07T08:03:46.878Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/97/be/954c35a62b9e31de66b0a43c225c9b6bb9e0f98d6b1dc110a2308e3644f5/google_auth_httplib2-0.4.0-py3-none-any.whl", hash = "sha256:8e55cfafa3358cba85f6cad4a886138e88e158d71e7e5c9ee5936a5c1507fb91", size = 9529, upload-time = "2026-05-07T08:02:12.375Z" }, +] + [[package]] name = "google-cloud-core" version = "2.6.0" @@ -1121,6 +1159,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/7e/f5/f66802a942d491edb555dd61e3a9961140fd64c90bce1eafd741609d334d/httpcore-1.0.9-py3-none-any.whl", hash = "sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55", size = 78784, upload-time = "2025-04-24T22:06:20.566Z" }, ] +[[package]] +name = "httplib2" +version = "0.32.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pyparsing" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/84/f5/ccf58de92d61e3ad921119668f54ed36ca1d0cf5dcc5c1657dfb164fd78b/httplib2-0.32.0.tar.gz", hash = "sha256:48a0ef30a42db65d8f3399045e1d09ab0ba66e3b9efc360d07f80ea55d286025", size = 254283, upload-time = "2026-06-26T10:13:56.265Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/33/a0/550eec327e5f5c7b732531c489f5307efec41f047b0d703bd4ca1e5ad2db/httplib2-0.32.0-py3-none-any.whl", hash = "sha256:dc6705cacdf3fb0a2aba7629fa33c90fd93e30035db0c157325826be177e4816", size = 93148, upload-time = "2026-06-26T10:13:54.985Z" }, +] + [[package]] name = "httpx" version = "0.28.1" @@ -1582,6 +1632,10 @@ dependencies = [ ] [package.dev-dependencies] +cli = [ + { name = "google-api-python-client" }, + { name = "openpyxl" }, +] dev = [ { name = "behave" }, { name = "black" }, @@ -1697,6 +1751,10 @@ requires-dist = [ ] [package.metadata.requires-dev] +cli = [ + { name = "google-api-python-client", specifier = "==2.184.0" }, + { name = "openpyxl", specifier = "==3.1.5" }, +] dev = [ { name = "behave", specifier = ">=1.3.3" }, { name = "black", specifier = ">=26.5.1" }, @@ -1710,6 +1768,18 @@ dev = [ { name = "requests", specifier = ">=2.34.2" }, ] +[[package]] +name = "openpyxl" +version = "3.1.5" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "et-xmlfile" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/3d/f9/88d94a75de065ea32619465d2f77b29a0469500e99012523b91cc4141cd1/openpyxl-3.1.5.tar.gz", hash = "sha256:cf0e3cf56142039133628b5acffe8ef0c12bc902d2aadd3e0fe5878dc08d1050", size = 186464, upload-time = "2024-06-28T14:03:44.161Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c0/da/977ded879c29cbd04de313843e76868e6e13408a94ed6b987245dc7c8506/openpyxl-3.1.5-py2.py3-none-any.whl", hash = "sha256:5282c12b107bffeef825f4617dc029afaf41d0ea60823bbb665ef3079dc79de2", size = 250910, upload-time = "2024-06-28T14:03:41.161Z" }, +] + [[package]] name = "opentelemetry-api" version = "1.39.1" @@ -3034,6 +3104,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c2/14/e2a54fabd4f08cd7af1c07030603c3356b74da07f7cc056e600436edfa17/tzlocal-5.3.1-py3-none-any.whl", hash = "sha256:eb1a66c3ef5847adf7a834f1be0800581b683b5608e74f86ecbcef8ab91bb85d", size = 18026, upload-time = "2025-03-05T21:17:39.857Z" }, ] +[[package]] +name = "uritemplate" +version = "4.2.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/98/60/f174043244c5306c9988380d2cb10009f91563fc4b31293d27e17201af56/uritemplate-4.2.0.tar.gz", hash = "sha256:480c2ed180878955863323eea31b0ede668795de182617fef9c6ca09e6ec9d0e", size = 33267, upload-time = "2025-06-02T15:12:06.318Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a9/99/3ae339466c9183ea5b8ae87b34c0b897eda475d2aec2307cae60e5cd4f29/uritemplate-4.2.0-py3-none-any.whl", hash = "sha256:962201ba1c4edcab02e60f9a0d3821e82dfc5d2d6662a21abd533879bdb8a686", size = 11488, upload-time = "2025-06-02T15:12:03.405Z" }, +] + [[package]] name = "urllib3" version = "2.7.0" From 15040e9fe1f70bbfcd25197d7b38079f7db81019 Mon Sep 17 00:00:00 2001 From: jross Date: Wed, 8 Jul 2026 10:08:14 -0600 Subject: [PATCH 2/4] chore(ci): drop CD refresh-materialized-views step Materialized-view refresh is handled by the nightly pg_cron job, so the production/staging/testing deploy workflows no longer need to run `python -m cli.cli refresh-materialized-views`. Refs BDMS-1034 Co-Authored-By: Claude Opus 4.8 --- .github/workflows/CD_production.yml | 10 ---------- .github/workflows/CD_staging.yml | 10 ---------- .github/workflows/CD_testing.yml | 10 ---------- 3 files changed, 30 deletions(-) diff --git a/.github/workflows/CD_production.yml b/.github/workflows/CD_production.yml index 5a0acce62..215c808e1 100644 --- a/.github/workflows/CD_production.yml +++ b/.github/workflows/CD_production.yml @@ -99,16 +99,6 @@ jobs: run: | uv run --no-dev alembic upgrade head - - name: Refresh materialized views on production database - env: - DB_DRIVER: "cloudsql" - CLOUD_SQL_INSTANCE_NAME: "${{ secrets.CLOUD_SQL_INSTANCE_NAME }}" - CLOUD_SQL_DATABASE: "${{ vars.CLOUD_SQL_DATABASE }}" - CLOUD_SQL_USER: "${{ secrets.CLOUD_SQL_USER }}" - CLOUD_SQL_IAM_AUTH: true - run: | - uv run --no-dev python -m cli.cli refresh-materialized-views - - name: Ensure envsubst is available run: | if ! command -v envsubst >/dev/null 2>&1; then diff --git a/.github/workflows/CD_staging.yml b/.github/workflows/CD_staging.yml index e2fa929e3..d6f6b1f6d 100644 --- a/.github/workflows/CD_staging.yml +++ b/.github/workflows/CD_staging.yml @@ -59,16 +59,6 @@ jobs: run: | uv run --no-dev alembic upgrade head - - name: Refresh materialized views on staging database - env: - DB_DRIVER: "cloudsql" - CLOUD_SQL_INSTANCE_NAME: "${{ secrets.CLOUD_SQL_INSTANCE_NAME }}" - CLOUD_SQL_DATABASE: "${{ vars.CLOUD_SQL_DATABASE }}" - CLOUD_SQL_USER: "${{ secrets.CLOUD_SQL_USER }}" - CLOUD_SQL_IAM_AUTH: true - run: | - uv run --no-dev python -m cli.cli refresh-materialized-views - - name: Ensure envsubst is available run: | if ! command -v envsubst >/dev/null 2>&1; then diff --git a/.github/workflows/CD_testing.yml b/.github/workflows/CD_testing.yml index 7004c5b60..0ec67f29c 100644 --- a/.github/workflows/CD_testing.yml +++ b/.github/workflows/CD_testing.yml @@ -59,16 +59,6 @@ jobs: run: | uv run --no-dev alembic upgrade head - - name: Refresh materialized views on staging database - env: - DB_DRIVER: "cloudsql" - CLOUD_SQL_INSTANCE_NAME: "${{ secrets.CLOUD_SQL_INSTANCE_NAME }}" - CLOUD_SQL_DATABASE: "${{ vars.CLOUD_SQL_DATABASE }}" - CLOUD_SQL_USER: "${{ secrets.CLOUD_SQL_USER }}" - CLOUD_SQL_IAM_AUTH: true - run: | - uv run --no-dev python -m cli.cli refresh-materialized-views - - name: Ensure envsubst is available run: | if ! command -v envsubst >/dev/null 2>&1; then From 986a93c5329a28ec1401b43fa51bd2e0be15a713 Mon Sep 17 00:00:00 2001 From: jross Date: Wed, 8 Jul 2026 11:49:55 -0600 Subject: [PATCH 3/4] refactor(cli): simplify chemistry CLI output and analyte mapping - CLI serializes its own JSON from the result payload; drop the pretty_json flag threaded through the service and the pre-baked `stdout` field on ChemistryUploadResult. - Remove the unused `--output json` option from `water-chemistry bulk-upload` and `sync-drive`; they are engineer-facing and have no scripted consumer (programmatic callers use the service functions directly). - Replace the FMapper/AnalyteField classes with a frozen `AnalyteMapping` dataclass, a flat `_ANALYTE_MAPPINGS` list, and a `lookup_analyte()` helper backed by a case-insensitive dict; drop the dead list/tuple and reverse-lookup branches. Co-Authored-By: Claude Opus 4.8 --- cli/cli.py | 30 +----- cli/service_adapter.py | 4 +- services/chemistry_lims.py | 214 ++++++++++++++++++------------------- 3 files changed, 110 insertions(+), 138 deletions(-) diff --git a/cli/cli.py b/cli/cli.py index 8a6432e4f..907437928 100644 --- a/cli/cli.py +++ b/cli/cli.py @@ -972,30 +972,21 @@ def water_chemistry_bulk_upload( readable=True, help="Path to LIMS .xlsx workbook containing chemistry results.", ), - output_format: OutputFormat | None = typer.Option( - None, - "--output", - help="Optional output format", - ), theme: ThemeMode = typer.Option( ThemeMode.auto, "--theme", help="Color theme: auto, light, dark." ), ): """ parse a LIMS chemistry workbook and load it into the NMA Major/Minor - chemistry tables. All-or-nothing: if any analyte already exists, or any row - fails to map, nothing is written. + chemistry tables. Each distinct lab sample (WCLab_ID) is appended as a new + lettered sample point; a lab sample already recorded for the well is + skipped. A row that fails to map or references an unknown well aborts the + whole file. """ from cli.service_adapter import chemistry_lims_xlsx colors = _palette(theme) - result = chemistry_lims_xlsx( - file_path, pretty_json=output_format == OutputFormat.json - ) - - if output_format == OutputFormat.json: - typer.echo(result.stdout) - raise typer.Exit(result.exit_code) + result = chemistry_lims_xlsx(file_path) payload = result.payload if isinstance(result.payload, dict) else {} summary = payload.get("summary", {}) @@ -1086,11 +1077,6 @@ def water_chemistry_sync_drive( "--dry-run", help="List new files without downloading, ingesting, or updating the manifest.", ), - output_format: OutputFormat | None = typer.Option( - None, - "--output", - help="Optional output format", - ), theme: ThemeMode = typer.Option( ThemeMode.auto, "--theme", help="Color theme: auto, light, dark." ), @@ -1101,8 +1087,6 @@ def water_chemistry_sync_drive( are ingested; a manifest of ingested files is kept in GCS so already-processed files are skipped. """ - import json as _json - from services.chemistry_drive import ChemistryDriveConfigError, sync_and_ingest colors = _palette(theme) @@ -1112,10 +1096,6 @@ def water_chemistry_sync_drive( typer.secho(str(exc), fg=colors["issue"], bold=True, err=True) raise typer.Exit(1) from exc - if output_format == OutputFormat.json: - typer.echo(_json.dumps(result.to_payload())) - raise typer.Exit(result.exit_code) - summary = result.to_payload()["summary"] header = ( "[CHEMISTRY DRIVE SYNC] DRY RUN" if result.dry_run else "[CHEMISTRY DRIVE SYNC]" diff --git a/cli/service_adapter.py b/cli/service_adapter.py index 9b7c4393e..b438b0040 100644 --- a/cli/service_adapter.py +++ b/cli/service_adapter.py @@ -89,13 +89,13 @@ def water_levels_csv(source_file: Path | str, *, pretty_json: bool = False): return result -def chemistry_lims_xlsx(source_file: Path | str, *, pretty_json: bool = False): +def chemistry_lims_xlsx(source_file: Path | str): from services.chemistry_lims import bulk_upload_chemistry if isinstance(source_file, str): source_file = Path(source_file) - result = bulk_upload_chemistry(source_file, pretty_json=pretty_json) + result = bulk_upload_chemistry(source_file) if result.stderr: print(result.stderr, file=sys.stderr) return result diff --git a/services/chemistry_lims.py b/services/chemistry_lims.py index 25bb6f553..d5e6c8b20 100644 --- a/services/chemistry_lims.py +++ b/services/chemistry_lims.py @@ -23,7 +23,8 @@ ``Chemistry SampleInfo`` table. This adaptation: * reads an ``.xlsx`` workbook with ``openpyxl``, -* maps each LIMS ``Param`` to an analyte code + target table via ``FMapper``, +* maps each LIMS ``Param`` to an analyte code + target table via + :func:`lookup_analyte`, * resolves each ``SamplePointID`` (the base well PointID) to a ``Thing`` by name, * appends each distinct lab sample (``WCLab_ID``) as a new @@ -37,7 +38,6 @@ from __future__ import annotations import io -import json import re import uuid from dataclasses import dataclass @@ -71,101 +71,99 @@ ANALYSES_AGENCY = "NMBGMR" -class AnalyteField: - def __init__(self, xlsfield, dbanalyte, table, units=None, method=None): - self.xlsfield = xlsfield - self.dbanalyte = dbanalyte - self.table = table - self.units = units - self.method = method - - -class FMapper: - def __init__(self): - self._map = [ - AnalyteField("alkalinity as caco3", "ALK", MAJOR, method="As CaCO3"), - AnalyteField("aluminum", "Al", MINOR), - AnalyteField("anions total", "TAn", MAJOR, EPM), - AnalyteField("antimony 121", "Sb", MINOR), - AnalyteField("antimony 123", "Sb", MINOR), - AnalyteField("antimony", "Sb", MINOR), - AnalyteField("arsenic", "As", MINOR), - AnalyteField("barium", "Ba", MINOR), - AnalyteField("beryllium", "Be", MINOR), - AnalyteField( - "bicarbonate (hco3)", "HCO3", MAJOR, method="Alkalinity as HC03" - ), - AnalyteField("boron 11", "B", MINOR), - AnalyteField("boron", "B", MINOR), - AnalyteField("bromide", "Br", MINOR), - AnalyteField("cadmium 111", "Cd", MINOR), - AnalyteField("cadmium", "Cd", MINOR), - AnalyteField("calcium", "Ca", MAJOR), - AnalyteField("carbonate (co3)", "CO3", MAJOR), - AnalyteField("cations total", "TCat", MAJOR, EPM), - AnalyteField("chloride", "Cl", MAJOR), - AnalyteField("chromium", "Cr", MINOR), - AnalyteField("cobalt", "Co", MINOR), - AnalyteField("copper 65", "Cu", MINOR), - AnalyteField("copper", "Cu", MINOR), - AnalyteField("fluoride", "F", MINOR), - AnalyteField("hardness", "HRD", MAJOR, MGL, method="As CaCO3"), - AnalyteField("iron", "Fe", MINOR), - AnalyteField("lead", "Pb", MINOR), - AnalyteField("lithium", "Li", MINOR), - AnalyteField("magnesium", "Mg", MAJOR), - AnalyteField("manganese", "Mn", MINOR), - AnalyteField("mercury", "Hg", MINOR), - AnalyteField("molybdenum 95", "Mo", MINOR), - AnalyteField("molybdenum", "Mo", MINOR), - AnalyteField("nickel", "Ni", MINOR), - AnalyteField("nitrate", "NO3", MINOR), - AnalyteField("nitrite", "NO2", MINOR), - AnalyteField("phosphate", "PO4", MINOR), - AnalyteField("percent difference", "IONBAL", MAJOR, PDIFF), - AnalyteField("potassium", "K", MAJOR), - AnalyteField("selenium", "Se", MINOR), - AnalyteField("siliconDioxide", "SiO2", MINOR), - AnalyteField("sio2", "SiO2", MINOR), - AnalyteField("silicon", "Si", MINOR), - AnalyteField("silver 107", "Ag", MINOR), - AnalyteField("silver", "Ag", MINOR), - AnalyteField("sodium", "Na", MAJOR), - AnalyteField("specific conductance", "CONDLAB", MAJOR, COND), - AnalyteField("strontium", "Sr", MINOR), - AnalyteField("sulfate", "SO4", MAJOR), - AnalyteField("tds calc", "TDS", MAJOR, method="Calculation"), - AnalyteField("thallium", "Tl", MINOR), - AnalyteField("thorium", "Th", MINOR), - AnalyteField("tin", "Sn", MINOR), - AnalyteField("titanium", "Ti", MINOR), - AnalyteField("uranium", "U", MINOR), - AnalyteField("vanadium", "V", MINOR), - AnalyteField("zinc 66", "Zn", MINOR), - AnalyteField("zinc", "Zn", MINOR), - AnalyteField("pH", "pHL", MAJOR, PH), - AnalyteField("ortho phosphate", "PO4", MINOR), - ] - - def values(self): - return self._map - - def get(self, key, attr="xlsfield"): - if key is None: - return None - for p in self._map: - value = getattr(p, attr) - if not isinstance(value, (list, tuple)): - value = (value,) - for vi in value: - if str(vi).lower() == str(key).lower(): - return p - return None +@dataclass(frozen=True) +class AnalyteMapping: + """Maps a LIMS ``Param`` name to its analyte code and target table. + + ``units`` overrides the LIMS-reported units when set; ``method`` is appended + to the LIMS analysis method when set. + """ + lims_param: str + analyte: str + table: str + units: str | None = None + method: str | None = None + + +# Every known LIMS ``Param`` -> analyte mapping (ported from AMPAPI chemfile.py). +_ANALYTE_MAPPINGS: list[AnalyteMapping] = [ + AnalyteMapping("alkalinity as caco3", "ALK", MAJOR, method="As CaCO3"), + AnalyteMapping("aluminum", "Al", MINOR), + AnalyteMapping("anions total", "TAn", MAJOR, EPM), + AnalyteMapping("antimony 121", "Sb", MINOR), + AnalyteMapping("antimony 123", "Sb", MINOR), + AnalyteMapping("antimony", "Sb", MINOR), + AnalyteMapping("arsenic", "As", MINOR), + AnalyteMapping("barium", "Ba", MINOR), + AnalyteMapping("beryllium", "Be", MINOR), + AnalyteMapping("bicarbonate (hco3)", "HCO3", MAJOR, method="Alkalinity as HC03"), + AnalyteMapping("boron 11", "B", MINOR), + AnalyteMapping("boron", "B", MINOR), + AnalyteMapping("bromide", "Br", MINOR), + AnalyteMapping("cadmium 111", "Cd", MINOR), + AnalyteMapping("cadmium", "Cd", MINOR), + AnalyteMapping("calcium", "Ca", MAJOR), + AnalyteMapping("carbonate (co3)", "CO3", MAJOR), + AnalyteMapping("cations total", "TCat", MAJOR, EPM), + AnalyteMapping("chloride", "Cl", MAJOR), + AnalyteMapping("chromium", "Cr", MINOR), + AnalyteMapping("cobalt", "Co", MINOR), + AnalyteMapping("copper 65", "Cu", MINOR), + AnalyteMapping("copper", "Cu", MINOR), + AnalyteMapping("fluoride", "F", MINOR), + AnalyteMapping("hardness", "HRD", MAJOR, MGL, method="As CaCO3"), + AnalyteMapping("iron", "Fe", MINOR), + AnalyteMapping("lead", "Pb", MINOR), + AnalyteMapping("lithium", "Li", MINOR), + AnalyteMapping("magnesium", "Mg", MAJOR), + AnalyteMapping("manganese", "Mn", MINOR), + AnalyteMapping("mercury", "Hg", MINOR), + AnalyteMapping("molybdenum 95", "Mo", MINOR), + AnalyteMapping("molybdenum", "Mo", MINOR), + AnalyteMapping("nickel", "Ni", MINOR), + AnalyteMapping("nitrate", "NO3", MINOR), + AnalyteMapping("nitrite", "NO2", MINOR), + AnalyteMapping("phosphate", "PO4", MINOR), + AnalyteMapping("percent difference", "IONBAL", MAJOR, PDIFF), + AnalyteMapping("potassium", "K", MAJOR), + AnalyteMapping("selenium", "Se", MINOR), + AnalyteMapping("siliconDioxide", "SiO2", MINOR), + AnalyteMapping("sio2", "SiO2", MINOR), + AnalyteMapping("silicon", "Si", MINOR), + AnalyteMapping("silver 107", "Ag", MINOR), + AnalyteMapping("silver", "Ag", MINOR), + AnalyteMapping("sodium", "Na", MAJOR), + AnalyteMapping("specific conductance", "CONDLAB", MAJOR, COND), + AnalyteMapping("strontium", "Sr", MINOR), + AnalyteMapping("sulfate", "SO4", MAJOR), + AnalyteMapping("tds calc", "TDS", MAJOR, method="Calculation"), + AnalyteMapping("thallium", "Tl", MINOR), + AnalyteMapping("thorium", "Th", MINOR), + AnalyteMapping("tin", "Sn", MINOR), + AnalyteMapping("titanium", "Ti", MINOR), + AnalyteMapping("uranium", "U", MINOR), + AnalyteMapping("vanadium", "V", MINOR), + AnalyteMapping("zinc 66", "Zn", MINOR), + AnalyteMapping("zinc", "Zn", MINOR), + AnalyteMapping("pH", "pHL", MAJOR, PH), + AnalyteMapping("ortho phosphate", "PO4", MINOR), +] + +# Case-insensitive lookup by LIMS ``Param`` name. +_ANALYTE_BY_PARAM: dict[str, AnalyteMapping] = { + m.lims_param.lower(): m for m in _ANALYTE_MAPPINGS +} + + +def lookup_analyte(param: str | None) -> AnalyteMapping | None: + """Return the mapping for a LIMS ``Param`` name, or ``None`` if unknown.""" + if param is None: + return None + return _ANALYTE_BY_PARAM.get(str(param).strip().lower()) -FM = FMapper() -# Target ORM model per FMapper table bucket. +# Target ORM model per analyte table bucket. _TABLE_MODEL = {MAJOR: NMA_MajorChemistry, MINOR: NMA_MinorTraceChemistry} @@ -176,7 +174,6 @@ class ChemistryMappingError(Exception): @dataclass class ChemistryUploadResult: exit_code: int - stdout: str stderr: str payload: dict[str, Any] @@ -272,15 +269,15 @@ def prep_record(record: dict) -> dict: Raises :class:`ChemistryMappingError` when the row cannot be mapped. """ param = _get(record, "Param") - pm = FM.get(param) - if pm is None: + mapping = lookup_analyte(param) + if mapping is None: raise ChemistryMappingError(f"Unmapped analyte Param={param!r}") pointid = _get(record, "SamplePointID") or _get(record, "CustomerSampleNumber") if not pointid: raise ChemistryMappingError("Missing SamplePointID") - units = pm.units or _get(record, "Results_Units") + units = mapping.units or _get(record, "Results_Units") reported = _get(record, "ReportedND") if reported is not None and str(reported).upper() == "ND": @@ -294,9 +291,11 @@ def prep_record(record: dict) -> dict: symbol = None analysis_method = _get(record, "Method") - if pm.method: + if mapping.method: analysis_method = ( - f"{analysis_method}, {pm.method}" if analysis_method else pm.method + f"{analysis_method}, {mapping.method}" + if analysis_method + else mapping.method ) analysis_date = _to_datetime(_get(record, "AnalysisTime")) @@ -304,8 +303,8 @@ def prep_record(record: dict) -> dict: wclab_id = _get(record, "SampleNumber") return { - "analyte": pm.dbanalyte, - "table": pm.table, + "analyte": mapping.analyte, + "table": mapping.table, "units": str(units) if units is not None else None, "symbol": symbol, "sample_value": sample_value, @@ -453,7 +452,7 @@ def _build_measurement( def bulk_upload_chemistry( - source: Path | str | bytes, *, pretty_json: bool = False + source: Path | str | bytes, ) -> ChemistryUploadResult: """Ingest a LIMS ``.xlsx`` workbook into the NMA chemistry tables. @@ -482,7 +481,6 @@ def bulk_upload_chemistry( validation_errors=[f"Could not read workbook: {exc}"], skipped_duplicates=[], created=[], - pretty_json=pretty_json, ) processed = len(raw_records) @@ -518,7 +516,6 @@ def bulk_upload_chemistry( validation_errors=validation_errors, skipped_duplicates=[], created=[], - pretty_json=pretty_json, ) # One sample = one lab sample (WCLab_ID) for a well. @@ -588,7 +585,6 @@ def bucket_key(r: dict) -> tuple[str, str | None]: validation_errors=validation_errors, skipped_duplicates=skipped_duplicates, created=created, - pretty_json=pretty_json, ) @@ -599,7 +595,6 @@ def _result( validation_errors: list[str], skipped_duplicates: list[dict], created: list[dict], - pretty_json: bool, ) -> ChemistryUploadResult: rows_with_issues = len(validation_errors) + len(skipped_duplicates) payload = { @@ -614,7 +609,6 @@ def _result( "skipped_duplicates": skipped_duplicates, "created_samples": created, } - stdout = json.dumps(payload, indent=2 if pretty_json else None) stderr_parts: list[str] = [] if validation_errors: stderr_parts.append("\n".join(validation_errors)) @@ -626,9 +620,7 @@ def _result( stderr = "\n".join(stderr_parts) # Only a data-quality abort is a failure; skipped duplicates are idempotent. exit_code = 1 if validation_errors else 0 - return ChemistryUploadResult( - exit_code=exit_code, stdout=stdout, stderr=stderr, payload=payload - ) + return ChemistryUploadResult(exit_code=exit_code, stderr=stderr, payload=payload) # ============= EOF ============================================= From b682bfcbb3da941a6ed86b358c7b378b835e4388 Mon Sep 17 00:00:00 2001 From: jross Date: Wed, 8 Jul 2026 11:51:07 -0600 Subject: [PATCH 4/4] docs(chemistry): update runbook for CLI cleanup Reflect the renamed analyte map (`lookup_analyte` / `_ANALYTE_MAPPINGS`, was `FMapper`) and the removal of the `--output json` option. Co-Authored-By: Claude Opus 4.8 --- docs/chemistry-ingestion-runbook.md | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/docs/chemistry-ingestion-runbook.md b/docs/chemistry-ingestion-runbook.md index d978edc95..6831d3f37 100644 --- a/docs/chemistry-ingestion-runbook.md +++ b/docs/chemistry-ingestion-runbook.md @@ -94,8 +94,6 @@ Then ingest: oco water-chemistry sync-drive # or point at a specific folder: oco water-chemistry sync-drive --folder-id -# machine-readable: -oco water-chemistry sync-drive --output json ``` Read the summary. Buckets: @@ -112,7 +110,7 @@ Read the summary. Buckets: | Reported cause | Meaning | Action | |----------------|---------|--------| -| `Unmapped analyte Param=...` | A LIMS `Param` name is not in the analyte map. | Send the Param name to engineering to add to `FMapper`. | +| `Unmapped analyte Param=...` | A LIMS `Param` name is not in the analyte map. | Send the Param name to engineering to add to `_ANALYTE_MAPPINGS` in `services/chemistry_lims.py`. | | `no matching Thing (well) found` | `SamplePointID` has no Ocotillo well. | Verify the PointID; ensure the well was transferred to Data Services first. | Exit code is non-zero if any file failed. @@ -146,7 +144,7 @@ oco water-chemistry bulk-upload --file /path/to/batch.xlsx ## 6. What the ingest does (summary) For each workbook: map each `Param` to an analyte code + target table (major vs -minor) via `FMapper`; compute the value (non-detects become +minor) via `lookup_analyte`; compute the value (non-detects become `LowerLimit × Dilution` with a `<` symbol); collapse duplicate (SamplePointID, WCLab_ID, analyte) rows (prefer EPA 200.7, or "low bromide" for Br); resolve the base `SamplePointID → Thing`. Then, per distinct lab sample @@ -168,8 +166,8 @@ matching well) aborts the whole file — nothing is written. - **`.xlsx` only.** Legacy `.xls` LIMS exports are not read; the file must be a modern `.xlsx`. - **Fixed analyte map.** Unknown `Param` names fail until engineering adds them - to `FMapper`. Only major + minor analytes are handled — field parameters and - radionuclides are out of scope. + to `_ANALYTE_MAPPINGS`. Only major + minor analytes are handled — field + parameters and radionuclides are out of scope. - **Well must exist first.** `SamplePointID` must already match an Ocotillo `Thing.name`; otherwise the file fails. - **Failed files retry loudly.** A file that fails (e.g. unmapped analyte or a