"""What every adapter agrees to, and the two things none of them may do. The zero-training directive (§3, §4, §40.1) names a stack of pretrained models — SAM, DINOv3, MegaDescriptor, Grounding DINO, CountGD, a hosted reasoner — plus a deterministic OpenCV/NumPy path that §4 says to prefer whenever it is better. Those have almost nothing in common at the point of use: one returns masks, one returns a 384-dimensional vector, one returns a breath rate. So this file does **not** try to give them a single `run`. What they do have in common is governance, and that is what is unified here: **An adapter says what it costs, and `None` means nobody measured it.** `MeasuredCost` has no defaults and no published-figure fallback. A latency copied from a paper is a claim about somebody else's GPU, and the farms this serves run a 2 vCPU / 4 GiB container. **An adapter cannot produce a result it has no model for.** There is deliberately no `run` on this class — the same reason `providers.InferenceProvider` has none. A base implementation would be a way to return something plausible with nothing behind it, and that is the single failure this service exists to prevent. Work happens on the object `load()` returns, and `load()` raises when the artefact is absent. The task protocols below are the narrow interfaces callers actually use. They are kept as small as `detectors.Detector` is, for the same reason: everything added here is something the replacement adapter has to reimplement on the day a licence forces a swap, and ADR 0017 is the record of that day arriving. """ from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Protocol, runtime_checkable import numpy as np from PIL import Image class Task(str, Enum): """What an adapter produces. A model may do several.""" DETECT = "detect" SEGMENT = "segment" EMBED = "embed" COUNT = "count" TRACK = "track" POSE = "pose" #: Structured reasoning from a hosted multimodal model. Named apart from the #: rest because §4 is explicit that it is "an experimental visual reasoner, #: not an authority", and a caller should have to type the difference. REASON = "reason" #: Deterministic signal processing — optical flow, FFT, contour geometry. #: No weights, no licence question, and §4 says to prefer it where it wins. MEASURE = "measure" class Modality(str, Enum): IMAGE = "image" VIDEO = "video" AUDIO = "audio" @dataclass(frozen=True) class MeasuredCost: """Latency and memory from a run that actually happened. Every field is required. There is no `estimated` variant and no default, because the only thing worse than not knowing what an adapter costs on a 2 vCPU box is believing a number nobody produced there. `hardware` is free text on purpose: it has to be able to say "MacBook, 8 performance cores, not the target" as easily as it says the container SKU, and a reader needs to see which one they are looking at. """ hardware: str threads: int #: What it ran on, specifically enough to re-run. A slug from #: `evaluation/dataset.json`, or a count of frames from a named set. sample: str runs: int median_seconds: float peak_rss_mb: float measured_on: str @property def fits_cpu_service(self) -> bool: """Whether this would survive the 2 vCPU / 4 GiB CPU worker. **This is a placement hint, not a verdict on the model.** A model that returns False here belongs on a GPU host, and that is a deployment decision rather than a reason to drop a capability. The distinction is recorded because the earlier version of this file got it wrong and would have excluded most of directive §3 on the strength of a container size. """ return ( self.peak_rss_mb <= CPU_SERVICE_MEMORY_CEILING_MB and self.median_seconds <= INLINE_LATENCY_CEILING_SECONDS ) #: Peak RSS above which an adapter will not sit comfortably beside the API on the #: existing CPU worker. ADR 0018 measured YOLOX-m at 591 MB and YOLOX-x at #: 1,003 MB. 2,000 MB leaves the Python process, onnxruntime's arenas and #: Pillow's decode buffers room inside 4 GiB. CPU_SERVICE_MEMORY_CEILING_MB = 2000.0 #: Wall-clock above which a capability cannot run inline on a request, wherever #: it is hosted. ADR 0018's phrasing: "A 16-second inline request is not a #: request; it is a timeout with a result attached." Past this a capability needs #: a queue, not a bigger box. INLINE_LATENCY_CEILING_SECONDS = 8.0 class Placement(str, Enum): """Where a leg of the stack should run. Three tiers, and the choice between them is a product decision as much as an engineering one. Animap is offline-first (ADR 0002): a capability that needs a round trip is one a farm cannot use in a shed with no signal, so pushing work off the phone has a cost that a latency table does not show. """ #: On the phone. The only tier that works with no signal at all. ON_DEVICE = "on_device" #: The existing 2 vCPU / 4 GiB CPU container, beside the API. CPU_SERVICE = "cpu_service" #: A GPU host. Available, and the right answer for most of directive §3 — #: the models it names are GPU-class work and it was written knowing that. GPU_SERVICE = "gpu_service" @dataclass(frozen=True) class AdapterSpec: """An adapter's identity, licence position and measured cost. This is committed code rather than a JSON card, and that is the point. `providers.ModelArtefact` reads a card, and a card is written by whoever writes the card — ADR 0017 records a watchdog defeating the licence gate by declaring `Apache-2.0` over a path to AGPL weights. The `runtime` here names which loader runs, which is a fact about the code and not a claim about terms, and `adapters.licences` holds what that runtime's weights are really licensed under. """ adapter_id: str #: Which loader runs. The key into `licences.RUNTIME_LICENCES`, and the only #: field the licence gate trusts. runtime: str tasks: tuple[Task, ...] modalities: tuple[Modality, ...] #: What the zero-training directive asks this model for, quoted closely #: enough that a reader can find the section. directive_role: str #: False for the deterministic methods — optical flow, FFT, contour #: geometry. They need no weights, so they have no artefact to be absent and #: no licence to refuse, which is most of why §4 prefers them. requires_artefact: bool = True #: `None` until somebody runs it and writes the number down. Reported as #: "not measured", never filled in from a paper. measured: MeasuredCost | None = None #: Where this leg should run. A recommendation with a reason, not a #: constraint — see `Placement`. placement: Placement = Placement.CPU_SERVICE #: Whether a GPU is needed for this to be usable at all, as opposed to #: merely faster. Recorded separately from `placement` because "runs on CPU #: but slowly" and "does not run on CPU" are different facts and only the #: second one closes a door. requires_gpu: bool = False placement_reason: str = "" notes: str = "" @dataclass(frozen=True) class Availability: """Whether an adapter can run, and if not, what would change that. `remedy` exists because "unavailable" without it is the answer that gets read as "broken". The service already distinguishes *"no validated model exists"* from *"this is not planned"* in `main._unavailable_reason`, and an adapter that cannot say which of those it is has lost the distinction. """ ready: bool #: Empty when ready. Otherwise says what is missing, not what went wrong. reason: str = "" remedy: str = "" def __post_init__(self) -> None: if not self.ready and not self.reason: raise ValueError( "An unavailable adapter must say why. A bare False is what a " "caller renders as a silent failure." ) class AdapterError(RuntimeError): """The adapter is present but could not do the work.""" class AdapterUnavailable(AdapterError): """No model behind this adapter, so there is nothing to run. Raised by `load()`, never returned as a result. A caller that catches this reports `unavailable` — the state `JobState.UNAVAILABLE` already exists for, and which is the honest answer for most of the stack today. """ def __init__(self, availability: Availability) -> None: self.availability = availability message = availability.reason if availability.remedy: message = f"{message} {availability.remedy}" super().__init__(message) class Adapter: """A pretrained model, or a deterministic method, behind one interface. **There is no `run` here, and adding one would be the bug.** Subclasses expose whichever task protocol they satisfy — `Embedder`, `Segmenter`, `Reasoner` — and only after `load()` has succeeded against a real artefact. A default implementation on this class would be a way to answer a farmer with no model in the loop. """ spec: AdapterSpec def availability(self) -> Availability: """Whether this adapter could run right now. Must not load anything. Called on `/health` and `/capabilities`, which a platform probe hits often enough that reading a hundred megabytes of weights to answer it would be its own outage. """ raise NotImplementedError def load(self) -> "Adapter": """Prepare the runtime, or raise `AdapterUnavailable`. Returns self so a caller can write `adapter.load().embed(image)` and have no path to `embed` that skipped the check. """ raise NotImplementedError def describe(self) -> dict[str, object]: """Everything a governance reader needs, including what is unmeasured.""" from app.adapters import licences availability = self.availability() licence = licences.RUNTIME_LICENCES.get(self.spec.runtime) cost = self.spec.measured return { "adapter_id": self.spec.adapter_id, "runtime": self.spec.runtime, "tasks": [t.value for t in self.spec.tasks], "modalities": [m.value for m in self.spec.modalities], "directive_role": self.spec.directive_role, "ready": availability.ready, "reason": availability.reason, "remedy": availability.remedy, "licence": licence.licence if licence else "unknown runtime", "licence_source": licence.source_url if licence else "", "servable": bool(licence and licence.servable), "placement": self.spec.placement.value, "requires_gpu": self.spec.requires_gpu, "placement_reason": self.spec.placement_reason, # The absence is the finding, so it is spelled rather than nulled. "measured": ( { "hardware": cost.hardware, "threads": cost.threads, "sample": cost.sample, "runs": cost.runs, "median_seconds": cost.median_seconds, "peak_rss_mb": cost.peak_rss_mb, "measured_on": cost.measured_on, "fits_cpu_service": cost.fits_cpu_service, } if cost is not None else "not measured" ), "notes": self.spec.notes, } # --- Task protocols. Narrow on purpose. -------------------------------------- @dataclass(frozen=True) class Region: """A box, a mask, or both. The common currency of detection and segmentation. `box` is in source-image pixels, matching `detectors.Detection`, so a caller that already knows how to read a YOLOX box does not learn a second convention. `mask` is a boolean array at source-image resolution, or `None` when the adapter only localises. """ label: str score: float box: tuple[float, float, float, float] mask: np.ndarray | None = None area_fraction: float = 0.0 @property def has_mask(self) -> bool: return self.mask is not None @runtime_checkable class Embedder(Protocol): """Frozen features. §3's instruction for DINOv3 is explicit that this comes before any fine-tuning: "Start with frozen embeddings + nearest-neighbor retrieval." """ #: Length of the vector `embed` returns. Recorded because a retrieval index #: built at one dimension and queried at another fails silently. dimensions: int def embed(self, image: Image.Image) -> np.ndarray: """One L2-normalised float32 vector. Normalised by the adapter rather than the caller, so cosine similarity is a dot product everywhere and no index has to remember which convention it was built under. """ ... @runtime_checkable class Segmenter(Protocol): def segment( self, image: Image.Image, *, concepts: tuple[str, ...] = () ) -> list[Region]: ... @runtime_checkable class OpenVocabularyDetector(Protocol): """Text-prompted detection. §4 names Grounding DINO as the fallback for when SAM's concept prompting is weak.""" def detect_text( self, image: Image.Image, prompts: tuple[str, ...] ) -> list[Region]: ... @runtime_checkable class ExemplarCounter(Protocol): """Zero-shot counting, optionally guided by example boxes (§4, CountGD).""" def count( self, image: Image.Image, *, text: str = "", exemplars: tuple[tuple[float, float, float, float], ...] = (), ) -> "CountEstimate": ... @dataclass(frozen=True) class CountEstimate: """A count, or an honest refusal to publish one. `value` is `None` when the method ran and the result should not be shown — the same shape `app/counting.py` already uses, where withholding is a first class outcome rather than an exception. §6.3 requires the distinction between *visible count*, *unique birds observed* and *reconciled population* to survive to the UI, so `kind` carries it. """ value: float | None kind: str withheld_reason: str = "" confidence: float | None = None @runtime_checkable class Reasoner(Protocol): """A hosted multimodal model. §4: "All calls must return structured JSON" and "The multimodal model is an **experimental visual reasoner**, not an authority." Both are enforced in the adapter rather than left to the prompt — see `adapters/multimodal.py`. """ def reason( self, images: list[Image.Image], *, schema: dict, rubric: str, ) -> dict: ... @dataclass(frozen=True) class Measurement: """A number a deterministic method produced, with its own quality verdict. Signal processing fails differently from a model: it does not become uncertain, it becomes wrong in a way that still returns a float. So a measurement carries the evidence that the signal was there at all — `support` is whatever the method uses to know it measured something rather than measuring noise, and `usable` is its own judgement about that. """ kind: str value: float | None unit: str usable: bool support: dict[str, float] = field(default_factory=dict) detail: str = ""