diff --git a/README.md b/README.md index de6ea91..489b2d9 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ improv has three storage layers: - **Columnar store** — queryable image metadata, provenance records, and plugin index tables (DuckDB+Parquet or VAST DB) - **OLTP database** — mutable organizing metadata: instruments, samples, datasets, ingest tasks (PostgreSQL or SQLite) -A **plugin system** extends provenance handling. Each plugin handles a specific provenance `kind` and optionally maintains an index table for fast querying. Plugins can be generic (geolocation, sample context) or instrument-specific (IFCB morphometric features, IFCB CNN classification scores). +A **plugin system** extends provenance handling. Plugins register via dependency injection — the core never imports them at load time. Each plugin handles a specific provenance `kind` and optionally maintains an index table for fast querying. Plugins are generic and parameterized (geolocation, sample context, machine classification); instrument-specific presets live under `improv.plugins.ifcb` (IFCB morphometric features, IFCB CNN classification), pinning a `kind`/`index_table` onto a generic plugin. ## Access patterns @@ -24,16 +24,25 @@ A **plugin system** extends provenance handling. Each plugin handles a specific ## REST API -improv exposes a FastAPI service with endpoints for image data and metadata, provenance, instruments, samples, datasets, and ingest task tracking. A thin HTTP client (`improv.client.ImprovClient`) is provided for ingest scripts that need OLTP access without direct database credentials. +improv exposes a FastAPI service with endpoints for image data and metadata, provenance, instruments, samples, datasets, and ingest task tracking. Classifier support adds taxonomy registration/lookup and two decode paths: a stateless decode (caller supplies a vector) and a decoded read that fetches an image's classification provenance and resolves each record against its own `model_version`. A thin HTTP client (`improv.client.ImprovClient`) is provided for ingest scripts that need OLTP access without direct database credentials, including taxonomy registration. ## Ingest architecture Batch producers (ingest pipelines, classifiers) use a hybrid approach: -- **OLTP operations** (register instruments, samples, ingest tasks) go through the REST API +- **OLTP operations** (register instruments, samples, ingest tasks, classifier taxonomies) go through the REST API - **High-volume writes** (image metadata, provenance, index tables, image bytes) go directly to the columnar store and object store -This avoids coupling ingest scripts to the database while keeping high-throughput writes off the HTTP path. +This avoids coupling ingest scripts to the database while keeping high-throughput writes off the HTTP path. Low-volume producers that prefer not to depend on the columnar store directly can post small provenance batches over REST (`POST /images/provenance/batch`, one instrument per batch). + +## Idempotency + +The provenance log is append-only, so writes are never overwritten — idempotency is achieved by **appending, then deduplicating at read time**. This works identically on every backend (VAST DB, DuckDB+Parquet, and any future one), because append is the only operation they all share. + +- A **provenance record** is identified by `(image_id, kind, source, data_hash)`, where `data_hash` is a canonical [RFC 8785 (JCS)](https://www.rfc-editor.org/rfc/rfc8785) hash of the `data` payload, stamped server-side. Re-posting a byte-identical record re-appends a row that collapses to one on read; a genuinely different payload (e.g. a new `model_version`) hashes differently and is retained. Canonicalization normalizes key order and number formatting (`1.0` == `1`) across producers in different languages, and rejects NaN/Infinity. +- **Index records** are deterministic projections deduplicated on their full column tuple. + +**Client contract:** a record's `timestamp` is the **event time** (when the image was collected, or the classifier result produced) — a property of the observed fact, captured once. It is *not* the time of the HTTP request. Retries must resend the identical record, so put real event time in `timestamp`, never wall-clock-at-send; otherwise each attempt looks distinct and will not deduplicate. (The server separately records its own write time.) ## Install @@ -51,5 +60,6 @@ pip install '.[service]' # adds FastAPI, CLI, migrations | `amplify-storage-utils` | Object storage (HashdirStore / S3) | | `pydantic` | Models and validation | | `pyarrow` | Columnar data exchange | +| `rfc8785` | Canonical (JCS) hashing of provenance payloads for idempotency | | `httpx` | Thin ingest client | | `fastapi`, `sqlalchemy`, `alembic` | Service extras | diff --git a/pyproject.toml b/pyproject.toml index b52ae62..36f0468 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,6 +10,7 @@ requires-python = ">=3.10" dependencies = [ "pydantic>=2.0", "pyarrow>=14.0", + "rfc8785>=0.1", "httpx", "amplify-db-utils @ git+https://github.com/WHOIGit/amplify-db-utils.git", "amplify-storage-utils @ git+https://github.com/WHOIGit/amplify-storage-utils.git@v1.7.0", @@ -20,6 +21,9 @@ db = [ "sqlalchemy>=2.0", "psycopg2-binary", ] +vastdb = [ + "amplify-db-utils[vastdb] @ git+https://github.com/WHOIGit/amplify-db-utils.git", +] service = [ "fastapi", "uvicorn[standard]", diff --git a/src/improv/api/app.py b/src/improv/api/app.py index 9eda241..47e7eff 100644 --- a/src/improv/api/app.py +++ b/src/improv/api/app.py @@ -21,7 +21,7 @@ def create_app(config: "ImprovConfig") -> FastAPI: from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker - from improv.api.routers import blobs, datasets, images, ingest_tasks, instruments, provenance, samples + from improv.api.routers import blobs, classification, datasets, images, ingest_tasks, instruments, provenance, samples from improv.oltp.models import Base from improv.service import ImageService from improv.store.tables import register_service_tables @@ -59,5 +59,6 @@ async def lifespan(app: FastAPI): app.include_router(blobs.router) app.include_router(datasets.router) app.include_router(ingest_tasks.router) + app.include_router(classification.router) return app diff --git a/src/improv/api/routers/classification.py b/src/improv/api/routers/classification.py new file mode 100644 index 0000000..d1b1624 --- /dev/null +++ b/src/improv/api/routers/classification.py @@ -0,0 +1,165 @@ +"""Classifier taxonomy + classification-decode endpoints. + +Thin routing over existing ImageService methods. No decode/taxonomy business +logic lives here — the service owns it (see ImageService.decode_classification +and register/get_classifier_taxonomy). Decoding always uses a record's own +model_version, never "latest" (which is display / new-work only). +""" + +from __future__ import annotations + +from fastapi import APIRouter, Depends, HTTPException, status + +from improv.api.deps import get_service +from improv.api.schemas import ( + DecodedClassificationResponse, + DecodeRequest, + TaxonomyCreate, + TaxonomyResponse, +) +from improv.service import ImageService + +router = APIRouter(tags=["classification"]) + + +def _taxonomy_response(tax) -> TaxonomyResponse: + return TaxonomyResponse( + classifier=tax.classifier, + model_version=tax.model_version, + class_names=tax.class_names, + created_at=tax.created_at, + ) + + +# --------------------------------------------------------------------------- +# Taxonomy registration + lookup +# --------------------------------------------------------------------------- + +@router.post( + "/classifiers/{classifier}/taxonomies", + response_model=TaxonomyResponse, + status_code=status.HTTP_201_CREATED, +) +def register_taxonomy( + classifier: str, + body: TaxonomyCreate, + service: ImageService = Depends(get_service), +) -> TaxonomyResponse: + taxonomy, created = service.register_classifier_taxonomy( + classifier, body.model_version, body.class_names + ) + if not created: + raise HTTPException( + status_code=409, + detail=f"Taxonomy for {classifier!r}/{body.model_version!r} already exists.", + ) + return _taxonomy_response(taxonomy) + + +# Literal /latest registered before /{model_version} so it isn't captured as a +# version (same pattern the provenance router documents). +@router.get( + "/classifiers/{classifier}/taxonomies/latest", + response_model=TaxonomyResponse, +) +def get_latest_taxonomy( + classifier: str, + service: ImageService = Depends(get_service), +) -> TaxonomyResponse: + taxonomy = service.get_latest_classifier_taxonomy(classifier) + if taxonomy is None: + raise HTTPException( + status_code=404, detail=f"No taxonomy registered for classifier {classifier!r}." + ) + return _taxonomy_response(taxonomy) + + +@router.get( + "/classifiers/{classifier}/taxonomies/{model_version}", + response_model=TaxonomyResponse, +) +def get_taxonomy( + classifier: str, + model_version: str, + service: ImageService = Depends(get_service), +) -> TaxonomyResponse: + taxonomy = service.get_classifier_taxonomy(classifier, model_version) + if taxonomy is None: + raise HTTPException( + status_code=404, + detail=f"No taxonomy for {classifier!r}/{model_version!r}.", + ) + return _taxonomy_response(taxonomy) + + +# --------------------------------------------------------------------------- +# Decoding — (A) stateless, (B) decoded classification read +# --------------------------------------------------------------------------- + +def _decode( + service: ImageService, + classifier: str, + model_version: str, + scores: list[float], + winner_index: int, +) -> DecodedClassificationResponse: + """Decode one vector, mapping the service's ValueErrors to HTTP status. + + Pre-checks taxonomy existence (→404) so a missing taxonomy is distinguished + from a length/range mismatch (→422); decode_classification raises an + undifferentiated ValueError for all three. + """ + if service.get_classifier_taxonomy(classifier, model_version) is None: + raise HTTPException( + status_code=404, + detail=f"No taxonomy for {classifier!r}/{model_version!r}.", + ) + try: + decoded = service.decode_classification( + classifier, model_version, scores, winner_index + ) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + return DecodedClassificationResponse(**decoded) + + +@router.post( + "/classifiers/{classifier}/decode", + response_model=DecodedClassificationResponse, +) +def decode( + classifier: str, + body: DecodeRequest, + service: ImageService = Depends(get_service), +) -> DecodedClassificationResponse: + return _decode( + service, classifier, body.model_version, body.scores, body.winner_index + ) + + +@router.get( + "/images/{image_id}/classification", + response_model=list[DecodedClassificationResponse], +) +def get_decoded_classification( + image_id: str, + kind: str, + instrument: str | None = None, + service: ImageService = Depends(get_service), +) -> list[DecodedClassificationResponse]: + """Read an image's classification-kind provenance, decoded. + + `classifier` is the plugin `kind`. Each record decodes against its OWN + model_version (from the stored payload), never "latest". + """ + records = service.get_provenance(image_id, kind=kind, instrument=instrument) + return [ + _decode( + service, + kind, + r.data["model_version"], + r.data["scores"], + r.data["winner_index"], + ) + for r in records + ] diff --git a/src/improv/api/routers/provenance.py b/src/improv/api/routers/provenance.py index 48d0b57..c0e8967 100644 --- a/src/improv/api/routers/provenance.py +++ b/src/improv/api/routers/provenance.py @@ -24,6 +24,8 @@ def _to_response(env: ProvenanceEnvelope) -> ProvenanceResponse: timestamp=env.timestamp, data=env.data, instrument=env.instrument, + year=env.year, + month=env.month, ) @@ -39,17 +41,34 @@ def ingest_provenance_batch( ) -> dict: envelopes = [ ProvenanceEnvelope( - image_id="", # overridden per record below — batch has no shared image_id + image_id=r.image_id, kind=r.kind, source=r.source, timestamp=r.timestamp, data=r.data, - instrument=instrument, + instrument=instrument, # uniform fallback hint; parser wins per record ) for r in body.records ] - service.ingest_provenance(envelopes) - return {"ingested": len(envelopes)} + + # A batch is assumed single-instrument. Resolve each record's instrument up + # front and reject (before any write) if the batch spans more than one, or if + # a record's image_id neither parses nor has the fallback hint. + try: + enriched = service.enrich_envelopes(envelopes) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + + instruments = {e.instrument for e in enriched} + if len(instruments) > 1: + raise HTTPException( + status_code=422, + detail=f"Batch spans multiple instruments {sorted(instruments)}; " + "post one instrument's records at a time.", + ) + + service.ingest_provenance(enriched) + return {"ingested": len(enriched)} @router.get( diff --git a/src/improv/api/schemas.py b/src/improv/api/schemas.py index daa20b7..c9e9666 100644 --- a/src/improv/api/schemas.py +++ b/src/improv/api/schemas.py @@ -54,6 +54,8 @@ class ProvenanceResponse(BaseModel): timestamp: datetime data: dict instrument: str | None = None + year: int | None = None + month: int | None = None class ProvenanceIngest(BaseModel): @@ -63,8 +65,45 @@ class ProvenanceIngest(BaseModel): data: dict +class ProvenanceBatchRecord(BaseModel): + # Batch records carry their own image_id (unlike single-record ingest, where + # image_id comes from the path). + image_id: str + kind: str + source: str + timestamp: datetime + data: dict + + class ProvenanceBatchIngest(BaseModel): - records: list[ProvenanceIngest] + records: list[ProvenanceBatchRecord] + + +# --------------------------------------------------------------------------- +# Classification / taxonomy +# --------------------------------------------------------------------------- + +class TaxonomyCreate(BaseModel): + model_version: str + class_names: list[str] + + +class TaxonomyResponse(BaseModel): + classifier: str + model_version: str + class_names: list[str] + created_at: datetime + + +class DecodeRequest(BaseModel): + model_version: str + scores: list[float] + winner_index: int + + +class DecodedClassificationResponse(BaseModel): + winner: str + scores: dict[str, float] # --------------------------------------------------------------------------- diff --git a/src/improv/client.py b/src/improv/client.py index 66ca927..5b2ef86 100644 --- a/src/improv/client.py +++ b/src/improv/client.py @@ -210,3 +210,40 @@ def get_ingest_task(self, task_id: str) -> dict | None: return None resp.raise_for_status() return resp.json() + + # ------------------------------------------------------------------ + # Classifier taxonomy + # ------------------------------------------------------------------ + + def register_classifier_taxonomy( + self, + classifier: str, + model_version: str, + class_names: list[str], + ) -> tuple[dict, bool]: + """Register a classifier taxonomy. Returns (taxonomy_dict, created). + + Returns created=False if the (classifier, model_version) already exists + (409). Batch producers register a taxonomy before ingesting the + positional score vectors that decode against it. + """ + resp = self._client.post( + f"/classifiers/{classifier}/taxonomies", + json={"model_version": model_version, "class_names": class_names}, + ) + if resp.status_code == 409: + return self.get_classifier_taxonomy(classifier, model_version), False + resp.raise_for_status() + return resp.json(), True + + def get_classifier_taxonomy( + self, classifier: str, model_version: str + ) -> dict | None: + """Get a taxonomy by exact (classifier, model_version). None if absent.""" + resp = self._client.get( + f"/classifiers/{classifier}/taxonomies/{model_version}" + ) + if resp.status_code == 404: + return None + resp.raise_for_status() + return resp.json() diff --git a/src/improv/hashing.py b/src/improv/hashing.py new file mode 100644 index 0000000..a4ef710 --- /dev/null +++ b/src/improv/hashing.py @@ -0,0 +1,28 @@ +"""Canonical hashing of provenance payloads. + +Idempotency for the append-only provenance log rests on a content hash: a +retried write re-appends a byte-identical row that is collapsed at read time, +while a genuinely different payload hashes differently and is retained. + +The hash is computed over the RFC 8785 JSON Canonicalization Scheme (JCS) form +of the payload, so two semantically-equal payloads hash identically regardless +of key order, whitespace, or numeric formatting (e.g. ``1.0`` and ``1`` both +canonicalize to ``1``). This holds across producers in different languages +(Python, MATLAB, JS), which flat ``json.dumps(sort_keys=True)`` does not +guarantee. NaN/Infinity are rejected — they are not valid JSON. +""" + +from __future__ import annotations + +import hashlib + +import rfc8785 + + +def canonical_data_hash(data: dict) -> str: + """Return the SHA-256 hex digest of *data*'s RFC 8785 canonical form. + + Raises ``ValueError`` (via rfc8785) if *data* contains NaN/Infinity or a + value with no canonical JSON representation. + """ + return hashlib.sha256(rfc8785.dumps(data)).hexdigest() diff --git a/src/improv/models/provenance.py b/src/improv/models/provenance.py index 1710b95..2cd5b81 100644 --- a/src/improv/models/provenance.py +++ b/src/improv/models/provenance.py @@ -22,3 +22,10 @@ class ProvenanceEnvelope(BaseModel): instrument: str | None = None year: int | None = None month: int | None = None + + # Idempotency columns — stamped server-side at write, never client-supplied. + # data_hash is the RFC 8785 canonical hash of `data` and, together with + # (image_id, kind, source), defines row identity for read-time dedup. + # written_at records when the row was appended (audit; retries differ here). + data_hash: str | None = None + written_at: datetime | None = None diff --git a/src/improv/oltp/migrations/versions/002_classifier_taxonomy.py b/src/improv/oltp/migrations/versions/002_classifier_taxonomy.py new file mode 100644 index 0000000..7be738a --- /dev/null +++ b/src/improv/oltp/migrations/versions/002_classifier_taxonomy.py @@ -0,0 +1,43 @@ +"""Classifier taxonomy (label-map) table. + +Revision ID: 002 +Revises: 001 +Create Date: 2026-07-06 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision: str = "002" +down_revision: str | None = "001" +branch_labels: str | None = None +depends_on: str | None = None + + +def upgrade() -> None: + op.create_table( + "classifier_taxonomy", + sa.Column("taxonomy_id", sa.String(36), nullable=False), + sa.Column("classifier", sa.String(), nullable=False), + sa.Column("model_version", sa.String(), nullable=False), + sa.Column("class_names", sa.JSON(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("taxonomy_id"), + sa.UniqueConstraint( + "classifier", "model_version", name="uq_classifier_taxonomy_version" + ), + ) + op.create_index( + "ix_classifier_taxonomy_classifier", + "classifier_taxonomy", + ["classifier"], + ) + + +def downgrade() -> None: + op.drop_index( + "ix_classifier_taxonomy_classifier", table_name="classifier_taxonomy" + ) + op.drop_table("classifier_taxonomy") diff --git a/src/improv/oltp/models.py b/src/improv/oltp/models.py index 35077bc..93254b0 100644 --- a/src/improv/oltp/models.py +++ b/src/improv/oltp/models.py @@ -152,3 +152,38 @@ class IngestTask(Base): def __repr__(self) -> str: return f"" + + +class ClassifierTaxonomy(Base): + """Ordered class-name list (label-map) for one classifier + model version. + + A machine classification record stores scores positionally (a float vector) + and the winner as an integer index; the class names live here, keyed by + (classifier, model_version), where classifier is the plugin kind + (e.g. "ifcb_cnn_classification"). Index position in class_names is the class + id. Append-only: any change to the class list/order requires a new + model_version, so historical vectors decode against their own version. + """ + + __tablename__ = "classifier_taxonomy" + __table_args__ = ( + UniqueConstraint( + "classifier", "model_version", name="uq_classifier_taxonomy_version" + ), + ) + + taxonomy_id: Mapped[str] = mapped_column( + String(36), primary_key=True, default=lambda: str(uuid.uuid4()) + ) + classifier: Mapped[str] = mapped_column(String, nullable=False, index=True) + model_version: Mapped[str] = mapped_column(String, nullable=False) + class_names: Mapped[list] = mapped_column(JSON, nullable=False) # ordered list[str] + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False + ) + + def __repr__(self) -> str: + return ( + f"" + ) diff --git a/src/improv/oltp/queries.py b/src/improv/oltp/queries.py index 0b01a9e..f672a93 100644 --- a/src/improv/oltp/queries.py +++ b/src/improv/oltp/queries.py @@ -7,7 +7,14 @@ from sqlalchemy.orm import Session -from improv.oltp.models import Dataset, DatasetSpan, IngestTask, Instrument, Sample +from improv.oltp.models import ( + ClassifierTaxonomy, + Dataset, + DatasetSpan, + IngestTask, + Instrument, + Sample, +) # --------------------------------------------------------------------------- @@ -225,3 +232,63 @@ def delete_ingest_task(session: Session, task_id: str) -> bool: session.delete(task) session.flush() return True + + +# --------------------------------------------------------------------------- +# ClassifierTaxonomy +# --------------------------------------------------------------------------- + +def get_classifier_taxonomy( + session: Session, + classifier: str, + model_version: str, +) -> ClassifierTaxonomy | None: + """Return the label-map for an exact (classifier, model_version).""" + return ( + session.query(ClassifierTaxonomy) + .filter( + ClassifierTaxonomy.classifier == classifier, + ClassifierTaxonomy.model_version == model_version, + ) + .first() + ) + + +def get_latest_classifier_taxonomy( + session: Session, + classifier: str, +) -> ClassifierTaxonomy | None: + """Return the most recently registered label-map for a classifier. + + For display / new work only — decode historical vectors against their own + model_version via get_classifier_taxonomy. + """ + return ( + session.query(ClassifierTaxonomy) + .filter(ClassifierTaxonomy.classifier == classifier) + .order_by(ClassifierTaxonomy.created_at.desc()) + .first() + ) + + +def register_classifier_taxonomy( + session: Session, + classifier: str, + model_version: str, + class_names: list[str], + now: datetime, +) -> ClassifierTaxonomy: + """Create or return the existing label-map for (classifier, model_version).""" + existing = get_classifier_taxonomy(session, classifier, model_version) + if existing is not None: + return existing + taxonomy = ClassifierTaxonomy( + taxonomy_id=str(uuid.uuid4()), + classifier=classifier, + model_version=model_version, + class_names=list(class_names), + created_at=now, + ) + session.add(taxonomy) + session.flush() + return taxonomy diff --git a/src/improv/plugins/__init__.py b/src/improv/plugins/__init__.py index bbbd258..8e62e11 100644 --- a/src/improv/plugins/__init__.py +++ b/src/improv/plugins/__init__.py @@ -1,7 +1,12 @@ -"""Provenance plugin protocol and built-in plugin exports. +"""Provenance plugin protocols and built-in (instrument-agnostic) plugin exports. Plugins drive dual-writes when provenance records arrive via the REST API. Batch producers bypass plugins and own their dual-writes directly. + +Only instrument-agnostic plugins are imported here. Instrument-specific +plugins live in their own subpackages (e.g. ``improv.plugins.ifcb``) and are +NOT imported at core load time — import them explicitly where they are wired +in, so that core can be used without their dependencies present. """ from __future__ import annotations @@ -9,8 +14,11 @@ from typing import TYPE_CHECKING, Protocol, runtime_checkable if TYPE_CHECKING: + from datetime import datetime + import pyarrow as pa from amplify_db_utils import ColumnarStore + from improv.models.provenance import ProvenanceEnvelope @@ -44,40 +52,89 @@ def extract_index_record( ... +# --------------------------------------------------------------------------- +# Optional query-capability protocols +# +# A plugin owns not just how its index is written but how it is queried. The +# service discovers these capabilities structurally (isinstance against the +# runtime-checkable protocol) rather than importing concrete plugin classes, +# so the read path stays decoupled from any specific plugin. +# +# IDEMPOTENCY CONTRACT: index tables are append-only, so a retried write +# re-appends rows. Every query implementation MUST deduplicate rows read from +# its index table before returning, via ``improv.store.dedup.dedup_rows`` — +# otherwise retries surface as duplicate results. See GeoLocationPlugin. +# query_spatial / SampleContextPlugin.query_by_sample_id for the pattern. +# --------------------------------------------------------------------------- + +@runtime_checkable +class SpatialQueryPlugin(Protocol): + """Capability: answer a lat/lon bounding-box query, returning image_ids.""" + + def query_spatial( + self, + store: "ColumnarStore", + lat_min: float, + lat_max: float, + lon_min: float, + lon_max: float, + time_start: "datetime | None" = None, + time_end: "datetime | None" = None, + ) -> list[str]: + ... + + +@runtime_checkable +class SampleQueryPlugin(Protocol): + """Capability: resolve a sample_id to its image_ids.""" + + def query_by_sample_id( + self, + store: "ColumnarStore", + sample_id: str, + source: str | None = None, + ) -> list[str]: + ... + + from improv.plugins.annotation import ( - FullFrameRegion, BBoxRegion, - IFCBCNNClassificationIndexRecord, - IFCBCNNClassificationPlugin, - MachineAnnotationRecord, + FullFrameRegion, RegionDescriptor, ) from improv.plugins.blob import BlobPlugin, BlobRecord -from improv.plugins.geolocation import GeoLocationIndexRecord, GeoLocationPlugin -from improv.plugins.ifcb_features import ( - IFCBFeaturesIndexRecord, - IFCBFeaturesPlugin, - IFCBFeaturesRecord, +from improv.plugins.classification import ( + MachineClassificationIndexRecord, + MachineClassificationPlugin, + MachineClassificationRecord, +) +from improv.plugins.geolocation import ( + GeoLocationIndexRecord, + GeoLocationPlugin, + GeoLocationRecord, +) +from improv.plugins.sample_context import ( + SampleContextPlugin, + SampleContextRecord, + SampleIndexRecord, ) -from improv.plugins.ifcb_id import IFCBImageIdParser -from improv.plugins.sample_context import SampleContextPlugin, SampleIndexRecord __all__ = [ "ProvenancePlugin", - "MachineAnnotationRecord", + "SpatialQueryPlugin", + "SampleQueryPlugin", "FullFrameRegion", "BBoxRegion", "RegionDescriptor", - "IFCBCNNClassificationPlugin", - "IFCBCNNClassificationIndexRecord", "BlobPlugin", "BlobRecord", - "IFCBFeaturesPlugin", - "IFCBFeaturesRecord", - "IFCBFeaturesIndexRecord", + "MachineClassificationPlugin", + "MachineClassificationIndexRecord", + "MachineClassificationRecord", "GeoLocationPlugin", + "GeoLocationRecord", "GeoLocationIndexRecord", "SampleContextPlugin", - "IFCBImageIdParser", + "SampleContextRecord", "SampleIndexRecord", ] diff --git a/src/improv/plugins/annotation.py b/src/improv/plugins/annotation.py index cfd9d34..0cfe400 100644 --- a/src/improv/plugins/annotation.py +++ b/src/improv/plugins/annotation.py @@ -1,44 +1,15 @@ -"""Annotation provenance plugins. +"""Human annotation schemas (stubs). -MachineAnnotationRecord is the shared payload schema for all machine annotation -kinds. Each classifier family gets its own plugin with its own kind and index -table — the index schema (which class columns exist) is classifier-family-specific. - -Currently implemented: - IFCBCNNClassificationPlugin kind="ifcb_cnn_classification" - -human_annotation: stubs (RegionDescriptor, FullFrameRegion, BBoxRegion) are -defined as data classes here for reference. Full plugin implementation deferred -until annotation tool integration (Photic / LabelStudio). +Machine classification payloads live in ``improv.plugins.classification`` +(MachineClassificationRecord). This module holds only the human-annotation +region stubs (RegionDescriptor, FullFrameRegion, BBoxRegion), defined as data +classes for reference. Full plugin implementation deferred until annotation +tool integration (Photic / LabelStudio). """ from __future__ import annotations from dataclasses import dataclass -from typing import TYPE_CHECKING - -from pydantic import BaseModel - -if TYPE_CHECKING: - from amplify_db_utils import ColumnarStore - from improv.models.provenance import ProvenanceEnvelope - - -# --------------------------------------------------------------------------- -# Shared machine annotation payload schema -# --------------------------------------------------------------------------- - -class MachineAnnotationRecord(BaseModel): - """Payload schema for machine annotation provenance records (goes in data field). - - Shared across all classifier families — the kind field on the envelope - identifies which classifier family produced the record. - """ - run_id: str - model_version: str # e.g. "ecotaxa-cnn-v4" - scores: dict[str, float] # class name → score (full distribution) - winner: str - winner_score: float # --------------------------------------------------------------------------- @@ -62,52 +33,3 @@ class BBoxRegion: # RegionDescriptor is the union of region types RegionDescriptor = FullFrameRegion | BBoxRegion - - -# --------------------------------------------------------------------------- -# IFCB CNN classification plugin -# --------------------------------------------------------------------------- - -class IFCBCNNClassificationIndexRecord(BaseModel): - """Wide index record — one row per ROI per classifier run.""" - image_id: str - run_id: str - model_version: str - winner: str - winner_score: float - # --- one column per class score (IFCB CNN taxonomy) --- - # e.g. Ceratium: float | None = None - # ... full class list from classifier taxonomy (to be expanded) - # Partition keys - instrument: str | None = None - year: int | None = None - month: int | None = None - - -class IFCBCNNClassificationPlugin: - kind = "ifcb_cnn_classification" - index_table = "ifcb_cnn_classification_index" - index_schema = IFCBCNNClassificationIndexRecord - partition_by = ["instrument", "model_version", "year", "month"] - - def create_index(self, store: "ColumnarStore") -> None: - store.create_table( - self.index_table, - self.index_schema, - partition_by=self.partition_by, - ) - - def extract_index_record(self, envelope: "ProvenanceEnvelope") -> dict | None: - data = envelope.data - scores = data.get("scores", {}) - return { - "image_id": envelope.image_id, - "run_id": data["run_id"], - "model_version": data["model_version"], - "winner": data["winner"], - "winner_score": data["winner_score"], - **scores, - "instrument": envelope.instrument, - "year": envelope.year, - "month": envelope.month, - } diff --git a/src/improv/plugins/classification.py b/src/improv/plugins/classification.py new file mode 100644 index 0000000..a9011d5 --- /dev/null +++ b/src/improv/plugins/classification.py @@ -0,0 +1,102 @@ +"""Machine classification provenance plugin (instrument-agnostic). + +Handles machine-classifier output for any classifier family. Scores are stored +**positionally** (a float vector, no class names); the winning class is an +integer index into that vector. Class names live once per classifier in the +ClassifierTaxonomy registry (OLTP), keyed by (classifier, model_version), and +are used to decode index -> name. See ImageService.decode_classification. + +Invariant: the vector order is defined by the taxonomy for the record's +model_version. Any change to the class list/order requires a new model_version, +so historical vectors always decode against their own version. + +This plugin maintains a narrow "winner-index" — one row per ROI per classifier +run recording only winner_index and winner_score. The full score vector is NOT +indexed: it lives in the provenance payload (envelope.data.scores) and is +retrievable via get_provenance. + +Each classifier family registers its own instance with a distinct kind and +index_table, e.g.:: + + MachineClassificationPlugin(kind="ecotaxa_cnn", index_table="ecotaxa_cnn_index") +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from pydantic import BaseModel + +if TYPE_CHECKING: + from amplify_db_utils import ColumnarStore + from improv.models.provenance import ProvenanceEnvelope + + +class MachineClassificationRecord(BaseModel): + """Payload schema for machine classification provenance records (data field). + + scores is positional — index position is the class id; names come from the + ClassifierTaxonomy for this (classifier, model_version). + """ + run_id: str + model_version: str + scores: list[float] # positional class scores, no names + winner_index: int # index into scores + winner_score: float # == scores[winner_index] (convenience) + + +class MachineClassificationIndexRecord(BaseModel): + """Narrow winner-index record — one row per ROI per classifier run.""" + image_id: str + run_id: str + model_version: str + winner_index: int + winner_score: float + # Partition keys + instrument: str | None = None + year: int | None = None + month: int | None = None + + +class MachineClassificationPlugin: + """Generic classifier plugin. Parameterize kind/index_table per family.""" + + def __init__( + self, + kind: str = "machine_classification", + index_table: str = "machine_classification_index", + partition_by: list[str] | None = None, + ) -> None: + self.kind = kind + self.index_table = index_table + self.index_schema = MachineClassificationIndexRecord + self.partition_by = partition_by or [ + "instrument", "model_version", "year", "month" + ] + + def create_index(self, store: "ColumnarStore") -> None: + store.create_table( + self.index_table, + self.index_schema, + partition_by=self.partition_by, + ) + + def extract_index_record(self, envelope: "ProvenanceEnvelope") -> dict | None: + data = envelope.data + winner_index = data["winner_index"] + scores = data["scores"] + if not 0 <= winner_index < len(scores): + raise ValueError( + f"winner_index {winner_index} out of range for {len(scores)} scores " + f"(image_id={envelope.image_id!r})" + ) + return { + "image_id": envelope.image_id, + "run_id": data["run_id"], + "model_version": data["model_version"], + "winner_index": winner_index, + "winner_score": data["winner_score"], + "instrument": envelope.instrument, + "year": envelope.year, + "month": envelope.month, + } diff --git a/src/improv/plugins/geolocation.py b/src/improv/plugins/geolocation.py index e53aaf3..b1ce39c 100644 --- a/src/improv/plugins/geolocation.py +++ b/src/improv/plugins/geolocation.py @@ -7,16 +7,24 @@ from __future__ import annotations -from datetime import datetime +from datetime import datetime, timezone from typing import TYPE_CHECKING from pydantic import BaseModel +from improv.store.dedup import dedup_rows + if TYPE_CHECKING: from amplify_db_utils import ColumnarStore from improv.models.provenance import ProvenanceEnvelope +def _as_utc(ts: datetime) -> datetime: + if ts.tzinfo is None: + return ts.replace(tzinfo=timezone.utc) + return ts + + class GeoLocationRecord(BaseModel): """Payload schema for geolocation provenance records (goes in data field).""" lat: float @@ -69,3 +77,45 @@ def extract_index_record(self, envelope: "ProvenanceEnvelope") -> dict | None: "year": envelope.year, "month": envelope.month, } + + # --- SpatialQueryPlugin capability ------------------------------------- + + def write_index( + self, + store: "ColumnarStore", + records: list[GeoLocationIndexRecord], + ) -> None: + """Batch-write geolocation index records.""" + store.write(self.index_table, [r.model_dump() for r in records]) + + def query_spatial( + self, + store: "ColumnarStore", + lat_min: float, + lat_max: float, + lon_min: float, + lon_max: float, + time_start: datetime | None = None, + time_end: datetime | None = None, + ) -> list[str]: + """Return image_ids within a lat/lon bounding box, optionally scoped by time. + + Time filtering uses the year/month partition keys for efficient pruning + (month-level precision). Callers may apply exact timestamp filtering + against the images table afterward. + """ + filters: dict = { + "lat": {"gte": lat_min, "lte": lat_max}, + "lon": {"gte": lon_min, "lte": lon_max}, + } + # Approximate time scoping via partition keys when both bounds share a year + if time_start is not None and time_end is not None: + time_start = _as_utc(time_start) + time_end = _as_utc(time_end) + if time_start.year == time_end.year: + filters["year"] = time_start.year + if time_start.month == time_end.month: + filters["month"] = time_start.month + + rows = dedup_rows(store.read(self.index_table, filters=filters)) + return [row["image_id"] for row in rows] diff --git a/src/improv/plugins/ifcb/__init__.py b/src/improv/plugins/ifcb/__init__.py new file mode 100644 index 0000000..5207b1e --- /dev/null +++ b/src/improv/plugins/ifcb/__init__.py @@ -0,0 +1,37 @@ +"""IFCB-specific provenance plugins and image ID parser. + +Not imported by ``improv.plugins`` at core load time. Import from here +explicitly when wiring an IFCB deployment:: + + from improv.plugins.ifcb import ( + IFCBFeaturesPlugin, + IFCBCNNClassificationPlugin, + IFCBImageIdParser, + ) + +Installed as part of the ``improv`` distribution; the ``improv[ifcb]`` extra +is the seam for IFCB-specific third-party dependencies (e.g. a future +ifcb-features port) — see pyproject.toml. +""" + +from __future__ import annotations + +from improv.plugins.ifcb.classification import ( + IFCBCNNClassificationIndexRecord, + IFCBCNNClassificationPlugin, +) +from improv.plugins.ifcb.features import ( + IFCBFeaturesIndexRecord, + IFCBFeaturesPlugin, + IFCBFeaturesRecord, +) +from improv.plugins.ifcb.image_id import IFCBImageIdParser + +__all__ = [ + "IFCBFeaturesPlugin", + "IFCBFeaturesRecord", + "IFCBFeaturesIndexRecord", + "IFCBCNNClassificationPlugin", + "IFCBCNNClassificationIndexRecord", + "IFCBImageIdParser", +] diff --git a/src/improv/plugins/ifcb/classification.py b/src/improv/plugins/ifcb/classification.py new file mode 100644 index 0000000..8f8fe5d --- /dev/null +++ b/src/improv/plugins/ifcb/classification.py @@ -0,0 +1,26 @@ +"""IFCB CNN classification plugin — a preset of the generic classifier plugin. + +The classification plugin is instrument-agnostic (see +improv.plugins.classification). This module only pins the IFCB CNN kind and +index-table names for back-compat and convenient wiring of IFCB deployments. +""" + +from __future__ import annotations + +from improv.plugins.classification import ( + MachineClassificationIndexRecord, + MachineClassificationPlugin, +) + +# Back-compat alias — the index schema is the generic narrow winner-index. +IFCBCNNClassificationIndexRecord = MachineClassificationIndexRecord + + +class IFCBCNNClassificationPlugin(MachineClassificationPlugin): + """MachineClassificationPlugin preset for IFCB CNN classifier output.""" + + def __init__(self) -> None: + super().__init__( + kind="ifcb_cnn_classification", + index_table="ifcb_cnn_classification_index", + ) diff --git a/src/improv/plugins/ifcb_features.py b/src/improv/plugins/ifcb/features.py similarity index 100% rename from src/improv/plugins/ifcb_features.py rename to src/improv/plugins/ifcb/features.py diff --git a/src/improv/plugins/ifcb_id.py b/src/improv/plugins/ifcb/image_id.py similarity index 100% rename from src/improv/plugins/ifcb_id.py rename to src/improv/plugins/ifcb/image_id.py diff --git a/src/improv/plugins/sample_context.py b/src/improv/plugins/sample_context.py index 95c5566..0b772ef 100644 --- a/src/improv/plugins/sample_context.py +++ b/src/improv/plugins/sample_context.py @@ -11,6 +11,8 @@ from pydantic import BaseModel +from improv.store.dedup import dedup_rows + if TYPE_CHECKING: from amplify_db_utils import ColumnarStore from improv.models.provenance import ProvenanceEnvelope @@ -56,3 +58,30 @@ def extract_index_record(self, envelope: "ProvenanceEnvelope") -> dict | None: "year": envelope.year, "month": envelope.month, } + + # --- SampleQueryPlugin capability -------------------------------------- + + def write_index( + self, + store: "ColumnarStore", + records: list[SampleIndexRecord], + ) -> None: + """Batch-write sample index records.""" + store.write(self.index_table, [r.model_dump() for r in records]) + + def query_by_sample_id( + self, + store: "ColumnarStore", + sample_id: str, + source: str | None = None, + ) -> list[str]: + """Return image_ids associated with a sample ID. + + An image may have multiple sample_index rows (one per naming scheme / + source). Pass source to restrict to a specific naming authority. + """ + filters: dict = {"sample_id": sample_id} + if source is not None: + filters["source"] = source + rows = dedup_rows(store.read(self.index_table, filters=filters)) + return [row["image_id"] for row in rows] diff --git a/src/improv/service.py b/src/improv/service.py index ac44855..4e65623 100644 --- a/src/improv/service.py +++ b/src/improv/service.py @@ -19,16 +19,19 @@ from __future__ import annotations +from collections import defaultdict from datetime import datetime, timezone from typing import TYPE_CHECKING, Iterator +from pydantic import BaseModel + +from improv.plugins import SpatialQueryPlugin from improv.store.images import ( bulk_get_images, get_image, get_images, write_images, ) -from improv.store.indexes import get_images_by_sample_id, query_spatial from improv.store.provenance import ( get_provenance, get_provenance_by_kind, @@ -44,7 +47,14 @@ from improv.ids import ImageIdParser from improv.models.image import ImageRecord from improv.models.provenance import ProvenanceEnvelope - from improv.oltp.models import Dataset, DatasetSpan, IngestTask, Instrument, Sample + from improv.oltp.models import ( + ClassifierTaxonomy, + Dataset, + DatasetSpan, + IngestTask, + Instrument, + Sample, + ) from improv.plugins import ProvenancePlugin @@ -118,11 +128,14 @@ def ingest_provenance( before writing, so that plugin index records have consistent keys. For each record, if a registered plugin handles its kind, calls - extract_index_record and writes the result to the plugin's index table. - Records with no registered plugin are stored with no index write. - - Pass ``write_indexes=False`` to skip per-record index writes entirely. - Batch producers should do so and perform their own batched store.write. + extract_index_record and collects the result. Index records are grouped + by target table and written with one batched store.write per table, so a + batch of N provenance records costs 1 + (number of distinct index + tables) writes rather than 1 + N. Records with no registered plugin are + stored with no index write. + + Pass ``write_indexes=False`` to skip index writes entirely. Batch + producers may do so and perform their own batched store.write. """ enriched = self.enrich_envelopes(records) @@ -131,13 +144,23 @@ def ingest_provenance( if not write_indexes: return + # Collect all index records first (extraction may raise on bad input, + # e.g. an out-of-range winner_index), then write each table's batch in + # one call. Grouping by table also merges any plugins that share one. + index_batches: dict[str, list[dict]] = defaultdict(list) for envelope in enriched: plugin = self._plugins.get(envelope.kind) if plugin is None: continue index_record = plugin.extract_index_record(envelope) if index_record is not None and plugin.index_table is not None: - self._store.write(plugin.index_table, [index_record]) + schema = plugin.index_schema + if isinstance(schema, type) and issubclass(schema, BaseModel): + index_record = schema(**index_record).model_dump() + index_batches[plugin.index_table].append(index_record) + + for table, index_records in index_batches.items(): + self._store.write(table, index_records) # ------------------------------------------------------------------ # Retrieval @@ -162,6 +185,13 @@ def get_provenance( ) return get_provenance(self._store, image_id, self._parsers, instrument) + def _spatial_plugin(self) -> "SpatialQueryPlugin | None": + """Return the first registered plugin advertising the spatial query capability.""" + for plugin in self._plugins.values(): + if isinstance(plugin, SpatialQueryPlugin): + return plugin + return None + def query_images( self, instrument: str, @@ -174,8 +204,8 @@ def query_images( ) -> "list[ImageRecord]": """Return images matching time range ± spatial bounding box. - When spatial bounds are provided, filters via the geolocation index - and intersects with the time-range result. + When spatial bounds are provided, filters via a registered + SpatialQueryPlugin and intersects with the time-range result. """ if any(v is not None for v in [lat_min, lat_max, lon_min, lon_max]): if None in (lat_min, lat_max, lon_min, lon_max): @@ -183,8 +213,13 @@ def query_images( "All four spatial bounds (lat_min, lat_max, lon_min, lon_max) " "must be provided together." ) + spatial = self._spatial_plugin() + if spatial is None: + raise RuntimeError( + "Spatial query requested but no SpatialQueryPlugin is registered." + ) geo_ids = set( - query_spatial( + spatial.query_spatial( self._store, lat_min, # type: ignore[arg-type] lat_max, # type: ignore[arg-type] @@ -421,6 +456,82 @@ def get_ingest_task(self, task_id: str) -> "IngestTask | None": from improv.oltp.queries import get_ingest_task return get_ingest_task(self._session, task_id) + # ------------------------------------------------------------------ + # Classifier taxonomy (label-map) + # ------------------------------------------------------------------ + + def register_classifier_taxonomy( + self, + classifier: str, + model_version: str, + class_names: "list[str]", + ) -> "tuple[ClassifierTaxonomy, bool]": + """Register a classifier's ordered label-map; return (taxonomy, created). + + classifier is the plugin kind (e.g. "ifcb_cnn_classification"). + Idempotent on (classifier, model_version). Any change to the class list + or order must use a new model_version. + """ + from improv.oltp.queries import ( + get_classifier_taxonomy, + register_classifier_taxonomy, + ) + existing = get_classifier_taxonomy(self._session, classifier, model_version) + if existing is not None: + return existing, False + taxonomy = register_classifier_taxonomy( + self._session, classifier, model_version, class_names, + now=datetime.now(tz=timezone.utc), + ) + self._session.commit() + return taxonomy, True + + def get_classifier_taxonomy( + self, classifier: str, model_version: str + ) -> "ClassifierTaxonomy | None": + from improv.oltp.queries import get_classifier_taxonomy + return get_classifier_taxonomy(self._session, classifier, model_version) + + def get_latest_classifier_taxonomy( + self, classifier: str + ) -> "ClassifierTaxonomy | None": + from improv.oltp.queries import get_latest_classifier_taxonomy + return get_latest_classifier_taxonomy(self._session, classifier) + + def decode_classification( + self, + classifier: str, + model_version: str, + scores: "list[float]", + winner_index: int, + ) -> dict: + """Turn a positional score vector into names via the exact-version label-map. + + Returns ``{"winner": , "scores": {name: score, ...}}``. + Raises ValueError if no taxonomy is registered for + (classifier, model_version) or its length does not match the vector. + """ + taxonomy = self.get_classifier_taxonomy(classifier, model_version) + if taxonomy is None: + raise ValueError( + f"No taxonomy registered for classifier={classifier!r} " + f"model_version={model_version!r}." + ) + names = taxonomy.class_names + if len(names) != len(scores): + raise ValueError( + f"Taxonomy for {classifier!r}/{model_version!r} has {len(names)} " + f"classes but score vector has {len(scores)}." + ) + if not 0 <= winner_index < len(names): + raise ValueError( + f"winner_index {winner_index} out of range for {len(names)} classes." + ) + return { + "winner": names[winner_index], + "scores": {name: score for name, score in zip(names, scores)}, + } + # ------------------------------------------------------------------ # Binary products # ------------------------------------------------------------------ diff --git a/src/improv/store/dedup.py b/src/improv/store/dedup.py new file mode 100644 index 0000000..bb1a5fb --- /dev/null +++ b/src/improv/store/dedup.py @@ -0,0 +1,40 @@ +"""Read-time deduplication for the append-only columnar store. + +Writes are append-only across every backend, so a retried write re-appends +rows. Dedup at read restores idempotency: byte-identical rows collapse to one, +while genuinely different rows are retained. + +Index rows are deterministic projections of provenance with no write-time-varying +column, so an exact retry produces a byte-identical row and full-row equality is +a sound identity. (Provenance rows carry a write-time ``written_at`` and an +opaque JSON ``data`` blob, so they dedup on an explicit identity key instead — +see ``store.provenance``.) +""" + +from __future__ import annotations + +from collections.abc import Iterable +from typing import Sequence + + +def dedup_rows( + rows: Iterable[dict], + exclude: Sequence[str] = (), +) -> list[dict]: + """Collapse rows identical across all columns (minus ``exclude``). + + Order-preserving; the first occurrence of each identity wins. All retained + column values must be hashable (scalar columns are — index rows have no + nested values). + """ + exclude_set = set(exclude) + seen: set[tuple] = set() + out: list[dict] = [] + for row in rows: + # Keys are unique per row, so sorting by key never compares values — + # values need only be hashable, not orderable. + key = tuple(sorted((k, v) for k, v in row.items() if k not in exclude_set)) + if key not in seen: + seen.add(key) + out.append(row) + return out diff --git a/src/improv/store/indexes.py b/src/improv/store/indexes.py deleted file mode 100644 index 295f022..0000000 --- a/src/improv/store/indexes.py +++ /dev/null @@ -1,91 +0,0 @@ -"""Columnar store operations for service-owned index tables. - -Cross-index joins: always use select="right". -When joining geolocation_index or sample_index against images, pass -select="right" to store.join(). Both tables share partition key columns; -select="both" returns duplicate columns. select="right" returns only the -payload (images) table's columns. -""" - -from __future__ import annotations - -from datetime import datetime, timezone -from typing import TYPE_CHECKING - -from improv.plugins.geolocation import GeoLocationIndexRecord -from improv.plugins.sample_context import SampleIndexRecord - -if TYPE_CHECKING: - from amplify_db_utils import ColumnarStore - - -def _as_utc(ts: datetime) -> datetime: - if ts.tzinfo is None: - return ts.replace(tzinfo=timezone.utc) - return ts - - -def write_geolocation( - store: "ColumnarStore", - records: list[GeoLocationIndexRecord], -) -> None: - """Write geolocation index records.""" - store.write("geolocation_index", [r.model_dump() for r in records]) - - -def query_spatial( - store: "ColumnarStore", - lat_min: float, - lat_max: float, - lon_min: float, - lon_max: float, - time_start: datetime | None = None, - time_end: datetime | None = None, -) -> list[str]: - """Return image_ids within a lat/lon bounding box, optionally scoped by time. - - Time filtering uses the year/month partition keys for efficient pruning - (month-level precision). Callers may apply exact timestamp filtering - against the images table afterward. - """ - filters: dict = { - "lat": {"gte": lat_min, "lte": lat_max}, - "lon": {"gte": lon_min, "lte": lon_max}, - } - # Approximate time scoping via partition keys when both bounds are in the same year - if time_start is not None and time_end is not None: - time_start = _as_utc(time_start) - time_end = _as_utc(time_end) - if time_start.year == time_end.year: - filters["year"] = time_start.year - if time_start.month == time_end.month: - filters["month"] = time_start.month - - return [ - row["image_id"] - for row in store.read("geolocation_index", filters=filters) - ] - - -def write_sample_index( - store: "ColumnarStore", - records: list[SampleIndexRecord], -) -> None: - """Write sample index records.""" - store.write("sample_index", [r.model_dump() for r in records]) - - -def get_images_by_sample_id( - store: "ColumnarStore", - sample_id: str, - source: str | None = None, -) -> list[str]: - """Return image_ids associated with a sample ID. - - An image may have multiple sample_index rows (one per naming scheme / source). - Pass source to restrict to a specific naming authority. - """ - filters: dict = {"sample_id": sample_id} - if source is not None: - filters["source"] = source - return [row["image_id"] for row in store.read("sample_index", filters=filters)] diff --git a/src/improv/store/provenance.py b/src/improv/store/provenance.py index d733428..a22b354 100644 --- a/src/improv/store/provenance.py +++ b/src/improv/store/provenance.py @@ -14,9 +14,16 @@ from datetime import datetime, timezone from typing import TYPE_CHECKING +from improv.hashing import canonical_data_hash from improv.ids import ImageIdParser, make_partition_keys from improv.models.provenance import ProvenanceEnvelope +# Columns that define provenance-row identity for idempotent read-time dedup. +# Two rows agreeing on all four are the same fact; a retry re-appends an +# identical row that collapses here. `written_at` is deliberately excluded so +# retries (which differ only in write time) are treated as duplicates. +_IDENTITY_KEYS = ("image_id", "kind", "source", "data_hash") + if TYPE_CHECKING: from amplify_db_utils import ColumnarStore @@ -58,13 +65,39 @@ def _row_to_envelope(row: dict) -> ProvenanceEnvelope: return ProvenanceEnvelope(**r) +def _dedup_identity(rows: list[dict]) -> list[dict]: + """Collapse rows sharing an identity key, keeping one per identity. + + Backend-neutral (pure Python) so idempotency behaves identically on every + columnar backend — the append-only WORM stores (VAST DB) and the + overwrite-capable ones (DuckDB/Parquet) alike. Rows sharing an identity key + are byte-identical except for `written_at`, so which copy is kept is + immaterial; first occurrence wins. + """ + seen: dict[tuple, dict] = {} + for row in rows: + key = tuple(row.get(k) for k in _IDENTITY_KEYS) + if key not in seen: + seen[key] = row + return list(seen.values()) + + def write_provenance( store: "ColumnarStore", records: list[ProvenanceEnvelope], parsers: list[ImageIdParser], ) -> None: - """Append provenance records to the columnar store.""" + """Append provenance records to the columnar store. + + Stamps each row with its canonical `data_hash` (identity) and a `written_at` + timestamp before writing. Writes are append-only; deduplication happens at + read time via the (image_id, kind, source, data_hash) identity key. + """ + now = datetime.now(timezone.utc) dicts = [_enrich(r.model_dump(), parsers) for r in records] + for d in dicts: + d["data_hash"] = canonical_data_hash(d["data"]) + d["written_at"] = now store.write("provenance", dicts) @@ -77,7 +110,8 @@ def get_provenance( """Return all provenance records for an image.""" keys = make_partition_keys(image_id, parsers, instrument_hint) filters: dict = {"image_id": image_id, **keys} - return [_row_to_envelope(row) for row in store.read("provenance", filters=filters)] + rows = _dedup_identity(list(store.read("provenance", filters=filters))) + return [_row_to_envelope(row) for row in rows] def get_provenance_by_kind( @@ -90,4 +124,5 @@ def get_provenance_by_kind( """Return provenance records of a specific kind for an image.""" keys = make_partition_keys(image_id, parsers, instrument_hint) filters: dict = {"image_id": image_id, "kind": kind, **keys} - return [_row_to_envelope(row) for row in store.read("provenance", filters=filters)] + rows = _dedup_identity(list(store.read("provenance", filters=filters))) + return [_row_to_envelope(row) for row in rows] diff --git a/tests/test_api/conftest.py b/tests/test_api/conftest.py index 83551e9..79ce2d4 100644 --- a/tests/test_api/conftest.py +++ b/tests/test_api/conftest.py @@ -15,14 +15,14 @@ from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker -from improv.api.routers import blobs, datasets, images, ingest_tasks, instruments, provenance, samples +from improv.api.routers import blobs, classification, datasets, images, ingest_tasks, instruments, provenance, samples from improv.oltp.models import Base from improv.plugins.geolocation import GeoLocationPlugin from improv.plugins.sample_context import SampleContextPlugin from improv.service import ImageService from improv.store.tables import register_service_tables -from tests.conftest import AlphaParser +from tests.conftest import AlphaParser, BetaParser @pytest.fixture @@ -43,7 +43,7 @@ def client(tmp_path): session = Session() plugins = [GeoLocationPlugin(), SampleContextPlugin()] - parsers = [AlphaParser()] + parsers = [AlphaParser(), BetaParser()] register_service_tables(store, plugins) service = ImageService( @@ -62,6 +62,7 @@ def client(tmp_path): app.include_router(blobs.router) app.include_router(datasets.router) app.include_router(ingest_tasks.router) + app.include_router(classification.router) with TestClient(app) as c: yield c diff --git a/tests/test_api/test_classification.py b/tests/test_api/test_classification.py new file mode 100644 index 0000000..c0ab8cb --- /dev/null +++ b/tests/test_api/test_classification.py @@ -0,0 +1,173 @@ +"""API tests for classifier taxonomy + classification-decode endpoints.""" + +from __future__ import annotations + +from datetime import datetime, timezone + + +TS = datetime(2024, 1, 15, 12, 0, tzinfo=timezone.utc) +IMAGE_ID = "ALPHA_20240115T120000_001" +CLASSIFIER = "ifcb_cnn_classification" + + +def _register(client, model_version, class_names): + return client.post( + f"/classifiers/{CLASSIFIER}/taxonomies", + json={"model_version": model_version, "class_names": class_names}, + ) + + +def _ingest_classification(client, model_version, scores, winner_index): + client.post( + "/images/ingest", + json=[{"image_id": IMAGE_ID, "timestamp": TS.isoformat(), "instrument": "ALPHA"}], + ) + return client.post( + f"/images/{IMAGE_ID}/provenance?instrument=ALPHA", + json={ + "kind": CLASSIFIER, + "source": "test-classifier", + "timestamp": TS.isoformat(), + "data": { + "run_id": "run-1", + "model_version": model_version, + "scores": scores, + "winner_index": winner_index, + "winner_score": scores[winner_index], + }, + }, + ) + + +# --------------------------------------------------------------------------- +# Taxonomy registration + lookup +# --------------------------------------------------------------------------- + +def test_register_taxonomy(client): + resp = _register(client, "v4", ["Ceratium", "Chaetoceros", "Dinophysis"]) + assert resp.status_code == 201 + body = resp.json() + assert body["classifier"] == CLASSIFIER + assert body["model_version"] == "v4" + assert body["class_names"] == ["Ceratium", "Chaetoceros", "Dinophysis"] + + +def test_register_taxonomy_idempotent(client): + _register(client, "v4", ["A", "B"]) + resp = _register(client, "v4", ["A", "B"]) + assert resp.status_code == 409 + # Still exactly one row, unchanged. + got = client.get(f"/classifiers/{CLASSIFIER}/taxonomies/v4").json() + assert got["class_names"] == ["A", "B"] + + +def test_get_taxonomy_by_version(client): + _register(client, "v4", ["Ceratium", "Chaetoceros"]) + resp = client.get(f"/classifiers/{CLASSIFIER}/taxonomies/v4") + assert resp.status_code == 200 + assert resp.json()["class_names"] == ["Ceratium", "Chaetoceros"] + + +def test_get_taxonomy_unknown_version(client): + resp = client.get(f"/classifiers/{CLASSIFIER}/taxonomies/nope") + assert resp.status_code == 404 + + +def test_get_latest_taxonomy(client): + _register(client, "v1", ["A"]) + _register(client, "v2", ["A", "B"]) + resp = client.get(f"/classifiers/{CLASSIFIER}/taxonomies/latest") + assert resp.status_code == 200 + assert resp.json()["model_version"] == "v2" + + +def test_get_latest_taxonomy_none(client): + resp = client.get(f"/classifiers/{CLASSIFIER}/taxonomies/latest") + assert resp.status_code == 404 + + +# --------------------------------------------------------------------------- +# (A) stateless decode +# --------------------------------------------------------------------------- + +def test_decode(client): + _register(client, "v4", ["Ceratium", "Chaetoceros", "Dinophysis"]) + resp = client.post( + f"/classifiers/{CLASSIFIER}/decode", + json={"model_version": "v4", "scores": [0.1, 0.7, 0.2], "winner_index": 1}, + ) + assert resp.status_code == 200 + body = resp.json() + assert body["winner"] == "Chaetoceros" + assert body["scores"] == {"Ceratium": 0.1, "Chaetoceros": 0.7, "Dinophysis": 0.2} + + +def test_decode_length_mismatch(client): + _register(client, "v4", ["A", "B", "C"]) + resp = client.post( + f"/classifiers/{CLASSIFIER}/decode", + json={"model_version": "v4", "scores": [0.5, 0.5], "winner_index": 0}, + ) + assert resp.status_code == 422 + + +def test_decode_unknown_taxonomy(client): + resp = client.post( + f"/classifiers/{CLASSIFIER}/decode", + json={"model_version": "v4", "scores": [0.5, 0.5], "winner_index": 0}, + ) + assert resp.status_code == 404 + + +# --------------------------------------------------------------------------- +# (B) decoded classification read +# --------------------------------------------------------------------------- + +def test_decoded_classification_read(client): + _register(client, "v4", ["Ceratium", "Chaetoceros", "Dinophysis"]) + _ingest_classification(client, "v4", [0.1, 0.7, 0.2], winner_index=1) + + resp = client.get( + f"/images/{IMAGE_ID}/classification?kind={CLASSIFIER}&instrument=ALPHA" + ) + assert resp.status_code == 200 + records = resp.json() + assert len(records) == 1 + assert records[0]["winner"] == "Chaetoceros" + assert records[0]["scores"]["Chaetoceros"] == 0.7 + + +def test_decoded_read_uses_records_own_version(client): + # Two versions with DIFFERENT orderings. Each stored record must decode + # against its own model_version, not "latest". + _register(client, "v1", ["Ceratium", "Chaetoceros"]) + _register(client, "v2", ["Chaetoceros", "Ceratium"]) # reversed order + + client.post( + "/images/ingest", + json=[{"image_id": IMAGE_ID, "timestamp": TS.isoformat(), "instrument": "ALPHA"}], + ) + # winner_index=0 under each version → different names because orderings differ. + for mv in ("v1", "v2"): + client.post( + f"/images/{IMAGE_ID}/provenance?instrument=ALPHA", + json={ + "kind": CLASSIFIER, + "source": "test", + "timestamp": TS.isoformat(), + "data": { + "run_id": f"run-{mv}", + "model_version": mv, + "scores": [0.9, 0.1], + "winner_index": 0, + "winner_score": 0.9, + }, + }, + ) + + records = client.get( + f"/images/{IMAGE_ID}/classification?kind={CLASSIFIER}&instrument=ALPHA" + ).json() + winners = {r["winner"] for r in records} + # v1[0]=Ceratium, v2[0]=Chaetoceros — proves per-record decoding, not latest. + assert winners == {"Ceratium", "Chaetoceros"} diff --git a/tests/test_api/test_client.py b/tests/test_api/test_client.py index 2dc0843..900d000 100644 --- a/tests/test_api/test_client.py +++ b/tests/test_api/test_client.py @@ -260,3 +260,40 @@ def test_get_ingest_task(improv_client): def test_get_ingest_task_not_found(improv_client): assert improv_client.get_ingest_task("NONEXISTENT") is None + + +# ------------------------------------------------------------------ +# Classifier taxonomy +# ------------------------------------------------------------------ + +CLASSIFIER = "ifcb_cnn_classification" + + +def test_register_classifier_taxonomy(improv_client): + taxonomy, created = improv_client.register_classifier_taxonomy( + CLASSIFIER, "v4", ["Ceratium", "Chaetoceros"] + ) + assert created is True + assert taxonomy["classifier"] == CLASSIFIER + assert taxonomy["model_version"] == "v4" + assert taxonomy["class_names"] == ["Ceratium", "Chaetoceros"] + + +def test_register_classifier_taxonomy_idempotent(improv_client): + improv_client.register_classifier_taxonomy(CLASSIFIER, "v4", ["A", "B"]) + taxonomy, created = improv_client.register_classifier_taxonomy( + CLASSIFIER, "v4", ["A", "B"] + ) + assert created is False + assert taxonomy["class_names"] == ["A", "B"] + + +def test_get_classifier_taxonomy(improv_client): + improv_client.register_classifier_taxonomy(CLASSIFIER, "v4", ["A", "B", "C"]) + taxonomy = improv_client.get_classifier_taxonomy(CLASSIFIER, "v4") + assert taxonomy is not None + assert taxonomy["class_names"] == ["A", "B", "C"] + + +def test_get_classifier_taxonomy_not_found(improv_client): + assert improv_client.get_classifier_taxonomy(CLASSIFIER, "nope") is None diff --git a/tests/test_api/test_provenance.py b/tests/test_api/test_provenance.py index 244a118..b2ba43e 100644 --- a/tests/test_api/test_provenance.py +++ b/tests/test_api/test_provenance.py @@ -71,3 +71,94 @@ def test_get_provenance_by_kind(client): assert resp.status_code == 200 records = resp.json() assert all(r["kind"] == "features" for r in records) + + +def test_provenance_response_includes_year_month(client): + ingest_image(client) + client.post( + f"/images/{IMAGE_ID}/provenance?instrument=ALPHA", + json={ + "kind": "features", + "source": "test", + "timestamp": TS.isoformat(), + "data": {"area": 100.0}, + }, + ) + records = client.get(f"/images/{IMAGE_ID}/provenance?instrument=ALPHA").json() + assert records[0]["year"] == TS.year + assert records[0]["month"] == TS.month + + +# --------------------------------------------------------------------------- +# Batch ingest +# --------------------------------------------------------------------------- + +IMAGE_ID_2 = "ALPHA_20240115T120000_002" + + +def _batch_record(image_id, area): + return { + "image_id": image_id, + "kind": "features", + "source": "test-batch", + "timestamp": TS.isoformat(), + "data": {"area": area}, + } + + +def test_batch_single_instrument_roundtrip(client): + client.post( + "/images/ingest", + json=[ + {"image_id": IMAGE_ID, "timestamp": TS.isoformat(), "instrument": "ALPHA"}, + {"image_id": IMAGE_ID_2, "timestamp": TS.isoformat(), "instrument": "ALPHA"}, + ], + ) + + resp = client.post( + "/images/provenance/batch", + json={"records": [ + _batch_record(IMAGE_ID, 100.0), + _batch_record(IMAGE_ID_2, 200.0), + ]}, + ) + assert resp.status_code == 201 + assert resp.json()["ingested"] == 2 + + r1 = client.get(f"/images/{IMAGE_ID}/provenance?instrument=ALPHA").json() + r2 = client.get(f"/images/{IMAGE_ID_2}/provenance?instrument=ALPHA").json() + assert r1[0]["data"]["area"] == 100.0 + assert r2[0]["data"]["area"] == 200.0 + # Each record landed under its own image_id, not an empty shared id. + assert r1[0]["image_id"] == IMAGE_ID + assert r2[0]["image_id"] == IMAGE_ID_2 + + +def test_batch_multi_instrument_rejected(client): + resp = client.post( + "/images/provenance/batch", + json={"records": [ + _batch_record("ALPHA_20240115T120000_001", 100.0), + _batch_record("BETA-20240115T120000-001", 200.0), + ]}, + ) + assert resp.status_code == 422 + # All-or-nothing: nothing written for either record's image_id. + assert client.get( + "/images/ALPHA_20240115T120000_001/provenance?instrument=ALPHA" + ).json() == [] + assert client.get( + "/images/BETA-20240115T120000-001/provenance?instrument=BETA" + ).json() == [] + + +def test_batch_unparseable_image_id_uses_instrument_hint(client): + resp = client.post( + "/images/provenance/batch?instrument=ALPHA", + json={"records": [_batch_record("no-parser-match", 100.0)]}, + ) + assert resp.status_code == 201 + records = client.get( + "/images/no-parser-match/provenance?instrument=ALPHA" + ).json() + assert records[0]["data"]["area"] == 100.0 diff --git a/tests/test_ifcb_id.py b/tests/test_ifcb_id.py index 149f205..24ddd9f 100644 --- a/tests/test_ifcb_id.py +++ b/tests/test_ifcb_id.py @@ -7,7 +7,7 @@ import pytest from improv.ids import make_partition_keys -from improv.plugins.ifcb_id import IFCBImageIdParser +from improv.plugins.ifcb import IFCBImageIdParser @pytest.fixture diff --git a/tests/test_plugins.py b/tests/test_plugins.py index ac7538a..d96efde 100644 --- a/tests/test_plugins.py +++ b/tests/test_plugins.py @@ -4,7 +4,13 @@ from datetime import datetime, timezone +import pytest + from improv.models.provenance import ProvenanceEnvelope +from improv.plugins.classification import ( + MachineClassificationIndexRecord, + MachineClassificationPlugin, +) from improv.plugins.geolocation import GeoLocationIndexRecord, GeoLocationPlugin from improv.plugins.sample_context import SampleContextPlugin, SampleIndexRecord @@ -115,3 +121,138 @@ def test_sample_context_plugin_creates_table(store): plugin = SampleContextPlugin() plugin.create_index(store) plugin.create_index(store) # idempotent + + +# --------------------------------------------------------------------------- +# MachineClassificationPlugin (generic, any classifier) +# --------------------------------------------------------------------------- + +def test_classification_plugin_parameterized(): + plugin = MachineClassificationPlugin( + kind="ecotaxa_cnn", index_table="ecotaxa_cnn_index" + ) + assert plugin.kind == "ecotaxa_cnn" + assert plugin.index_table == "ecotaxa_cnn_index" + assert plugin.index_schema is MachineClassificationIndexRecord + assert "model_version" in plugin.partition_by + + +def test_classification_plugin_extract_narrow_index(): + """Non-IFCB classifier: index carries winner_index only, no names/scores.""" + plugin = MachineClassificationPlugin( + kind="ecotaxa_cnn", index_table="ecotaxa_cnn_index" + ) + env = make_env( + "ecotaxa_cnn", + { + "run_id": "run-1", + "model_version": "ecotaxa-cnn-v4", + "scores": [0.9, 0.1], + "winner_index": 0, + "winner_score": 0.9, + }, + ) + record = plugin.extract_index_record(env) + assert record is not None + assert record["winner_index"] == 0 + assert record["winner_score"] == 0.9 + assert record["run_id"] == "run-1" + assert record["instrument"] == "ALPHA" + assert record["year"] == 2024 + # positional score vector and names are NOT indexed + assert "scores" not in record + assert "winner" not in record + + +def test_classification_plugin_winner_index_out_of_range(): + plugin = MachineClassificationPlugin(kind="ecotaxa_cnn") + env = make_env( + "ecotaxa_cnn", + { + "run_id": "run-1", + "model_version": "v1", + "scores": [0.9, 0.1], + "winner_index": 2, # out of range + "winner_score": 0.9, + }, + ) + with pytest.raises(ValueError): + plugin.extract_index_record(env) + + +def test_classification_plugin_creates_table(store): + plugin = MachineClassificationPlugin( + kind="ecotaxa_cnn", index_table="ecotaxa_cnn_index" + ) + plugin.create_index(store) + plugin.create_index(store) # idempotent + + +def test_ifcb_cnn_preset_backcompat(): + from improv.plugins.ifcb import ( + IFCBCNNClassificationIndexRecord, + IFCBCNNClassificationPlugin, + ) + plugin = IFCBCNNClassificationPlugin() + assert plugin.kind == "ifcb_cnn_classification" + assert plugin.index_table == "ifcb_cnn_classification_index" + assert IFCBCNNClassificationIndexRecord is MachineClassificationIndexRecord + assert isinstance(plugin, MachineClassificationPlugin) + + +# --------------------------------------------------------------------------- +# Index idempotency — retries collapse at read (matches provenance dedup) +# --------------------------------------------------------------------------- + +def _geo_index_record(source="nav_v1", version="1.0"): + return GeoLocationIndexRecord( + image_id="ALPHA_20240115T120000_001", + lat=41.5, + lon=-70.5, + source=source, + version=version, + computed_at=datetime(2024, 1, 15, 12, 0, tzinfo=timezone.utc), + instrument="ALPHA", + year=2024, + month=1, + ) + + +def test_geolocation_query_dedups_retries(store): + plugin = GeoLocationPlugin() + plugin.create_index(store) + rec = _geo_index_record() + plugin.write_index(store, [rec]) + plugin.write_index(store, [rec]) # retry re-appends an identical row + + ids = plugin.query_spatial(store, 40.0, 42.0, -71.0, -70.0) + assert ids == ["ALPHA_20240115T120000_001"] + + +def test_geolocation_query_keeps_distinct_rows(store): + """Rows differing in any column (e.g. source/version) are distinct facts.""" + plugin = GeoLocationPlugin() + plugin.create_index(store) + plugin.write_index(store, [_geo_index_record(source="nav_v1", version="1.0")]) + plugin.write_index(store, [_geo_index_record(source="nav_v2", version="2.0")]) + + ids = plugin.query_spatial(store, 40.0, 42.0, -71.0, -70.0) + assert len(ids) == 2 # both retained; not collapsed + + +def test_sample_context_query_dedups_retries(store): + plugin = SampleContextPlugin() + plugin.create_index(store) + rec = SampleIndexRecord( + image_id="ALPHA_20240115T120000_001", + sample_id="SAMPLE_001", + source="ifcb_system", + instrument="ALPHA", + year=2024, + month=1, + ) + plugin.write_index(store, [rec]) + plugin.write_index(store, [rec]) # retry + + ids = plugin.query_by_sample_id(store, "SAMPLE_001") + assert ids == ["ALPHA_20240115T120000_001"] diff --git a/tests/test_service.py b/tests/test_service.py index dbb2984..a09367a 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -83,8 +83,8 @@ def test_ingest_provenance_plugin_dual_write(service, store_with_tables): env = geo_envelope(record.image_id, lat=41.5, lon=-70.5) service.ingest_provenance([env]) - from improv.store.indexes import query_spatial - ids = query_spatial(store_with_tables, 41.0, 42.0, -71.0, -70.0) + from improv.plugins.geolocation import GeoLocationPlugin + ids = GeoLocationPlugin().query_spatial(store_with_tables, 41.0, 42.0, -71.0, -70.0) assert record.image_id in ids @@ -170,6 +170,65 @@ def test_get_dataset_images(service, session): assert r2.image_id not in ids +# --------------------------------------------------------------------------- +# Classifier taxonomy (label-map) +# --------------------------------------------------------------------------- + +def test_register_classifier_taxonomy_idempotent(service): + tax, created = service.register_classifier_taxonomy( + "ifcb_cnn_classification", "v4", ["Ceratium", "Chaetoceros"] + ) + assert created is True + assert tax.class_names == ["Ceratium", "Chaetoceros"] + + again, created2 = service.register_classifier_taxonomy( + "ifcb_cnn_classification", "v4", ["ignored"] + ) + assert created2 is False + assert again.taxonomy_id == tax.taxonomy_id + assert again.class_names == ["Ceratium", "Chaetoceros"] # unchanged + + +def test_get_classifier_taxonomy_exact_version(service): + service.register_classifier_taxonomy("clf", "v1", ["A", "B"]) + service.register_classifier_taxonomy("clf", "v2", ["A", "B", "C"]) + + v1 = service.get_classifier_taxonomy("clf", "v1") + v2 = service.get_classifier_taxonomy("clf", "v2") + assert v1.class_names == ["A", "B"] + assert v2.class_names == ["A", "B", "C"] + assert service.get_classifier_taxonomy("clf", "missing") is None + + +def test_get_latest_classifier_taxonomy(service): + service.register_classifier_taxonomy("clf", "v1", ["A"]) + service.register_classifier_taxonomy("clf", "v2", ["A", "B"]) + latest = service.get_latest_classifier_taxonomy("clf") + assert latest.model_version == "v2" + + +def test_decode_classification_roundtrip(service): + service.register_classifier_taxonomy( + "ifcb_cnn_classification", "v4", ["Ceratium", "Chaetoceros", "Dinophysis"] + ) + decoded = service.decode_classification( + "ifcb_cnn_classification", "v4", [0.1, 0.7, 0.2], winner_index=1 + ) + assert decoded["winner"] == "Chaetoceros" + assert decoded["scores"] == {"Ceratium": 0.1, "Chaetoceros": 0.7, "Dinophysis": 0.2} + + +def test_decode_classification_unknown_version(service): + with pytest.raises(ValueError): + service.decode_classification("clf", "nope", [0.5, 0.5], winner_index=0) + + +def test_decode_classification_length_mismatch(service): + service.register_classifier_taxonomy("clf", "v1", ["A", "B"]) + with pytest.raises(ValueError): + service.decode_classification("clf", "v1", [0.5, 0.3, 0.2], winner_index=0) + + # --------------------------------------------------------------------------- # Blob key # --------------------------------------------------------------------------- @@ -196,3 +255,27 @@ def test_get_blob_key_none_when_no_blob(service): record = alpha_record("001") service.ingest_images([record]) assert service.get_blob_key(record.image_id) is None + + +def test_batch_provenance_writes_one_index_call_per_table(service, monkeypatch): + """A batch of N same-kind records → one batched index write, not N.""" + recs = [alpha_record(f"00{i}") for i in range(1, 4)] + service.ingest_images(recs) + envs = [ + geo_envelope(r.image_id, lat=41.0 + i * 0.1, lon=-70.0) + for i, r in enumerate(recs) + ] + + calls: list[tuple[str, int]] = [] + orig = service._store.write + + def spy(table, records, *args, **kwargs): + calls.append((table, len(records))) + return orig(table, records, *args, **kwargs) + + monkeypatch.setattr(service._store, "write", spy) + service.ingest_provenance(envs) + + assert ("provenance", 3) in calls + # geolocation_index written once with all 3 rows, not three single-row writes + assert [c for c in calls if c[0] == "geolocation_index"] == [("geolocation_index", 3)] diff --git a/tests/test_store_provenance.py b/tests/test_store_provenance.py index 76071d0..a6e9ba4 100644 --- a/tests/test_store_provenance.py +++ b/tests/test_store_provenance.py @@ -90,3 +90,69 @@ def test_not_found_returns_empty(store_with_tables, parsers): store_with_tables, "ALPHA_20240115T120000_999", parsers ) assert results == [] + + +def test_retry_same_record_is_idempotent(store_with_tables, parsers): + """Re-writing an identical record collapses at read time (retry safety).""" + image_id = "ALPHA_20240115T120000_001" + env = make_envelope(image_id, "geolocation", {"lat": 41.5, "lon": -70.0}) + write_provenance(store_with_tables, [env], parsers) + write_provenance(store_with_tables, [env], parsers) # retry + + results = get_provenance(store_with_tables, image_id, parsers) + assert len(results) == 1 + assert results[0].data == {"lat": 41.5, "lon": -70.0} + + +def test_distinct_payloads_both_kept(store_with_tables, parsers): + """Same (image_id, kind, source) but different data are distinct facts.""" + image_id = "ALPHA_20240115T120000_001" + write_provenance( + store_with_tables, + [make_envelope(image_id, "classification", {"model": "cnn_v1", "score": 0.9})], + parsers, + ) + write_provenance( + store_with_tables, + [make_envelope(image_id, "classification", {"model": "cnn_v2", "score": 0.8})], + parsers, + ) + + results = get_provenance_by_kind(store_with_tables, image_id, "classification", parsers) + assert len(results) == 2 + + +def test_dedup_is_key_order_insensitive(store_with_tables, parsers): + """Canonical (JCS) hashing dedups payloads that differ only in key order.""" + image_id = "ALPHA_20240115T120000_001" + write_provenance( + store_with_tables, + [make_envelope(image_id, "features", {"area": 100.0, "perimeter": 40.0})], + parsers, + ) + write_provenance( + store_with_tables, + [make_envelope(image_id, "features", {"perimeter": 40.0, "area": 100.0})], + parsers, + ) + + results = get_provenance(store_with_tables, image_id, parsers) + assert len(results) == 1 + + +def test_dedup_treats_int_and_float_as_equal(store_with_tables, parsers): + """JCS number canonicalization: 1 and 1.0 hash identically, so they dedup.""" + image_id = "ALPHA_20240115T120000_001" + write_provenance( + store_with_tables, + [make_envelope(image_id, "features", {"count": 1})], + parsers, + ) + write_provenance( + store_with_tables, + [make_envelope(image_id, "features", {"count": 1.0})], + parsers, + ) + + results = get_provenance(store_with_tables, image_id, parsers) + assert len(results) == 1