feat: add transfer-results command for generating transfer results summary

Author: jirhiker

alembic/versions/43bc34504ee6_merge_migrations_after_staging_merge.py

@@ -1,7 +1,7 @@
 """Add unique index for NGWMN well construction
 
 Revision ID: 50d1c2a3b4c5
-Revises: 43bc34504ee6
+Revises: 3cb924ca51fd
 Create Date: 2026-01-31 00:27:12.204176
 
 """
@@ -12,7 +12,7 @@
 
 # revision identifiers, used by Alembic.
 revision: str = "50d1c2a3b4c5"
-down_revision: Union[str, Sequence[str], None] = "43bc34504ee6"
+down_revision: Union[str, Sequence[str], None] = "3cb924ca51fd"
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 

alembic/versions/50d1c2a3b4c5_add_unique_index_ngwmn_wellconstruction.py

@@ -109,6 +109,33 @@ def associate_assets_command(
     associate_assets(root_directory)
 
 
+@cli.command("transfer-results")
+def transfer_results(
+    summary_path: Path = typer.Option(
+        Path("transfers") / "metrics" / "transfer_results_summary.md",
+        "--summary-path",
+        help="Output path for markdown summary table.",
+    ),
+    sample_limit: int = typer.Option(
+        25,
+        "--sample-limit",
+        min=1,
+        help="Max missing/extra key samples stored per transfer.",
+    ),
+    theme: ThemeMode = typer.Option(
+        ThemeMode.auto, "--theme", help="Color theme: auto, light, dark."
+    ),
+):
+    from transfers.transfer_results_builder import TransferResultsBuilder
+
+    builder = TransferResultsBuilder(sample_limit=sample_limit)
+    results = builder.build()
+    summary_path.parent.mkdir(parents=True, exist_ok=True)
+    TransferResultsBuilder.write_summary(summary_path, results)
+    typer.echo(f"Wrote comparison summary: {summary_path}")
+    typer.echo(f"Transfer comparisons: {len(results.results)}")
+
+
 @cli.command("well-inventory-csv")
 def well_inventory_csv(
     file_path: str = typer.Argument(

cli/cli.py

@@ -108,7 +108,7 @@ dependencies = [
 package = true
 
 [tool.setuptools]
-packages = ["alembic", "cli", "core", "db", "schemas", "services"]
+packages = ["alembic", "cli", "core", "db", "schemas", "services", "transfers"]
 
 [project.scripts]
 oco = "cli.cli:cli"

pyproject.toml

@@ -18,13 +18,15 @@
 import textwrap
 import uuid
 from pathlib import Path
+from types import SimpleNamespace
+
+from sqlalchemy import select
+from typer.testing import CliRunner
 
 from cli.cli import cli
 from cli.service_adapter import WellInventoryResult
 from db import FieldActivity, FieldEvent, Observation, Sample
 from db.engine import session_ctx
-from sqlalchemy import select
-from typer.testing import CliRunner
 
 
 def test_initialize_lexicon_invokes_initializer(monkeypatch):
@@ -95,6 +97,50 @@ def fake_well_inventory(file_path):
     assert "[WELL INVENTORY IMPORT] SUCCESS" in result.output
 
 
+def test_transfer_results_command_writes_summary(monkeypatch, tmp_path):
+    captured: dict[str, object] = {}
+
+    class FakeBuilder:
+        def __init__(self, sample_limit: int = 25):
+            captured["sample_limit"] = sample_limit
+
+        def build(self):
+            captured["built"] = True
+            return SimpleNamespace(
+                results={"WellData": object(), "WaterLevels": object()}
+            )
+
+        @staticmethod
+        def write_summary(path, comparison):
+            captured["summary_path"] = Path(path)
+            captured["result_count"] = len(comparison.results)
+
+    monkeypatch.setattr(
+        "transfers.transfer_results_builder.TransferResultsBuilder", FakeBuilder
+    )
+
+    summary_path = tmp_path / "metrics" / "summary.md"
+    runner = CliRunner()
+    result = runner.invoke(
+        cli,
+        [
+            "transfer-results",
+            "--summary-path",
+            str(summary_path),
+            "--sample-limit",
+            "11",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+    assert captured["sample_limit"] == 11
+    assert captured["built"] is True
+    assert captured["summary_path"] == summary_path
+    assert captured["result_count"] == 2
+    assert f"Wrote comparison summary: {summary_path}" in result.output
+    assert "Transfer comparisons: 2" in result.output
+
+
 def test_well_inventory_csv_command_reports_validation_errors(monkeypatch, tmp_path):
     inventory_file = tmp_path / "inventory.csv"
     inventory_file.write_text("header\nvalue\n")
@@ -198,10 +244,12 @@ def test_water_levels_cli_persists_observations(tmp_path, water_well_thing):
     """
 
     def _write_csv(path: Path, *, well_name: str, notes: str):
-        csv_text = textwrap.dedent(f"""\
+        csv_text = textwrap.dedent(
+            f"""\
             field_staff,well_name_point_id,field_event_date_time,measurement_date_time,sampler,sample_method,mp_height,level_status,depth_to_water_ft,data_quality,water_level_notes
             CLI Tester,{well_name},2025-02-15T08:00:00-07:00,2025-02-15T10:30:00-07:00,Groundwater Team,electric tape,1.5,stable,42.5,approved,{notes}
-            """)
+            """
+        )
         path.write_text(csv_text)
 
     unique_notes = f"pytest-{uuid.uuid4()}"

tests/test_cli_commands.py

@@ -20,6 +20,7 @@
 from dataclasses import dataclass
 
 from dotenv import load_dotenv
+
 from transfers.thing_transfer import (
     transfer_rock_sample_locations,
     transfer_springs,
@@ -698,9 +699,10 @@ def main():
     profile_artifacts = transfer_all(metrics)
 
     metrics.close()
-    metrics.save_to_storage_bucket()
-    save_log_to_bucket()
-    upload_profile_artifacts(profile_artifacts)
+    if get_bool_env("SAVE_TO_BUCKET", False):
+        metrics.save_to_storage_bucket()
+        save_log_to_bucket()
+        upload_profile_artifacts(profile_artifacts)
     message("END--------------------------------------")
 
 

transfers/transfer.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 from pathlib import Path
 from typing import Any
 
@@ -21,7 +22,6 @@
     replace_nans,
     get_transferable_wells,
 )
-import os
 
 
 def _normalize_key(value: Any) -> str | None:
@@ -79,9 +79,11 @@ def _build_one(self, spec: TransferComparisonSpec) -> TransferResult:
         if spec.source_filter:
             source_df = spec.source_filter(source_df)
         comparison_df = source_df
+        if spec.agreed_filter:
+            comparison_df = spec.agreed_filter(comparison_df)
         enabled = self._is_enabled(spec)
         if not enabled:
-            comparison_df = source_df.iloc[0:0]
+            comparison_df = comparison_df.iloc[0:0]
         elif spec.transfer_name == "WellData":
             comparison_df = self._agreed_welldata_df()
 
@@ -179,9 +181,8 @@ def write_summary(path: Path, comparison: TransferComparisonResults) -> None:
         ]
         for name in sorted(comparison.results.keys()):
             r = comparison.results[name]
-            missing_agreed = r.agreed_transfer_row_count - r.destination_row_count
             lines.append(
                 f"| {name} | {r.source_csv} | {r.source_row_count} | {r.agreed_transfer_row_count} | "
-                f"{r.destination_model} | {r.destination_row_count} | {missing_agreed} |"
+                f"{r.destination_model} | {r.destination_row_count} | {r.missing_in_destination_count} |"
             )
         path.write_text("\n".join(lines) + "\n")