From 230b04a79d3dbe1eb63d8ee77676f075f44f8bf7 Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Sun, 4 Oct 2026 20:55:39 +0800 Subject: [PATCH] =?UTF-8?q?=EF=BB=BFAdd=20a=20sensor=20bundle=20contract?= =?UTF-8?q?=20separating=20family,=20backend=20and=20profile?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Product name decided the pipeline: Mini worked live but replay refused it. A bundle descriptor now declares profile_id, product_family and processing_backend, with computational and visualization assets kept apart, and one resolver serves live and recorded input. Legacy bundles load through a documented adapter that never guesses a backend from a folder name. --- docs/sensor-bundles.md | 161 ++++++++ src/orisys/sensors/__init__.py | 146 +++++++ src/orisys/sensors/backends.py | 285 ++++++++++++++ src/orisys/sensors/calibration.py | 280 ++++++++++++++ src/orisys/sensors/capabilities.py | 209 +++++++++++ src/orisys/sensors/descriptor.py | 216 +++++++++++ src/orisys/sensors/fingerprint.py | 118 ++++++ src/orisys/sensors/legacy.py | 221 +++++++++++ src/orisys/sensors/resolve.py | 585 +++++++++++++++++++++++++++++ tests/test_sensor_bundle.py | 458 ++++++++++++++++++++++ 10 files changed, 2679 insertions(+) create mode 100644 docs/sensor-bundles.md create mode 100644 src/orisys/sensors/__init__.py create mode 100644 src/orisys/sensors/backends.py create mode 100644 src/orisys/sensors/calibration.py create mode 100644 src/orisys/sensors/capabilities.py create mode 100644 src/orisys/sensors/descriptor.py create mode 100644 src/orisys/sensors/fingerprint.py create mode 100644 src/orisys/sensors/legacy.py create mode 100644 src/orisys/sensors/resolve.py create mode 100644 tests/test_sensor_bundle.py diff --git a/docs/sensor-bundles.md b/docs/sensor-bundles.md new file mode 100644 index 0000000..3b7eba3 --- /dev/null +++ b/docs/sensor-bundles.md @@ -0,0 +1,161 @@ +# Sensor bundles and the processing contract + +A sensor bundle is a folder that describes one processing profile. This document owns the bundle +descriptor, the resolver, the backend registry and the fingerprints. `sensor.json` keeps its +existing meaning: it is the processing configuration, and its `type` field is the image-processing +arrangement (`compound_eye`, `single_camera`, `uv_unroll`) - never a product family. + +## Why the descriptor exists + +"Which pipeline processes these frames" used to be inferred from a product name. Those are four +different things, and collapsing them is what made Mini work live but be refused in replay: + +| Concept | Meaning | Example | +|---|---|---| +| Product family | What the operator calls it | Box, Mini, Finger | +| Processing backend | Which SDK pipeline processes the images | `compound_eye`, `uv_unroll` | +| Configuration profile | Camera, geometry and algorithm settings for one design revision | `orisys.mini.prototype-r1` | +| Device instance | The physical unit | Serial number, its calibration | + +Several families can share one backend, one family can have several profiles, and a recorded clip +has no device instance at all - only a profile and, optionally, a compatible calibration. + +## Descriptor + +`manifest.json` in the bundle root. Every field except the five identifying keys is optional, so a +bundle with no manifest still resolves through the compatibility adapter. + +```json +{ + "schema_version": 1, + "profile_id": "orisys.mini.prototype-r1", + "display_name": "Mini prototype R1", + "product_family": "mini", + "processing_backend": "compound_eye", + "sensor_json": "sensor.json", + "default_calibration": "force_calibration.json", + "computational_assets": { "unroll_map": "assets/30fps-0703/unroll_map_uv.npz" }, + "assets": { "scene": "scene/finger-display-1.0.0.oryscene" } +} +``` + +- **`computational_assets`** are required to process at all. A missing one fails the profile. +- **`assets`** are visualization only. A missing one is reported and never blocks processing - a + missing 3D scene is not a missing deformation field. +- Paths resolve relative to the bundle root, so moving or renaming the folder changes nothing. + Absolute paths and `..` escapes are rejected at parse time. +- An unsupported `processing_backend` or `schema_version` raises `DescriptorError`. The supported + descriptor schema version is 1. + +## Compatibility for existing bundles + +`resolve_bundle` reads a new-shape descriptor when `schema_version` is present. Otherwise - no +manifest, or the older `name`/`sensor_json`/`assets_dir`/`viz_pose` shape - the adapter describes +it. That is the only place a folder name is consulted, and only for the built-in set: + +| Folder | Product family | Backend (when the config does not decide) | +|---|---|---| +| `box` | box | `compound_eye` | +| `mini` | mini | `compound_eye` | +| `finger` | finger | `uv_unroll` | +| `finger-mems` | finger-mems | `compound_eye` | + +For a bundle without a descriptor the backend follows the configuration, using the rule +`Sensor._load_config` already applies: `type` when present, else `compound_eye` when the config has +a `stitcher` block, else `single_camera`. Only `finger` declares `type` today; `box`, `mini` and +`finger-mems` are decided by the stitcher. A folder that is neither built-in nor carries a +descriptor is rejected - never routed to `compound_eye` because box works that way. + +## Resolver + +```python +from orisys import resolve_bundle, validate_profile, create_processor, CalibrationSelection + +profile = resolve_bundle("sensors/mini") # or any bundle folder +profile = resolve_bundle("sensors/mini", calibration_selection=CalibrationSelection.NONE) +session = validate_profile(profile) # re-check against the filesystem now +rebuilt = create_processor(profile, source=0) # live camera +rebuilt = create_processor(profile, source="clip.mp4") # recorded frames, same semantics +``` + +`resolve_bundle(bundle_path, *, calibration_selection=None) -> ResolvedSensorProfile`. The profile +is plain data - no Qt, no application preferences, no device handle - so the same object drives a +camera and a recorded clip. `_set_calibration` on a device affects only its own calibration +selection. + +`validate_profile(profile) -> ValidationReport` re-reads the filesystem. It reports +`fingerprint.stale` when a computational asset changed since the profile was resolved. + +`create_processor(profile, *, sensor=None, source=None, include=..., contact=...)` returns a +`ProcessingSession` with `start`/`wait`/`latest`/`reset`/`pause`/`resume`/`stop`. It raises +`ProcessorCreationError` naming what is missing rather than returning a session that does nothing. + +### Backends + +| Identifier | Requires | Produces | +|---|---|---| +| `compound_eye` | `camera`, `stitcher`, `tracker` config | stitched frame, flow, force | +| `single_camera` | `camera` config | frames only - no deformation field, so no force | +| `uv_unroll` | `camera` config, the mapped computational asset | unrolled frame, flow, force | + +`known_backends()` and `spec_for(...)` are the registry. `uv_unroll` is not yet materialised from a +profile alone, because it needs geometry, a UV domain, a tactile processor and a capture object +assembled together - that is `orisys.finger.FingerRuntime.open`'s shape. `create_processor` says so +instead of pretending. + +## Capabilities + +Each capability is decided by the facts it depends on, not by the presence of a file: + +| Capability | Unavailable when | +|---|---| +| Video processing | the configuration cannot be loaded | +| Optical flow | no `tracker` block | +| Raw force | the backend produces no deformation field | +| Calibrated force | per axis: the calibration has no curve for it | +| Deformation | the backend computes no deformation field | +| 3D visualization | no scene asset is declared, or the declared one is missing | + +Deformation is relative unless a calibration says otherwise; it is reported with its meaning rather +than as a bare "available". `ValidationReport.state` is `ok`, `degraded` (usable, something +optional is missing) or `failed`. + +## Calibration + +`CalibrationSelection` is `BUNDLE_DEFAULT` (the default), `NONE`, or an explicit path. + +- `NONE` is a choice, not a failure: raw force stays available and the calibration fingerprint + records `none`. +- An explicitly selected calibration that is missing or unparseable is an **error**. It is never + replaced by another file. +- Applicability is `declared` when the document carries `meta.sensor_type`, `unknown` when legacy + metadata is absent, or `none`. `unknown` means it applies but its fit is unverified - which is + what the shipped `box` calibration reports. + +| Read | Meaning | +|---|---| +| `FrameSnapshot.normal_force_raw` / `shear_x_raw` / `shear_y_raw` | device counts, pre-calibration | +| `FrameSnapshot.normal_force` / `shear_x` / `shear_y` | newtons when a curve applies, counts otherwise | +| `FrameSnapshot.depth`, `centroid`, `contour`, `vnormal` | relative quantities, not newtons | + +`Sensor.force_triplets()` and `_force_raw` remain the raw and unfiltered view. + +## Fingerprints + +`processing_fingerprint` hashes the effective backend, the paired data, and each computational asset's role name with +its content hash. It excludes the bundle location, `display_name`, and every visualization asset: a +relocated or cosmetically renamed bundle keeps its fingerprint, and a missing scene never changes it. +It includes the config minus `name` and `version`. + +`calibration_fingerprint` hashes the selected document's content, or the constant `none`. + +`profile.inventory()` returns the reproducible snapshot: bundle-relative paths, asset hashes and +both fingerprints. Compare an inventory, not a path, to decide whether two installations would +compute the same numbers. + +## Resetting the reference + +Both backends process externally supplied frames through the same profile. After a gap in the +input - a pause, a seek, a scene change - call `session.reset()`. It is queued to the stage that +owns the tracker, so the reference frame is rebased on the frame about to be computed rather than +on a buffer the pipeline is still reading. diff --git a/src/orisys/sensors/__init__.py b/src/orisys/sensors/__init__.py new file mode 100644 index 0000000..4c3c38d --- /dev/null +++ b/src/orisys/sensors/__init__.py @@ -0,0 +1,146 @@ +"""Sensor bundle resolution: profile identity, capabilities, calibration and fingerprints. + +This package owns the answer to "which pipeline does this folder drive, what can it do, and would +two of them compute the same numbers". It replaces reading a product name as if it were a +processing backend. + +The public surface is small on purpose:: + + from orisys import resolve_bundle, validate_profile, create_processor, CalibrationSelection + + profile = resolve_bundle(r"sensors/mini") + print(profile.processing_backend, profile.processing_fingerprint) + for line in profile.summary_lines(): + print(line) + session = create_processor(profile, source=0) # live camera + # or: create_processor(profile, source="clip.mp4") # recorded frames, same semantics + +``ResolvedSensorProfile`` is plain data: no Qt, no application preferences, no device handle. A +device contributes only its calibration selection. +""" + +from __future__ import annotations + +from .backends import ( + BACKENDS, + BackendSpec, + ProcessingSession, + ProcessorCreationError, + create_processor, + known_backends, + spec_for, +) +from .calibration import ( + CalibrationSelection, + ResolvedCalibration, + axis_availability, + calibration_fingerprint, + resolve_calibration, +) +from .capabilities import ( + CALIBRATED_FORCE, + DEFORMATION, + OPTICAL_FLOW, + RAW_FORCE, + VIDEO_PROCESSING, + VISUALIZATION_3D, + Availability, + AxisAvailability, + Capabilities, + Issue, + ValidationReport, + ok, + report, + unavailable, +) +from .descriptor import ( + BACKEND_COMPOUND_EYE, + BACKEND_SINGLE_CAMERA, + BACKEND_UV_UNROLL, + CURRENT_DESCRIPTOR_SCHEMA, + DESCRIPTOR_FILENAME, + SUPPORTED_BACKENDS, + SUPPORTED_DESCRIPTOR_SCHEMAS, + BundleDescriptor, + DescriptorError, + parse_descriptor, +) +from .fingerprint import ( + CALIBRATION_FINGERPRINT_ALGORITHM, + FINGERPRINT_ALGORITHM, + dependency_inventory, + file_digest, + processing_fingerprint, +) +from .legacy import ( + BUILT_IN_PROFILES, + LEGACY_MANIFEST_KEYS, + backend_from_config, + describe_legacy_bundle, +) +from .resolve import ( + ResolvedSensorProfile, + resolve_bundle, + resolve_named_bundle, + validate_profile, +) + +__all__ = [ + # entry points + "resolve_bundle", + "resolve_named_bundle", + "validate_profile", + "create_processor", + # resolved data + "ResolvedSensorProfile", + "ResolvedCalibration", + "CalibrationSelection", + "ValidationReport", + "Capabilities", + "Availability", + "AxisAvailability", + "Issue", + # descriptor + "BundleDescriptor", + "DescriptorError", + "parse_descriptor", + "describe_legacy_bundle", + "backend_from_config", + "BUILT_IN_PROFILES", + "LEGACY_MANIFEST_KEYS", + "DESCRIPTOR_FILENAME", + "CURRENT_DESCRIPTOR_SCHEMA", + "SUPPORTED_DESCRIPTOR_SCHEMAS", + # backends + "BACKENDS", + "BackendSpec", + "ProcessingSession", + "ProcessorCreationError", + "known_backends", + "spec_for", + "BACKEND_COMPOUND_EYE", + "BACKEND_SINGLE_CAMERA", + "BACKEND_UV_UNROLL", + "SUPPORTED_BACKENDS", + # calibration helpers + "resolve_calibration", + "axis_availability", + "calibration_fingerprint", + # capabilities + "VIDEO_PROCESSING", + "OPTICAL_FLOW", + "RAW_FORCE", + "CALIBRATED_FORCE", + "DEFORMATION", + "VISUALIZATION_3D", + "ok", + "report", + "unavailable", + # reproducibility + "processing_fingerprint", + "calibration_fingerprint", + "dependency_inventory", + "file_digest", + "FINGERPRINT_ALGORITHM", + "CALIBRATION_FINGERPRINT_ALGORITHM", +] diff --git a/src/orisys/sensors/backends.py b/src/orisys/sensors/backends.py new file mode 100644 index 0000000..d4190c8 --- /dev/null +++ b/src/orisys/sensors/backends.py @@ -0,0 +1,285 @@ +"""Backend registry: turn a resolved profile into a processing session. + +Selection is by backend identifier, never by product name. The identifiers are the SDK's existing +pipeline vocabulary - the same spellings :func:`orisys.frame_processors.create_frame_processor` +already accepts - so a profile backed by ``compound_eye`` builds a stitcher and a tile processor, +while a ``single_camera`` profile is a plain frame path with no geometry. + +A backend that needs material the SDK does not yet build from a profile says so plainly instead of +producing something that looks like it worked. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Mapping + +from ..frame_processors import create_frame_processor +from ..pipeline import PauseLevel, StreamPipeline +from ..stitch.api import Stitcher +from .capabilities import Issue +from .descriptor import ( + BACKEND_COMPOUND_EYE, + BACKEND_SINGLE_CAMERA, + BACKEND_UV_UNROLL, + SUPPORTED_BACKENDS, +) +from .resolve import ResolvedSensorProfile, validate_profile + + +class ProcessorCreationError(RuntimeError): + """The profile cannot be turned into a processing session, with the reason why.""" + + +@dataclass(frozen=True) +class BackendSpec: + """What one backend identifier requires and what it can produce.""" + + backend_id: str + #: Configuration blocks the backend cannot run without. + requires_config_keys: tuple[str, ...] = () + #: Computational asset roles it needs, if any. + requires_asset_roles: tuple[str, ...] = () + #: Whether the produced session yields a force reading. + produces_force: bool = True + description: str = "" + + +BACKENDS: Mapping[str, BackendSpec] = { + BACKEND_COMPOUND_EYE: BackendSpec( + backend_id=BACKEND_COMPOUND_EYE, + requires_config_keys=("stitcher", "camera", "tracker"), + produces_force=True, + description="compound-eye tiles stitched, then optical flow and HHD", + ), + BACKEND_SINGLE_CAMERA: BackendSpec( + backend_id=BACKEND_SINGLE_CAMERA, + requires_config_keys=("camera",), + produces_force=False, + description="single camera frame with no tiling and no deformation geometry", + ), + BACKEND_UV_UNROLL: BackendSpec( + backend_id=BACKEND_UV_UNROLL, + requires_config_keys=("camera",), + requires_asset_roles=("unroll_map",), + produces_force=True, + description="finger UV unroll onto a mesh domain, then optical flow and HHD", + ), +} + + +def known_backends() -> tuple[str, ...]: + """The backend identifiers this SDK can build.""" + return tuple(BACKENDS) + + +def spec_for(backend_id: str) -> BackendSpec: + """The spec for ``backend_id``, or an explicit error naming the supported set.""" + try: + return BACKENDS[backend_id] + except KeyError as exc: + raise ProcessorCreationError( + f"unsupported processing backend {backend_id!r}; supported: {sorted(BACKENDS)}" + ) from exc + + +def _missing_requirements( + spec: BackendSpec, profile: ResolvedSensorProfile +) -> list[str]: + missing: list[str] = [] + for key in spec.requires_config_keys: + if key not in profile.config: + missing.append(f"config block {key!r}") + for role in spec.requires_asset_roles: + path = profile.computational_assets.get(role) + if path is None: + missing.append(f"computational asset {role!r}") + elif not path.is_file(): + missing.append(f"computational asset {role!r} (missing at {path})") + return missing + + +class ProcessingSession: + """One profile, one process: the unit both live acquisition and video analysis drive. + + ``process`` is a :class:`~orisys.pipeline.StreamPipeline` when the backend runs the three-stage + pipeline. ``streams`` is False for a backend that only produces frames, in which case + ``wait()`` hands back the frames themselves rather than a force snapshot. + """ + + def __init__( + self, + *, + profile: ResolvedSensorProfile, + backend_id: str, + process: Any, + streams: bool, + sensor: Any = None, + ) -> None: + self.profile = profile + self.backend_id = backend_id + self.process = process + self.streams = streams + self.sensor = sensor + + # ------------------------------------------------------------------ lifecycle + def start(self) -> "ProcessingSession": + if self.streams and hasattr(self.process, "start"): + self.process.start() + return self + + def __enter__(self) -> "ProcessingSession": + return self.start() + + def __exit__(self, exc_type, exc, tb) -> bool: + self.stop() + return False + + def stop(self, timeout: float = 5.0) -> None: + if self.streams and hasattr(self.process, "stop"): + self.process.stop(timeout=timeout) + + def wait(self, timeout: float | None = None): + """The next result: a force snapshot when streaming, else the next frame.""" + if self.streams: + return self.process.wait(timeout=timeout) + return self.process(timeout=timeout) + + def reset(self) -> None: + """Reset the pipeline's reference: the flow tracker and force-filter state.""" + if self.streams: + self.process.reset() + return + # Frame-only backend: the reference lives on the sensor's tracker. + sensor = self.sensor + if sensor is not None and hasattr(sensor, "reset"): + sensor.reset() + + def pause(self, level: PauseLevel | str = PauseLevel.CAPTURE_ONLY) -> None: + """Hold the pipeline. Only a streaming backend can be held.""" + if not self.streams: + raise ProcessorCreationError( + f"backend {self.backend_id!r} does not run a pipeline, so it cannot be paused" + ) + self.process.pause(level) + + def resume(self, *, reset: bool = True) -> None: + if not self.streams: + raise ProcessorCreationError( + f"backend {self.backend_id!r} does not run a pipeline, so it cannot be resumed" + ) + self.process.resume(reset=reset) + + @property + def paused(self) -> bool: + return bool(self.streams and getattr(self.process, "paused", False)) + + @property + def running(self) -> bool: + return bool(self.streams and getattr(self.process, "running", False)) + + def latest(self): + """The most recent snapshot or frame, or ``None``.""" + if self.streams and hasattr(self.process, "latest"): + return self.process.latest() + return None + + def capabilities(self): + return self.profile.capabilities + + +def _stream_factory(sensor: Any, config: Mapping[str, Any], *, include=("image",), contact=False): + """Build the three-stage pipeline for a sensor-like object.""" + + def producer(): + return sensor.get_img() + + return StreamPipeline(sensor, producer, contact=contact, include=include) + + +def create_processor( + profile: ResolvedSensorProfile, + *, + sensor: Any = None, + source: Any = None, + include=("image",), + contact: bool = False, + verbose: bool = False, +): + """Build the processing session for ``profile``. + + ``sensor`` may be supplied when the caller already owns a device (a live ``Sensor`` trained on + this profile's ``sensor.json``); otherwise a new one is built from the profile. ``source`` is + the capture source - a camera index or a video path - and is only used when building a sensor. + Both live and recorded input therefore go through this one function. + + Raises :class:`ProcessorCreationError` when the profile cannot be processed, naming what is + missing, rather than returning something that silently does nothing. + """ + validation = validate_profile(profile) + if not validation.usable: + reasons = "; ".join(str(issue) for issue in validation.errors) + raise ProcessorCreationError( + f"profile {profile.profile_id!r} cannot be processed: {reasons}" + ) + + backend_id = profile.processing_backend + spec = spec_for(backend_id) + missing = _missing_requirements(spec, profile) + if missing: + raise ProcessorCreationError( + f"profile {profile.profile_id!r} with backend {backend_id!r} is missing " + + ", ".join(missing) + ) + + if backend_id == BACKEND_UV_UNROLL: + # The UV backend needs geometry, a UV domain, a tactile processor and a capture object + # assembled together, which is FingerRuntime's shape rather than a config-driven Sensor. + # Building it here would duplicate that assembly, so it is not silently faked. + raise ProcessorCreationError( + "backend 'uv_unroll' cannot yet be materialised from a profile alone; open it " + "through orisys.finger.FingerRuntime.open(export_dir=..., config=...) until this " + "backend builds its UV domain from the profile's computational assets" + ) + + if sensor is None: + from ..sensor import Sensor + + sensor = Sensor( + source if source is not None else 0, + config_name=str(profile.sensor_json_path), + verbose=verbose, + ) + elif verbose: + print(f"[processor] using the supplied sensor for {profile.profile_id}") + + if backend_id == BACKEND_COMPOUND_EYE: + config_stitcher = profile.config.get("stitcher") + stitcher = Stitcher( + isstitch=True, config_stitcher=config_stitcher, verbose=verbose + ) + sensor.frame_processor = create_frame_processor( + BACKEND_COMPOUND_EYE, stitcher=stitcher + ) + return ProcessingSession( + profile=profile, + backend_id=backend_id, + process=_stream_factory(sensor, profile.config, include=include, contact=contact), + streams=True, + sensor=sensor, + ) + + # single_camera: the sensor already builds its own pass-through processor from the config. + sensor.frame_processor = create_frame_processor(BACKEND_SINGLE_CAMERA) + + def next_frame(timeout: float | None = None): + frame = sensor.get_img() + return frame + + return ProcessingSession( + profile=profile, + backend_id=backend_id, + process=next_frame, + streams=False, + sensor=sensor, + ) diff --git a/src/orisys/sensors/calibration.py b/src/orisys/sensors/calibration.py new file mode 100644 index 0000000..164cf45 --- /dev/null +++ b/src/orisys/sensors/calibration.py @@ -0,0 +1,280 @@ +"""Calibration selection and compatibility for a resolved profile. + +Three outcomes are distinct and must not be collapsed: + +* an explicitly selected calibration that loads and applies - the caller asked for it; +* an explicit "no calibration" - the caller asked for raw output and gets it, reported as such; +* a selected calibration that does not load or does not fit - an error, never a quiet + substitution with some other file. + +A calibration file merely existing proves nothing, so presence is never the check: the document +is parsed through :mod:`orisys.calibration`, its units and channels are read, and its declared +applicability is compared with the profile it was selected for. Legacy documents that carry no +identifying metadata are reported as *unknown applicability* rather than as compatible. +""" + +from __future__ import annotations + +import enum +import hashlib +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Mapping + +from .. import calibration as cal_mod +from .capabilities import Availability, Issue, ok as _ok, unavailable + +#: Axis availability when no calibration curve is present for that axis. +_NO_CURVE = "no calibration curve for this axis" + +#: Sentinel digest for "explicitly no calibration". A distinct constant rather than an empty +#: string, so a stored fingerprint always says which state it recorded. +_NONE_FINGERPRINT = "none" + + +class CalibrationSelection(enum.Enum): + """How the caller wants the calibration chosen.""" + + #: Use the bundle's declared default, else ``force_calibration.json`` beside ``sensor.json``. + BUNDLE_DEFAULT = "bundle_default" + #: Explicitly run uncalibrated. Raw force is still available; it is a choice, not a failure. + NONE = "none" + #: Use this file - see :meth:`from_value`. + EXPLICIT = "explicit" + + @classmethod + def from_value(cls, value: "CalibrationSelection | str | Path | None") -> tuple["CalibrationSelection", Path | None]: + """Normalise a caller-supplied selection into ``(mode, path)``.""" + if value is None: + return cls.BUNDLE_DEFAULT, None + if isinstance(value, cls): + return value, None + text = str(value).strip() + if not text or text.lower() in ("none", "raw", "off"): + return cls.NONE, None + if text.lower() in ("default", "bundle", "bundle_default"): + return cls.BUNDLE_DEFAULT, None + return cls.EXPLICIT, Path(text).expanduser() + + +@dataclass(frozen=True) +class ResolvedCalibration: + """The calibration side of a resolved profile.""" + + selection: CalibrationSelection + path: Path | None + document: "cal_mod.Calibration | None" = None + #: ``"declared"`` when the document carries identity metadata, ``"unknown"`` when it does + #: not (legacy), ``"none"`` for an explicit uncalibrated run. + applicability: str = "none" + #: Axes this calibration provides a curve for. + axes: tuple[str, ...] = () + unit: str = "" + fingerprint: str = "" + issues: tuple[Issue, ...] = () + + @property + def applied(self) -> bool: + return self.document is not None + + def axis_detail(self, axis: str) -> str: + if self.document is None: + return _NO_CURVE if self.selection is CalibrationSelection.NONE else "no calibration" + return "" if axis in self.axes else _NO_CURVE + + def as_dict(self) -> dict[str, object]: + return { + "selection": self.selection.value, + "path": None if self.path is None else str(self.path), + "applicability": self.applicability, + "axes": list(self.axes), + "unit": self.unit, + "fingerprint": self.fingerprint, + } + + +def _identity_of(document: "cal_mod.Calibration") -> Mapping[str, str]: + identity = getattr(document, "identity", None) + return identity if isinstance(identity, Mapping) else {} + + +def resolve_calibration( + *, + bundle_root: Path, + declared_default: Path | None, + profile_id: str, + product_family: str, + selection: CalibrationSelection | str | Path | None = None, +) -> ResolvedCalibration: + """Resolve and validate the calibration for one profile. + + Returns a :class:`ResolvedCalibration` whose ``issues`` carry every reason a selected + calibration could not be used. It never falls back to a different file. + """ + mode, explicit = CalibrationSelection.from_value(selection) + if mode is CalibrationSelection.NONE: + return ResolvedCalibration( + selection=mode, path=None, applicability="none", fingerprint=_NONE_FINGERPRINT + ) + + if mode is CalibrationSelection.EXPLICIT: + path = explicit + else: + path = declared_default + if path is None: + fallback = bundle_root / "force_calibration.json" + path = fallback if fallback.is_file() else None + + if path is None: + # Not an error: a profile may legitimately ship no calibration. Calibrated force is + # simply unavailable, per axis, and the caller can still select a file explicitly. + issue = Issue( + code="calibration.absent", + message="no calibration for this profile; raw force is available", + path=bundle_root, + ) + return ResolvedCalibration( + selection=mode, + path=None, + applicability="none", + fingerprint=_NONE_FINGERPRINT, + issues=(issue,), + ) + + path = Path(path) + if not path.is_file(): + return ResolvedCalibration( + selection=mode, + path=path, + issues=( + Issue( + code="calibration.missing", + message="the selected calibration file does not exist", + path=path, + ), + ), + ) + + try: + document = cal_mod.load(str(path), strict=True) + except Exception as exc: # noqa: BLE001 - reported as an actionable issue + return ResolvedCalibration( + selection=mode, + path=path, + issues=( + Issue( + code="calibration.unreadable", + message=f"selected calibration could not be loaded: {exc}", + path=path, + ), + ), + ) + if document is None: # defensive: strict load raises rather than returning None + return ResolvedCalibration( + selection=mode, + path=path, + issues=( + Issue( + code="calibration.unreadable", + message="selected calibration produced no document", + path=path, + ), + ), + ) + + axes = tuple(sorted(document.channels)) + unit = str(getattr(document, "unit", "") or "") + issues: list[Issue] = [] + if unit and unit.upper() not in ("N", "NEWTON", "NEWTONS"): + issues.append( + Issue( + code="calibration.unit", + message=f"calibration unit {unit!r} is not a force unit in newtons", + path=path, + ) + ) + + identity = _identity_of(document) + declared_type = str(identity.get("sensor_type") or "") + if declared_type: + applicability = "declared" + if declared_type != product_family: + issues.append( + Issue( + code="calibration.family-mismatch", + message=( + f"calibration declares sensor_type {declared_type!r} but the profile is " + f"{product_family!r}; it was selected explicitly, so it applies as given" + ), + path=path, + ) + ) + else: + # No metadata to compare against: honest answer is that applicability is unknown. + applicability = "unknown" + issues.append( + Issue( + code="calibration.applicability-unknown", + message=( + "calibration carries no identifying metadata (no meta.sensor_type); " + f"its applicability to {profile_id} cannot be verified" + ), + path=path, + ) + ) + + if not axes: + issues.append( + Issue( + code="calibration.no-channels", + message="calibration defines no usable channel curve", + path=path, + ) + ) + + return ResolvedCalibration( + selection=mode, + path=path, + document=document, + applicability=applicability, + axes=axes, + unit=unit, + fingerprint=calibration_fingerprint(document, path), + issues=tuple(issues), + ) + + +def axis_availability(resolved: ResolvedCalibration) -> tuple[object, object, object]: + """``(fn, fx, fy)`` availability, each with its own reason when unavailable.""" + if resolved.document is None: + reason = ( + "explicitly uncalibrated" + if resolved.selection is CalibrationSelection.NONE + else (resolved.issues[0].message if resolved.issues else "no calibration") + ) + blank = unavailable(reason) + return blank, blank, blank + out: list[Availability] = [] + for axis in ("fn", "fx", "fy"): + if axis in resolved.axes: + out.append(_ok(f"applicability {resolved.applicability}")) + else: + out.append( + unavailable(_NO_CURVE, detail=f"{resolved.path.name} has no {axis} curve") + ) + return out[0], out[1], out[2] + + +def calibration_fingerprint(document: "cal_mod.Calibration", path: Path) -> str: + """Content hash of the calibration document. + + Hashes what the document *says*, not where it lives: selecting an identical calibration from + a different directory must produce the same fingerprint. + """ + try: + payload = document.to_dict() + except Exception: # noqa: BLE001 - fall back to the raw bytes rather than fail + payload = path.read_bytes().decode("utf-8", errors="replace") + text = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) + return hashlib.sha256(text.encode("utf-8")).hexdigest() diff --git a/src/orisys/sensors/capabilities.py b/src/orisys/sensors/capabilities.py new file mode 100644 index 0000000..3f7cad4 --- /dev/null +++ b/src/orisys/sensors/capabilities.py @@ -0,0 +1,209 @@ +"""Capability reporting: what a resolved profile can actually do, and why not. + +The picker needs answers, not guesses. "``sensor.json`` exists so processing works" and +"``scene/`` exists so 3D works" are exactly the two inferences this module exists to replace: +each capability is decided by the asset and configuration facts that capability actually needs, +and an unavailable one carries the reason it is unavailable. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Iterable, Mapping + +__all__ = [ + "Availability", + "AxisAvailability", + "Capabilities", + "Issue", + "ValidationReport", + "ok", + "report", + "unavailable", +] + +#: Capability identifiers, stable for the UI and for stored reports. +VIDEO_PROCESSING = "video_processing" +OPTICAL_FLOW = "optical_flow" +RAW_FORCE = "raw_force" +CALIBRATED_FORCE = "calibrated_force" +DEFORMATION = "deformation" +VISUALIZATION_3D = "visualization_3d" + +#: Units a "deformation" output can carry. The capability is reported with the meaning it has, +#: never as a bare "available". +DEFORMATION_MEANINGS: Mapping[str, str] = { + "optical_flow_px": "per-frame optical flow in pixels", + "displacement_relative": "relative displacement, uncalibrated", + "depth_relative": "relative depth from the deformation field", +} + + +@dataclass(frozen=True) +class Availability: + """Whether one capability is usable, and the reason when it is not.""" + + available: bool + reason: str = "" + #: Free-form detail the caller may show, such as which axis is missing. + detail: str = "" + + def as_dict(self) -> dict[str, object]: + payload: dict[str, object] = {"available": self.available} + if self.reason: + payload["reason"] = self.reason + if self.detail: + payload["detail"] = self.detail + return payload + + +def ok(detail: str = "") -> Availability: + return Availability(True, detail=detail) + + +def unavailable(reason: str, detail: str = "") -> Availability: + return Availability(False, reason=reason, detail=detail) + + +@dataclass(frozen=True) +class AxisAvailability: + """Per-axis availability, so "calibrated" is never a single yes/no.""" + + fn: Availability + fx: Availability + fy: Availability + + def any_available(self) -> bool: + return self.fn.available or self.fx.available or self.fy.available + + def as_dict(self) -> dict[str, dict[str, object]]: + return { + "fn": self.fn.as_dict(), + "fx": self.fx.as_dict(), + "fy": self.fy.as_dict(), + } + + +@dataclass(frozen=True) +class Capabilities: + """Every capability the picker and the analysis layer ask about.""" + + video_processing: Availability + optical_flow: Availability + raw_force: Availability + calibrated_force: AxisAvailability + deformation: Availability + visualization_3d: Availability + + def as_dict(self) -> dict[str, object]: + return { + VIDEO_PROCESSING: self.video_processing.as_dict(), + OPTICAL_FLOW: self.optical_flow.as_dict(), + RAW_FORCE: self.raw_force.as_dict(), + CALIBRATED_FORCE: self.calibrated_force.as_dict(), + DEFORMATION: self.deformation.as_dict(), + VISUALIZATION_3D: self.visualization_3d.as_dict(), + } + + def summary_lines(self) -> list[str]: + """One human-readable line per capability, for a status panel.""" + lines = [ + f"video processing: {_fmt(self.video_processing)}", + f"optical flow: {_fmt(self.optical_flow)}", + f"raw force: {_fmt(self.raw_force)}", + ] + for axis, availability in ( + ("fn", self.calibrated_force.fn), + ("fx", self.calibrated_force.fx), + ("fy", self.calibrated_force.fy), + ): + lines.append(f"calibrated {axis}: {_fmt(availability)}") + lines.append(f"deformation: {_fmt(self.deformation)}") + lines.append(f"3D visualization: {_fmt(self.visualization_3d)}") + return lines + + +def _fmt(availability: Availability) -> str: + if availability.available: + return "available" + (f" ({availability.detail})" if availability.detail else "") + return "unavailable" + (f" - {availability.reason}" if availability.reason else "") + + +@dataclass(frozen=True) +class Issue: + """One validation finding, addressed at a file when there is one.""" + + code: str + message: str + path: Path | None = None + + def __str__(self) -> str: + where = f" [{self.path}]" if self.path is not None else "" + return f"{self.code}: {self.message}{where}" + + +@dataclass(frozen=True) +class ValidationReport: + """The outcome of validating a resolved profile. + + ``ok`` means processing can start. ``degraded`` means it can start but something optional is + missing, so the UI should say what. ``failed`` means it cannot. + """ + + state: str + errors: tuple[Issue, ...] = () + warnings: tuple[Issue, ...] = () + capabilities: Capabilities | None = None + + @property + def ok(self) -> bool: + return self.state == "ok" + + @property + def usable(self) -> bool: + return self.state != "failed" + + def as_dict(self) -> dict[str, object]: + return { + "state": self.state, + "errors": [ + {"code": i.code, "message": i.message, "path": None if i.path is None else str(i.path)} + for i in self.errors + ], + "warnings": [ + {"code": i.code, "message": i.message, "path": None if i.path is None else str(i.path)} + for i in self.warnings + ], + "capabilities": None if self.capabilities is None else self.capabilities.as_dict(), + } + + def summary_lines(self) -> list[str]: + lines = [f"validation: {self.state}"] + lines.extend(f" error: {issue}" for issue in self.errors) + lines.extend(f" warning: {issue}" for issue in self.warnings) + if self.capabilities is not None: + lines.extend(f" {line}" for line in self.capabilities.summary_lines()) + return lines + + +def report( + errors: Iterable[Issue], + warnings: Iterable[Issue], + capabilities: Capabilities | None = None, +) -> ValidationReport: + """Build a report, deriving the state from what was found.""" + error_list = tuple(errors) + warning_list = tuple(warnings) + if error_list: + state = "failed" + elif warning_list: + state = "degraded" + else: + state = "ok" + return ValidationReport( + state=state, + errors=error_list, + warnings=warning_list, + capabilities=capabilities, + ) diff --git a/src/orisys/sensors/descriptor.py b/src/orisys/sensors/descriptor.py new file mode 100644 index 0000000..955d49a --- /dev/null +++ b/src/orisys/sensors/descriptor.py @@ -0,0 +1,216 @@ +"""Sensor bundle descriptor: the versioned identity of a configuration profile. + +A bundle folder holds three different things that ``sensor_type`` used to conflate: + +* ``product_family`` - what the operator calls the product (box, mini, finger); +* ``processing_backend`` - which SDK pipeline processes the images; +* ``profile_id`` - the identity of this particular configuration revision. + +The descriptor states those explicitly in ``manifest.json``. ``sensor.json`` keeps its +existing meaning: it is the processing configuration, and its ``type`` field stays the +image-processing arrangement (``compound_eye`` / ``single_camera`` / ``uv_unroll``), never a +product family. + +Every path in a descriptor resolves relative to the bundle root, so moving or renaming the +folder changes nothing about what it means. Bundles that predate this descriptor are handled +by :mod:`orisys.sensors.legacy`; nothing is ever inferred from a folder name. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Mapping + +CURRENT_DESCRIPTOR_SCHEMA = 1 +SUPPORTED_DESCRIPTOR_SCHEMAS: tuple[int, ...] = (1,) + +#: Backend identifiers the SDK can build. Kept in step with +#: :func:`orisys.frame_processors.create_frame_processor`, which is where the vocabulary already +#: lives; the registry asserts the two agree. +BACKEND_COMPOUND_EYE = "compound_eye" +BACKEND_SINGLE_CAMERA = "single_camera" +BACKEND_UV_UNROLL = "uv_unroll" +SUPPORTED_BACKENDS: tuple[str, ...] = ( + BACKEND_COMPOUND_EYE, + BACKEND_SINGLE_CAMERA, + BACKEND_UV_UNROLL, +) + +DESCRIPTOR_FILENAME = "manifest.json" + + +class DescriptorError(ValueError): + """A bundle descriptor is missing, malformed, or names something unsupported.""" + + +def _resolve_relative(bundle_root: Path, raw: str, *, field_name: str) -> Path: + """Resolve a descriptor path inside the bundle, refusing to escape it. + + Relative paths are the contract: an absolute path would make the descriptor depend on where + it was installed, which is exactly what the fingerprint must not include. + """ + text = str(raw).strip() + if not text: + raise DescriptorError(f"{field_name}: empty path") + candidate = Path(text) + if candidate.is_absolute(): + raise DescriptorError( + f"{field_name}: {text!r} is absolute; descriptor paths resolve relative to the bundle" + ) + resolved = (bundle_root / candidate).resolve() + try: + resolved.relative_to(bundle_root.resolve()) + except ValueError as exc: + raise DescriptorError( + f"{field_name}: {text!r} escapes the bundle folder" + ) from exc + return resolved + + +@dataclass(frozen=True) +class BundleDescriptor: + """A parsed ``manifest.json``, with every optional field filled in.""" + + schema_version: int + profile_id: str + display_name: str + product_family: str + processing_backend: str + bundle_root: Path + sensor_json: Path + default_calibration: Path | None = None + #: Assets required to process at all. A missing one makes the profile fail, not degrade. + computational_assets: Mapping[str, Path] = field(default_factory=dict) + #: Optional visualization assets. A missing one never blocks image processing. + assets: Mapping[str, Path] = field(default_factory=dict) + #: True when this came from a manifest rather than a compatibility adapter. + from_manifest: bool = False + #: Where ``processing_backend`` was decided, for diagnostics. + backend_source: str = "manifest" + diagnostics: tuple[str, ...] = () + + def as_dict(self) -> dict[str, Any]: + """The descriptor as JSON-serialisable data, with paths made bundle-relative.""" + root = self.bundle_root + + def rel(path: Path | None) -> str | None: + if path is None: + return None + try: + return path.relative_to(root).as_posix() + except ValueError: + return path.as_posix() + + return { + "schema_version": self.schema_version, + "profile_id": self.profile_id, + "display_name": self.display_name, + "product_family": self.product_family, + "processing_backend": self.processing_backend, + "sensor_json": rel(self.sensor_json), + "default_calibration": rel(self.default_calibration), + "computational_assets": { + role: rel(path) for role, path in sorted(self.computational_assets.items()) + }, + "assets": {role: rel(path) for role, path in sorted(self.assets.items())}, + } + + +def _require_str(data: Mapping[str, Any], key: str, *, where: str) -> str: + value = data.get(key) + if not isinstance(value, str) or not value.strip(): + raise DescriptorError(f"{where}: {key!r} must be a non-empty string") + return value.strip() + + +def _asset_map( + data: Mapping[str, Any], key: str, bundle_root: Path +) -> dict[str, Path]: + raw = data.get(key) + if raw is None: + return {} + if not isinstance(raw, Mapping): + raise DescriptorError(f"{key!r} must be an object of {{role: relative path}}") + out: dict[str, Path] = {} + for role, value in raw.items(): + if not isinstance(role, str) or not role.strip(): + raise DescriptorError(f"{key!r}: asset roles must be non-empty strings") + if not isinstance(value, str): + raise DescriptorError(f"{key!r}.{role}: path must be a string") + out[str(role).strip()] = _resolve_relative( + bundle_root, value, field_name=f"{key}.{role}" + ) + return out + + +def parse_descriptor(bundle_root: str | Path) -> BundleDescriptor: + """Read ``manifest.json`` from ``bundle_root`` into a :class:`BundleDescriptor`. + + Raises :class:`DescriptorError` when the file is absent, malformed, names an unsupported + backend, or declares an unsupported schema version. It never guesses: a bundle without a + usable manifest must go through :func:`orisys.sensors.legacy.describe_legacy_bundle`. + """ + root = Path(bundle_root).expanduser() + if not root.is_dir(): + raise DescriptorError(f"not a bundle folder: {root}") + manifest = root / DESCRIPTOR_FILENAME + if not manifest.is_file(): + raise DescriptorError( + f"{root.name}: no {DESCRIPTOR_FILENAME}. A bundle needs one to be resolved; " + "legacy built-ins are handled by the compatibility adapter" + ) + try: + data = json.loads(manifest.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise DescriptorError(f"{manifest}: unreadable descriptor: {exc}") from exc + if not isinstance(data, Mapping): + raise DescriptorError(f"{manifest}: descriptor must be a JSON object") + + raw_schema = data.get("schema_version") + try: + schema = int(raw_schema) # type: ignore[arg-type] + except (TypeError, ValueError) as exc: + raise DescriptorError( + f"{manifest}: schema_version must be an integer, got {raw_schema!r}" + ) from exc + if schema not in SUPPORTED_DESCRIPTOR_SCHEMAS: + raise DescriptorError( + f"{manifest}: unsupported descriptor schema_version {schema}; " + f"supported: {list(SUPPORTED_DESCRIPTOR_SCHEMAS)}" + ) + + backend = _require_str(data, "processing_backend", where=str(manifest)) + if backend not in SUPPORTED_BACKENDS: + raise DescriptorError( + f"{manifest}: unsupported processing_backend {backend!r}; " + f"supported: {list(SUPPORTED_BACKENDS)}" + ) + + sensor_json = _resolve_relative( + root, + str(data.get("sensor_json") or "sensor.json"), + field_name="sensor_json", + ) + default_calibration = None + raw_cal = data.get("default_calibration") + if raw_cal: + default_calibration = _resolve_relative( + root, str(raw_cal), field_name="default_calibration" + ) + + return BundleDescriptor( + schema_version=schema, + profile_id=_require_str(data, "profile_id", where=str(manifest)), + display_name=_require_str(data, "display_name", where=str(manifest)), + product_family=_require_str(data, "product_family", where=str(manifest)), + processing_backend=backend, + bundle_root=root.resolve(), + sensor_json=sensor_json, + default_calibration=default_calibration, + computational_assets=_asset_map(data, "computational_assets", root.resolve()), + assets=_asset_map(data, "assets", root.resolve()), + from_manifest=True, + backend_source="manifest", + ) diff --git a/src/orisys/sensors/fingerprint.py b/src/orisys/sensors/fingerprint.py new file mode 100644 index 0000000..192ac13 --- /dev/null +++ b/src/orisys/sensors/fingerprint.py @@ -0,0 +1,118 @@ +"""Processing and calibration fingerprints. + +A fingerprint answers one question: would two installations compute the same numbers? It is +therefore built from what decides the numbers and nothing else. + +Included: the effective processing backend, the identity of the analysis assets it consumes, and +the sensor configuration that defines the pipeline stages. + +Excluded on purpose: the absolute location of the bundle, its display name, and every +visualization asset. Moving or renaming a bundle must not change the fingerprint, and a missing +scene must not either - neither alters the numbers. The calibration is a separate fingerprint +because a device can be recalibrated without the pipeline changing, and the pipeline can change +without the calibration changing. +""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +from typing import Any, Mapping + +#: Bumped when the payload definition changes in a way that must invalidate stored fingerprints. +FINGERPRINT_ALGORITHM = "orisys-processing-1" + +#: Bumped alongside the payload for the calibration half. +CALIBRATION_FINGERPRINT_ALGORITHM = "orisys-calibration-1" + + +def file_digest(path: Path) -> str: + """SHA-256 of a file's bytes, or a marker when it cannot be read. + + An unreadable required asset is a validation error; returning a marker here keeps the + fingerprint deterministic rather than raising from inside a descriptive call. + """ + try: + return hashlib.sha256(Path(path).read_bytes()).hexdigest() + except OSError as exc: + return f"unreadable:{type(exc).__name__}" + + +def processing_fingerprint( + *, + profile_id: str, + backend_id: str, + config: Mapping[str, Any] | Any, + computational_assets: Mapping[str, Path], +) -> str: + """Fingerprint of everything that decides the computed numbers. + + Asset roles are included with their *relative* role name, never their path: the role says what + the file is for, the path would only say where the folder happens to live. + """ + asset_payload = { + role: file_digest(path) for role, path in sorted(computational_assets.items()) + } + payload = { + "algorithm": FINGERPRINT_ALGORITHM, + "profile_id": str(profile_id), + "backend": str(backend_id), + "config": _stable_config(config), + "computational_assets": asset_payload, + } + return _digest(payload) + + +def _stable_config(config: Mapping[str, Any] | Any) -> Any: + """The config as comparable data, with presentation-only fields dropped. + + ``name``/``version`` identify the document but do not change the numbers, so they are not part + of the processing identity. Everything else is included verbatim: a change to any pipeline + stage must show up, and enumerating the *important* keys is how a silent change gets missed. + """ + if not isinstance(config, Mapping): + return None + skip = {"name", "version", "display_name"} + return {key: value for key, value in config.items() if key not in skip} + + +def _digest(payload: Any) -> str: + text = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def dependency_inventory(profile: Any) -> dict[str, Any]: + """The resolved configuration and asset inventory a reproducible snapshot needs. + + Paths are relative to the bundle so the inventory can be stored, compared and moved without + leaking an installation layout. + """ + root = Path(profile.bundle_root) + + def relative(path: Path) -> str: + try: + return Path(path).relative_to(root).as_posix() + except ValueError: + return Path(path).as_posix() + + return { + "algorithm": FINGERPRINT_ALGORITHM, + "profile_id": profile.profile_id, + "product_family": profile.product_family, + "processing_backend": profile.processing_backend, + "backend_source": profile.backend_source, + "sensor_config": relative(profile.sensor_json_path), + "computational_assets": { + role: {"path": relative(path), "sha256": file_digest(path)} + for role, path in sorted(profile.computational_assets.items()) + }, + "visualization_assets": { + role: relative(path) for role, path in sorted(profile.assets.items()) + }, + "default_calibration": ( + None if profile.default_calibration is None else relative(profile.default_calibration) + ), + "processing_fingerprint": profile.processing_fingerprint, + "calibration_fingerprint": profile.calibration_fingerprint, + } diff --git a/src/orisys/sensors/legacy.py b/src/orisys/sensors/legacy.py new file mode 100644 index 0000000..bc21f99 --- /dev/null +++ b/src/orisys/sensors/legacy.py @@ -0,0 +1,221 @@ +"""Compatibility adapter for bundles that predate the descriptor contract. + +Two shapes are supported, both explicit and both without guessing: + +1. an old-shape ``manifest.json`` - the keys :mod:`orisys.configs` already reads + (``name``, ``sensor_json``, ``assets_dir``, ``viz_pose``). Its identity comes from the + built-in table below, its computational assets from the known layout of ``assets_dir``. +2. no ``manifest.json`` at all - the built-in table alone. + +Anything else raises :class:`DescriptorError`. A directory name is never taken as evidence of a +backend, and an unrecognised bundle is never routed to ``compound_eye`` "because box works that +way". +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Mapping + +from .descriptor import ( + BACKEND_COMPOUND_EYE, + BACKEND_SINGLE_CAMERA, + BACKEND_UV_UNROLL, + CURRENT_DESCRIPTOR_SCHEMA, + DESCRIPTOR_FILENAME, + BundleDescriptor, + DescriptorError, +) + +#: Built-in folder names and what they are, as of the legacy format. +#: +#: ``product_family`` is the operator-facing family. ``backend`` is only the *default*; for a +#: bundle with no descriptor the backend that actually applies is derived from the configuration +#: (see :func:`backend_from_config`), which is what the SDK has always done in +#: ``Sensor._load_config``. The two agree for every bundle shipped today, and a disagreement is +#: reported as a diagnostic rather than silently resolved. +BUILT_IN_PROFILES: Mapping[str, Mapping[str, str]] = { + "box": { + "product_family": "box", + "display_name": "Box", + "backend": BACKEND_COMPOUND_EYE, + }, + "mini": { + "product_family": "mini", + "display_name": "Mini", + "backend": BACKEND_COMPOUND_EYE, + }, + "finger": { + "product_family": "finger", + "display_name": "Finger", + "backend": BACKEND_UV_UNROLL, + }, + "finger-mems": { + "product_family": "finger-mems", + "display_name": "Finger (MEMS)", + "backend": BACKEND_COMPOUND_EYE, + }, +} + +#: Computational assets the UV unroll backend needs, relative to the bundle root. These are the +#: files ``orisys.unroll.precompute.build_or_load_domain`` consumes and the map it produces. +UV_UNROLL_COMPUTATIONAL_ASSETS: Mapping[str, str] = { + "unroll_map": "assets/30fps-0703/unroll_map_uv.npz", +} + +#: Assets that only feed visualization. A missing one never blocks image processing. +DEFAULT_VISUALIZATION_ASSETS: Mapping[str, str] = { + "viz_pose": "assets/visualization/world_to_cam.npy", + "mesh": "assets/visualization/mesh.npz", +} + +#: Old-shape manifest keys this adapter understands. ``assets_dir`` names a folder, not a file, +#: so it is expanded through the tables above rather than read verbatim. +LEGACY_MANIFEST_KEYS: tuple[str, ...] = ("name", "sensor_json", "assets_dir", "viz_pose") + + +def backend_from_config(config: Mapping[str, Any] | Any) -> str: + """The backend the configuration declares, exactly as ``Sensor._load_config`` reads it. + + ``sensor.json.type`` when present, else ``compound_eye`` when the config carries a stitcher + block, else ``single_camera``. This is the existing SDK rule; the adapter reuses it rather + than inventing a second one. + """ + if not isinstance(config, Mapping): + return BACKEND_SINGLE_CAMERA + declared = config.get("type") + if isinstance(declared, str) and declared.strip(): + return _canonical_backend(declared) + return BACKEND_COMPOUND_EYE if "stitcher" in config else BACKEND_SINGLE_CAMERA + + +def _canonical_backend(value: str) -> str: + """Map a config's ``type`` spelling onto a supported backend identifier.""" + text = str(value).strip().lower() + if text in ("compound_eye", "compound-eye", "compound"): + return BACKEND_COMPOUND_EYE + if text in ("single_camera", "camera", "mono"): + return BACKEND_SINGLE_CAMERA + if text in ("uv_unroll", "unroll"): + return BACKEND_UV_UNROLL + raise DescriptorError( + f"unsupported type {value!r}; supported: compound_eye, single_camera, uv_unroll" + ) + + +def _existing(bundle_root: Path, relatives: Mapping[str, str]) -> dict[str, Path]: + """Only the listed assets that are actually present, resolved inside the bundle.""" + out: dict[str, Path] = {} + for role, relative in relatives.items(): + candidate = (bundle_root / relative).resolve() + if candidate.is_file(): + out[role] = candidate + return out + + +def _read_manifest(bundle_root: Path) -> dict[str, Any] | None: + path = bundle_root / DESCRIPTOR_FILENAME + if not path.is_file(): + return None + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None + return data if isinstance(data, dict) else None + + +def describe_legacy_bundle( + bundle_root: str | Path, + *, + config: Mapping[str, Any] | None = None, +) -> BundleDescriptor: + """Describe a pre-descriptor bundle, or raise :class:`DescriptorError`. + + ``config`` is the already-parsed ``sensor.json`` when the caller has it; passing it avoids a + second read and lets the backend follow the configuration rather than a guess. + """ + root = Path(bundle_root).expanduser().resolve() + if not root.is_dir(): + raise DescriptorError(f"not a bundle folder: {root}") + folder = root.name + entry = BUILT_IN_PROFILES.get(folder) + if entry is None: + raise DescriptorError( + f"{folder}: not a built-in bundle and no usable {DESCRIPTOR_FILENAME}. " + "Add a descriptor declaring profile_id, display_name, product_family and " + "processing_backend; a folder name is not enough to identify a profile" + ) + + manifest = _read_manifest(root) + diagnostics: list[str] = [] + if manifest is None: + diagnostics.append( + f"{folder}: no {DESCRIPTOR_FILENAME}; identity taken from the built-in profile table" + ) + else: + unknown = sorted(set(manifest) - set(LEGACY_MANIFEST_KEYS) - {"schema_version"}) + diagnostics.append( + f"{folder}: legacy-shape {DESCRIPTOR_FILENAME}" + + (f" (unrecognised keys ignored: {unknown})" if unknown else "") + ) + + declared_backend = entry["backend"] + effective = ( + backend_from_config(config) + if config is not None + else declared_backend + ) + if effective != declared_backend: + diagnostics.append( + f"{folder}: configuration implies backend {effective!r} while the built-in profile " + f"table says {declared_backend!r}; the configuration wins for processing" + ) + + sensor_json = _resolve_sensor_json(root, manifest) + computational: dict[str, Path] = {} + if effective == BACKEND_UV_UNROLL: + computational = _existing(root, UV_UNROLL_COMPUTATIONAL_ASSETS) + if not computational: + diagnostics.append( + f"{folder}: uv_unroll is declared but no computational asset is present " + f"(looked for {sorted(UV_UNROLL_COMPUTATIONAL_ASSETS.values())})" + ) + + assets = _existing(root, DEFAULT_VISUALIZATION_ASSETS) + viz_pose = manifest.get("viz_pose") if manifest else None + if isinstance(viz_pose, str) and viz_pose.strip(): + candidate = (root / viz_pose).resolve() + if candidate.is_file(): + assets["viz_pose"] = candidate + + calibration = root / "force_calibration.json" + + return BundleDescriptor( + schema_version=CURRENT_DESCRIPTOR_SCHEMA, + profile_id=f"orisys.{folder}.legacy", + display_name=entry["display_name"], + product_family=entry["product_family"], + processing_backend=effective, + bundle_root=root, + sensor_json=sensor_json, + default_calibration=calibration if calibration.is_file() else None, + computational_assets=computational, + assets=assets, + from_manifest=manifest is not None, + backend_source=( + "config" if config is not None and effective != declared_backend else "builtin-table" + ), + diagnostics=tuple(diagnostics), + ) + + +def _resolve_sensor_json(root: Path, manifest: Mapping[str, Any] | None) -> Path: + """``sensor.json``, honouring a legacy manifest's ``sensor_json`` when it names a file.""" + if manifest is not None: + raw = manifest.get("sensor_json") + if isinstance(raw, str) and raw.strip(): + candidate = (root / raw).resolve() + if candidate.is_file(): + return candidate + return (root / "sensor.json").resolve() diff --git a/src/orisys/sensors/resolve.py b/src/orisys/sensors/resolve.py new file mode 100644 index 0000000..5f586bd --- /dev/null +++ b/src/orisys/sensors/resolve.py @@ -0,0 +1,585 @@ +"""Resolve one sensor bundle into a validated, reproducible profile. + +``resolve_bundle`` is the single entry point for both live acquisition and offline video +analysis. It answers, for one bundle folder: + +* what the profile *is* - profile identity, product family, processing backend; +* what it *needs* - the computational assets and the configuration that decide the numbers; +* what it *can do* - a :class:`~orisys.sensors.capabilities.Capabilities` report, per axis where + that matters; +* what it *would compute* - a processing fingerprint that survives being moved or renamed. + +The resolved profile is plain data: no Qt, no Studio preferences, no device handle. A device +instance contributes only its calibration selection, which is why the same profile serves a live +camera and a recorded clip. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Mapping + +from .. import configs as sdk_configs +from .capabilities import ( + AxisAvailability, + Capabilities, + Issue, + ValidationReport, + ok as _ok, + report as _report, + unavailable, +) +from .calibration import ( + CalibrationSelection, + ResolvedCalibration, + axis_availability, + resolve_calibration, +) +from .descriptor import ( + BACKEND_COMPOUND_EYE, + BACKEND_SINGLE_CAMERA, + BACKEND_UV_UNROLL, + DESCRIPTOR_FILENAME, + SUPPORTED_BACKENDS, + BundleDescriptor, + DescriptorError, + parse_descriptor, +) +from .fingerprint import dependency_inventory, processing_fingerprint +from .legacy import describe_legacy_bundle + +#: A video/analysis pipeline always needs an image and a way to measure motion. +_MIN_CAMERA_KEYS = ("resolution_stitch",) +_MIN_TRACKER_KEYS = ("algorithm",) + + +@dataclass(frozen=True) +class ResolvedSensorProfile: + """One bundle, resolved: identity, dependencies, calibration, capabilities, fingerprint.""" + + schema_version: int + profile_id: str + display_name: str + product_family: str + processing_backend: str + bundle_root: Path + sensor_json_path: Path + config: Mapping[str, Any] + computational_assets: Mapping[str, Path] + assets: Mapping[str, Path] + default_calibration: Path | None + calibration: ResolvedCalibration + capabilities: Capabilities + validation: ValidationReport + processing_fingerprint: str + calibration_fingerprint: str + backend_source: str = "manifest" + descriptor_diagnostics: tuple[str, ...] = () + diagnostics: tuple[str, ...] = field(default_factory=tuple) + + # ---------------------------------------------------------------- helpers + def capability(self, name: str): + """Look one capability up by identifier.""" + from .capabilities import ( + CALIBRATED_FORCE, + DEFORMATION, + OPTICAL_FLOW, + RAW_FORCE, + VIDEO_PROCESSING, + VISUALIZATION_3D, + ) + + table = { + VIDEO_PROCESSING: self.capabilities.video_processing, + OPTICAL_FLOW: self.capabilities.optical_flow, + RAW_FORCE: self.capabilities.raw_force, + DEFORMATION: self.capabilities.deformation, + VISUALIZATION_3D: self.capabilities.visualization_3d, + } + if name == CALIBRATED_FORCE: + return self.capabilities.calibrated_force + if name not in table: + raise KeyError( + f"unknown capability {name!r}; expected one of " + f"{sorted([*table, CALIBRATED_FORCE])}" + ) + return table[name] + + def inventory(self) -> dict[str, Any]: + """The reproducible snapshot: relative paths, hashes and both fingerprints.""" + return dependency_inventory(self) + + def summary_lines(self) -> list[str]: + """A short block for a status panel or a log header.""" + return [ + f"profile: {self.profile_id} ({self.display_name})", + f"product family: {self.product_family}", + f"processing backend: {self.processing_backend} (from {self.backend_source})", + f"processing fingerprint: {self.processing_fingerprint[:16]}", + f"calibration fingerprint: {self.calibration_fingerprint[:16]}", + *self.validation.summary_lines(), + ] + + +def _read_config( + descriptor: BundleDescriptor, issues: list[Issue] +) -> Mapping[str, Any] | None: + """Parse ``sensor.json`` through the SDK's existing loader.""" + path = descriptor.sensor_json + if not path.is_file(): + issues.append( + Issue( + code="config.missing", + message="the bundle's sensor.json is missing", + path=path, + ) + ) + return None + try: + config = sdk_configs.load_config(str(path)) + except Exception as exc: # noqa: BLE001 - reported as an actionable issue + issues.append( + Issue( + code="config.unreadable", + message=f"sensor.json could not be loaded: {exc}", + path=path, + ) + ) + return None + if not isinstance(config, Mapping): + issues.append( + Issue( + code="config.invalid", + message="sensor.json is not a JSON object", + path=path, + ) + ) + return None + return config + + +def _check_required_assets( + descriptor: BundleDescriptor, issues: list[Issue] +) -> None: + """A missing computational asset fails the profile; a missing visualization one does not.""" + for role, path in sorted(descriptor.computational_assets.items()): + if not path.is_file(): + issues.append( + Issue( + code="assets.computational-missing", + message=( + f"computational asset {role!r} is required by backend " + f"{descriptor.processing_backend!r} but is missing" + ), + path=path, + ) + ) + if descriptor.processing_backend == BACKEND_UV_UNROLL and not descriptor.computational_assets: + issues.append( + Issue( + code="assets.computational-undeclared", + message=( + "backend 'uv_unroll' needs a UV unroll map but the bundle declares no " + "computational asset; expected one under 'computational_assets'" + ), + path=descriptor.bundle_root, + ) + ) + + +def _check_backend_requirements( + descriptor: BundleDescriptor, config: Mapping[str, Any] | None, issues: list[Issue] +) -> None: + """Backend-specific configuration requirements, checked against the config itself.""" + if config is None: + return + backend = descriptor.processing_backend + if backend == BACKEND_COMPOUND_EYE and "stitcher" not in config: + issues.append( + Issue( + code="config.backend-mismatch", + message=( + "backend 'compound_eye' needs a 'stitcher' block to assemble tiles; " + "sensor.json has none" + ), + path=descriptor.sensor_json, + ) + ) + camera = config.get("camera") + if not isinstance(camera, Mapping): + issues.append( + Issue( + code="config.camera-missing", + message="sensor.json has no 'camera' block; resolution and fps are undefined", + path=descriptor.sensor_json, + ) + ) + else: + for key in _MIN_CAMERA_KEYS: + if key not in camera: + issues.append( + Issue( + code="config.camera-incomplete", + message=f"camera.{key} is required for image processing", + path=descriptor.sensor_json, + ) + ) + tracker = config.get("tracker") + if not isinstance(tracker, Mapping): + issues.append( + Issue( + code="config.tracker-missing", + message="sensor.json has no 'tracker' block; optical flow has no algorithm", + path=descriptor.sensor_json, + ) + ) + else: + for key in _MIN_TRACKER_KEYS: + if key not in tracker: + issues.append( + Issue( + code="config.tracker-incomplete", + message=f"tracker.{key} is required for optical flow", + path=descriptor.sensor_json, + ) + ) + + +def _capabilities_for( + descriptor: BundleDescriptor, + config: Mapping[str, Any] | None, + calibration: ResolvedCalibration, + issues: list[Issue], +) -> Capabilities: + """Decide each capability from the facts it actually depends on.""" + backend = descriptor.processing_backend + config_ok = config is not None + + video = ( + _ok(f"{backend} backend") + if config_ok + else unavailable("sensor.json could not be loaded") + ) + flow = ( + _ok(str((config or {}).get("tracker", {}).get("algorithm", "")) or "tracker present") + if config_ok and isinstance((config or {}).get("tracker"), Mapping) + else unavailable("no tracker block in sensor.json") + ) + if backend == BACKEND_SINGLE_CAMERA: + # A raw single-camera frame has no geometry to decompose, so there is no force output. + force = unavailable( + "the single_camera backend produces a frame only; force needs a deformation field" + ) + deformation = unavailable("no deformation field is computed for single_camera") + else: + force = ( + _ok("from the deformation field") + if config_ok + else unavailable("sensor.json could not be loaded") + ) + deformation = _ok("relative depth from the deformation field") + + fn, fx, fy = axis_availability(calibration) + + scene_roles = [role for role in descriptor.assets if role in ("scene", "scene_package")] + if scene_roles: + missing = [ + role for role in scene_roles if not descriptor.assets[role].is_file() + ] + three_d = ( + unavailable("the declared scene package is missing", detail=", ".join(missing)) + if missing + else _ok("scene package resolved") + ) + else: + three_d = unavailable("this profile declares no scene package") + + return Capabilities( + video_processing=video, + optical_flow=flow, + raw_force=force, + calibrated_force=AxisAvailability(fn=fn, fx=fx, fy=fy), + deformation=deformation, + visualization_3d=three_d, + ) + + +def _is_descriptor_manifest(bundle_root: Path) -> bool: + """Whether this folder carries a *new-shape* descriptor rather than a legacy manifest. + + The distinction is ``schema_version``: a manifest without it is the older + ``name``/``sensor_json``/``assets_dir``/``viz_pose`` shape, which the compatibility adapter + describes. Reading a legacy manifest with the strict parser would reject a bundle that is + perfectly usable. + """ + manifest = bundle_root / DESCRIPTOR_FILENAME + if not manifest.is_file(): + return False + try: + data = json.loads(manifest.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + # Unreadable: let the strict parser raise the actionable error. + return True + return isinstance(data, Mapping) and "schema_version" in data + + +def _peek_config(bundle_root: Path) -> Mapping[str, Any] | None: + """The bundle's configuration, when it can be read, for the legacy adapter's backend rule.""" + manifest = bundle_root / DESCRIPTOR_FILENAME + sensor_json = bundle_root / "sensor.json" + if manifest.is_file(): + try: + data = json.loads(manifest.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + data = None + if isinstance(data, Mapping): + named = data.get("sensor_json") + if isinstance(named, str) and named.strip(): + candidate = bundle_root / named + if candidate.is_file(): + sensor_json = candidate + if not sensor_json.is_file(): + return None + try: + return sdk_configs.load_config(str(sensor_json)) + except Exception: # noqa: BLE001 - the resolver reports an unreadable config itself + return None + + +def resolve_bundle( + bundle_path: str | Path, + *, + calibration_selection: CalibrationSelection | str | Path | None = None, +) -> ResolvedSensorProfile: + """Resolve a bundle folder into a :class:`ResolvedSensorProfile`. + + ``bundle_path`` is a folder. Its descriptor is read when present; otherwise the documented + legacy adapter describes it, which raises for anything that is not a known built-in. Nothing + is ever inferred from the folder name alone. + + ``calibration_selection`` is ``BUNDLE_DEFAULT`` (the default), ``NONE``, or an explicit path. + An invalid selection is reported in the profile's validation errors - it is never replaced by + a different calibration. + """ + root = Path(bundle_path).expanduser().resolve() + issues: list[Issue] = [] + warnings: list[Issue] = [] + + # ---- descriptor, with legacy compatibility + descriptor: BundleDescriptor + if _is_descriptor_manifest(root): + try: + descriptor = parse_descriptor(root) + except DescriptorError as exc: + raise DescriptorError(f"{root}: {exc}") from exc + else: + # No manifest, or an old-shape one (the finger bundle ships ``name``/``assets_dir``/ + # ``viz_pose`` and no schema_version). Either way the compatibility adapter describes it. + descriptor = describe_legacy_bundle(root) + + warnings.extend( + Issue(code="descriptor.legacy", message=text, path=root) + for text in descriptor.diagnostics + ) + + # ---- configuration + config = _read_config(descriptor, issues) + + # A legacy descriptor's backend came from the built-in table when no config was available at + # describe time. Now that the config is parsed, let the configuration decide, and say so when + # the two disagree rather than picking one silently. + backend = descriptor.processing_backend + backend_source = descriptor.backend_source + if not descriptor.from_manifest and config is not None: + from .legacy import backend_from_config + + try: + from_config = backend_from_config(config) + except DescriptorError as exc: + issues.append( + Issue(code="config.backend", message=str(exc), path=descriptor.sensor_json) + ) + from_config = backend + if from_config != backend: + warnings.append( + Issue( + code="descriptor.backend-mismatch", + message=( + f"built-in profile table says {backend!r} while sensor.json implies " + f"{from_config!r}; using {from_config!r}" + ), + path=descriptor.sensor_json, + ) + ) + backend = from_config + backend_source = "config" + + if backend not in SUPPORTED_BACKENDS: # defensive: both sources already validate + issues.append( + Issue( + code="backend.unsupported", + message=f"unsupported processing backend {backend!r}", + path=root, + ) + ) + + # ---- required assets and backend requirements + _check_required_assets(descriptor, issues) + _check_backend_requirements(descriptor, config, issues) + + # ---- calibration + calibration = resolve_calibration( + bundle_root=descriptor.bundle_root, + declared_default=descriptor.default_calibration, + profile_id=descriptor.profile_id, + product_family=descriptor.product_family, + selection=calibration_selection, + ) + for issue in calibration.issues: + # A calibration that was merely *absent* is a capability statement, not a fault; an + # explicitly selected calibration that failed is an error. + if issue.code in ("calibration.absent",): + warnings.append(issue) + elif issue.code in ( + "calibration.applicability-unknown", + "calibration.family-mismatch", + ): + warnings.append(issue) + else: + issues.append(issue) + if calibration.document is None and calibration.axes: + issues.append( + Issue( + code="calibration.unapplied", + message="calibration declares axes but produced no document", + path=calibration.path, + ) + ) + + capabilities = _capabilities_for(descriptor, config, calibration, issues) + + validation = _report(issues, warnings, capabilities) + + fingerprint = processing_fingerprint( + profile_id=descriptor.profile_id, + backend_id=backend, + config=config or {}, + computational_assets=descriptor.computational_assets, + ) + + diagnostics = tuple( + [*descriptor.diagnostics] + + [str(issue) for issue in warnings] + ) + + return ResolvedSensorProfile( + schema_version=descriptor.schema_version, + profile_id=descriptor.profile_id, + display_name=descriptor.display_name, + product_family=descriptor.product_family, + processing_backend=backend, + bundle_root=descriptor.bundle_root, + sensor_json_path=descriptor.sensor_json, + config=config or {}, + computational_assets=descriptor.computational_assets, + assets=descriptor.assets, + default_calibration=descriptor.default_calibration, + calibration=calibration, + capabilities=capabilities, + validation=validation, + processing_fingerprint=fingerprint, + calibration_fingerprint=calibration.fingerprint, + backend_source=backend_source, + descriptor_diagnostics=descriptor.diagnostics, + diagnostics=diagnostics, + ) + + +def validate_profile(profile: ResolvedSensorProfile) -> ValidationReport: + """Re-validate a resolved profile against the filesystem as it is now. + + Resolution already validated; this exists for the caller that stored a profile (or resolved it + earlier) and wants to know whether it is still usable before starting a session. + """ + issues: list[Issue] = [] + warnings: list[Issue] = [Issue(code="validation.warning", message=text) for text in profile.diagnostics] + + if not profile.bundle_root.is_dir(): + issues.append( + Issue( + code="bundle.missing", + message="the bundle folder no longer exists", + path=profile.bundle_root, + ) + ) + if not profile.sensor_json_path.is_file(): + issues.append( + Issue( + code="config.missing", + message="sensor.json is missing", + path=profile.sensor_json_path, + ) + ) + for role, path in sorted(profile.computational_assets.items()): + if not path.is_file(): + issues.append( + Issue( + code="assets.computational-missing", + message=f"computational asset {role!r} is missing", + path=path, + ) + ) + if profile.processing_backend not in SUPPORTED_BACKENDS: + issues.append( + Issue( + code="backend.unsupported", + message=f"unsupported processing backend {profile.processing_backend!r}", + ) + ) + if profile.calibration.document is None and profile.calibration.axes: + issues.append( + Issue( + code="calibration.unapplied", + message="the recorded calibration is not applied", + path=profile.calibration.path, + ) + ) + + # A record of the *current* fingerprint: a stored profile whose assets changed is stale. + current = processing_fingerprint( + profile_id=profile.profile_id, + backend_id=profile.processing_backend, + config=profile.config, + computational_assets=profile.computational_assets, + ) + if current != profile.processing_fingerprint: + issues.append( + Issue( + code="fingerprint.stale", + message=( + "a computational asset changed since this profile was resolved; " + "resolve it again before using it" + ), + path=profile.bundle_root, + ) + ) + + return _report(issues, warnings, profile.capabilities) + + +def resolve_named_bundle( + name: str, + *, + calibration_selection: CalibrationSelection | str | Path | None = None, +) -> ResolvedSensorProfile | None: + """Resolve a built-in bundle by its registry name, or ``None`` when it is not installed. + + This is the compatibility bridge for callers that still hold a product name. It resolves the + folder through :mod:`orisys.configs` and then goes through the same path as any other bundle. + """ + bundle = sdk_configs.resolve_sensor_bundle(str(name)) + if bundle is None: + return None + return resolve_bundle(bundle.root, calibration_selection=calibration_selection) diff --git a/tests/test_sensor_bundle.py b/tests/test_sensor_bundle.py new file mode 100644 index 0000000..1adda6d --- /dev/null +++ b/tests/test_sensor_bundle.py @@ -0,0 +1,458 @@ +"""Tests for the sensor bundle contract: descriptor, resolver, registry, fingerprints. + +Each test states the property it protects, because the point of this layer is that a product name +stops deciding which pipeline runs. The bundles built here are small: a resolver test does not need +a real camera configuration, only the specific files the rule under test reads. +""" + +from __future__ import annotations + +import json +import shutil +from pathlib import Path + +import pytest + +from orisys.sensors import ( + BACKEND_COMPOUND_EYE, + BACKEND_SINGLE_CAMERA, + BACKEND_UV_UNROLL, + BUILT_IN_PROFILES, + CalibrationSelection, + DescriptorError, + ProcessorCreationError, + backend_from_config, + create_processor, + known_backends, + parse_descriptor, + resolve_bundle, + spec_for, + validate_profile, +) + +REPO_ROOT = Path(__file__).resolve().parents[1] +SDK_SENSORS = REPO_ROOT / "sensors" + +MINIMAL_CONFIG = { + "name": "prototype", + "version": "1.0", + "type": "compound_eye", + "camera": {"resolution_stitch": [64, 64], "fps": 30}, + "stitcher": {"index": [0]}, + "tracker": {"algorithm": "dis"}, + "contact": {}, +} + + +def _write_bundle( + root: Path, + *, + descriptor: dict | None = None, + config: dict | None = None, + extra: dict[str, bytes] | None = None, + legacy_manifest: dict | None = None, +) -> Path: + """A bundle folder: ``sensor.json`` plus whichever manifest shape is asked for.""" + root.mkdir(parents=True, exist_ok=True) + (root / "sensor.json").write_text( + json.dumps(config if config is not None else MINIMAL_CONFIG), encoding="utf-8" + ) + if descriptor is not None: + (root / "manifest.json").write_text(json.dumps(descriptor), encoding="utf-8") + elif legacy_manifest is not None: + (root / "manifest.json").write_text( + json.dumps(legacy_manifest), encoding="utf-8" + ) + for relative, payload in (extra or {}).items(): + target = root / relative + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(payload) + return root + + +def _descriptor(**overrides) -> dict: + payload = { + "schema_version": 1, + "profile_id": "test.profile", + "display_name": "Test Profile", + "product_family": "mini", + "processing_backend": BACKEND_COMPOUND_EYE, + "sensor_json": "sensor.json", + } + payload.update(overrides) + return payload + + +# ------------------------------------------------------------------ descriptor + + +def test_descriptor_is_required_to_identify_an_unknown_folder(tmp_path: Path): + """A folder name is not evidence of a backend.""" + root = _write_bundle(tmp_path / "mystery") + with pytest.raises(DescriptorError) as excinfo: + resolve_bundle(root) + assert "not a built-in bundle" in str(excinfo.value) + + +def test_descriptor_rejects_unknown_backend_and_schema_version(tmp_path: Path): + root = _write_bundle( + tmp_path / "b", descriptor=_descriptor(processing_backend="nonsense") + ) + with pytest.raises(DescriptorError) as excinfo: + parse_descriptor(root) + assert "unsupported processing_backend" in str(excinfo.value) + + root2 = _write_bundle(tmp_path / "c", descriptor=_descriptor(schema_version=99)) + with pytest.raises(DescriptorError) as excinfo2: + parse_descriptor(root2) + assert "unsupported descriptor schema_version" in str(excinfo2.value) + + +def test_descriptor_rejects_paths_that_leave_the_bundle(tmp_path: Path): + root = _write_bundle( + tmp_path / "d", + descriptor=_descriptor(computational_assets={"map": "../outside.npz"}), + ) + with pytest.raises(DescriptorError) as excinfo: + parse_descriptor(root) + assert "escapes the bundle folder" in str(excinfo.value) + + root2 = _write_bundle( + tmp_path / "e", + descriptor=_descriptor(computational_assets={"map": str(tmp_path / "abs.npz")}), + ) + with pytest.raises(DescriptorError) as excinfo2: + parse_descriptor(root2) + assert "absolute" in str(excinfo2.value) + + +def test_descriptor_round_trips_with_relative_paths(tmp_path: Path): + root = _write_bundle( + tmp_path / "rt", + descriptor=_descriptor( + computational_assets={"map": "assets/map.npz"}, + assets={"scene": "scene/x.oryscene"}, + ), + extra={"assets/map.npz": b"payload", "scene/x.oryscene": b"scene"}, + ) + descriptor = parse_descriptor(root) + payload = descriptor.as_dict() + assert payload["computational_assets"] == {"map": "assets/map.npz"} + assert payload["assets"] == {"scene": "scene/x.oryscene"} + assert not Path(payload["sensor_json"]).is_absolute() + + +# ------------------------------------------------------------------ legacy adapter + + +@pytest.mark.parametrize("folder", sorted(BUILT_IN_PROFILES)) +def test_every_built_in_resolves_through_compatibility(folder: str): + """The acceptance rule: existing supported bundle formats keep loading.""" + root = SDK_SENSORS / folder + if not root.is_dir(): + pytest.skip(f"{folder} is not installed in this checkout") + profile = resolve_bundle(root) + assert profile.product_family == BUILT_IN_PROFILES[folder]["product_family"] + assert profile.processing_backend in known_backends() + assert profile.validation.state in ("ok", "degraded", "failed") + assert profile.processing_fingerprint + + +def test_backend_follows_the_configuration_not_the_folder_name(): + """finger's sensor.json says single_camera; its backend comes from the SDK's own rule.""" + assert backend_from_config({"type": "single_camera"}) == BACKEND_SINGLE_CAMERA + assert backend_from_config({"stitcher": {}}) == BACKEND_COMPOUND_EYE + assert backend_from_config({}) == BACKEND_SINGLE_CAMERA + assert backend_from_config({"type": "uv_unroll"}) == BACKEND_UV_UNROLL + with pytest.raises(DescriptorError): + backend_from_config({"type": "made_up"}) + + +def test_single_camera_is_never_read_as_a_product_family(): + """``single_camera`` describes the image arrangement, not what the product is.""" + profile = resolve_bundle(SDK_SENSORS / "finger") + assert profile.product_family == "finger" + assert profile.processing_backend == BACKEND_UV_UNROLL + + +def test_legacy_manifest_is_read_by_the_adapter_not_the_strict_parser(tmp_path: Path): + """finger ships an old-shape manifest; it must not be rejected for lacking schema_version.""" + root = _write_bundle( + tmp_path / "finger", + legacy_manifest={ + "name": "finger", + "sensor_json": "sensor.json", + "assets_dir": "assets/30fps-0703", + "viz_pose": "assets/visualization/world_to_cam.npy", + }, + config={**MINIMAL_CONFIG, "type": "uv_unroll"}, + extra={"assets/visualization/world_to_cam.npy": b"pose"}, + ) + profile = resolve_bundle(root) + assert profile.product_family == "finger" + # Declared uv_unroll without the map: an error, not a silent downgrade. + assert profile.validation.state == "failed" + assert any( + issue.code == "assets.computational-undeclared" + for issue in profile.validation.errors + ) + # The viz pose is picked up as a visualization asset, never as computational. + assert "viz_pose" in profile.assets + assert "viz_pose" not in profile.computational_assets + + +# ------------------------------------------------------------------ backend registry + + +def test_registry_refuses_an_unknown_identifier(): + with pytest.raises(ProcessorCreationError) as excinfo: + spec_for("nonsense") + assert "unsupported processing backend" in str(excinfo.value) + + +def test_mini_uses_its_supported_backend_without_pretending_to_be_box(): + profile = resolve_bundle(SDK_SENSORS / "mini") + assert profile.product_family == "mini" + assert profile.processing_backend == BACKEND_COMPOUND_EYE + assert profile.profile_id != resolve_bundle(SDK_SENSORS / "box").profile_id + + +def test_finger_mems_is_assessed_from_its_configuration(tmp_path: Path): + """Its name must not imply MEMS acquisition support. + + The studio-side ``finger-mems`` bundle copies ``box``'s configuration, so the folder name is + the only thing that makes it look special. Its backend has to come from that configuration. + """ + config = {**MINIMAL_CONFIG, "type": "compound_eye", "stitcher": {"index": [0]}} + root = _write_bundle(tmp_path / "finger-mems", config=config) + profile = resolve_bundle(root) + assert profile.product_family == "finger-mems" + assert profile.processing_backend == BACKEND_COMPOUND_EYE + # No calibration and no metadata: raw yes, calibrated no, per axis. + assert profile.capabilities.raw_force.available + assert not profile.capabilities.calibrated_force.any_available() + assert profile.capabilities.calibrated_force.fn.reason + # A UV backend is never inferred from the name. + assert profile.processing_backend != BACKEND_UV_UNROLL + + +def test_create_processor_names_what_is_missing_rather_than_faking_it(): + """uv_unroll needs a UV domain assembled outside a config-driven Sensor.""" + profile = resolve_bundle(SDK_SENSORS / "finger") + with pytest.raises(ProcessorCreationError) as excinfo: + create_processor(profile) + assert "uv_unroll" in str(excinfo.value) + + +# ------------------------------------------------------------------ capabilities + + +def test_missing_optional_scene_does_not_block_processing(tmp_path: Path): + root = _write_bundle( + tmp_path / "sceneless", + descriptor=_descriptor( + computational_assets={"map": "assets/map.npz"}, + assets={"scene": "scene/absent.oryscene"}, + ), + extra={"assets/map.npz": b"map"}, + ) + profile = resolve_bundle(root) + assert profile.validation.usable + assert profile.capabilities.video_processing.available + assert profile.capabilities.optical_flow.available + assert not profile.capabilities.visualization_3d.available + assert "missing" in profile.capabilities.visualization_3d.reason + + +def test_missing_computational_asset_fails_the_profile(tmp_path: Path): + root = _write_bundle( + tmp_path / "nogeom", + descriptor=_descriptor(computational_assets={"map": "assets/absent.npz"}), + ) + profile = resolve_bundle(root) + assert profile.validation.state == "failed" + assert any( + issue.code == "assets.computational-missing" + for issue in profile.validation.errors + ) + + +def test_single_camera_reports_no_force_capability(tmp_path: Path): + config = {key: value for key, value in MINIMAL_CONFIG.items() if key != "stitcher"} + config["type"] = "single_camera" + root = _write_bundle( + tmp_path / "mono", + descriptor=_descriptor( + processing_backend=BACKEND_SINGLE_CAMERA, product_family="custom" + ), + config=config, + ) + profile = resolve_bundle(root) + assert profile.capabilities.video_processing.available + assert not profile.capabilities.raw_force.available + assert "single_camera" in profile.capabilities.raw_force.reason + + +# ------------------------------------------------------------------ calibration + + +def test_explicit_none_is_a_choice_not_a_failure(tmp_path: Path): + root = _write_bundle(tmp_path / "nocal", descriptor=_descriptor()) + profile = resolve_bundle(root, calibration_selection=CalibrationSelection.NONE) + assert profile.calibration.selection is CalibrationSelection.NONE + assert profile.calibration_fingerprint == "none" + assert not profile.capabilities.calibrated_force.any_available() + assert all( + "explicitly uncalibrated" in axis.reason + for axis in ( + profile.capabilities.calibrated_force.fn, + profile.capabilities.calibrated_force.fx, + profile.capabilities.calibrated_force.fy, + ) + ) + + +def test_an_unreadable_selected_calibration_is_an_error_not_a_substitution(tmp_path: Path): + root = _write_bundle(tmp_path / "badcal", descriptor=_descriptor()) + profile = resolve_bundle( + root, calibration_selection=str(tmp_path / "does-not-exist.json") + ) + assert any( + issue.code == "calibration.missing" for issue in profile.validation.errors + ) + assert profile.calibration.document is None + assert profile.validation.state == "failed" + + +def test_legacy_calibration_without_metadata_reports_unknown_applicability(): + """box ships a calibration with no identifying metadata; it applies, but is flagged.""" + root = SDK_SENSORS / "box" + if not (root / "force_calibration.json").is_file(): + pytest.skip("box ships no calibration in this checkout") + profile = resolve_bundle(root) + assert profile.calibration.applied + assert profile.calibration.applicability == "unknown" + assert any( + issue.code == "calibration.applicability-unknown" + for issue in profile.validation.warnings + ) + + +def test_calibrated_force_is_reported_per_axis(tmp_path: Path): + calibration = { + "schema_version": 2, + "unit": "N", + "channels": { + "normal": {"method": "piecewise_linear", "points": [[0.0, 0.0], [10.0, 1.0]]} + }, + } + root = _write_bundle( + tmp_path / "fnonly", + descriptor=_descriptor(default_calibration="force_calibration.json"), + extra={"force_calibration.json": json.dumps(calibration).encode()}, + ) + profile = resolve_bundle(root) + axes = profile.capabilities.calibrated_force + assert axes.fn.available + assert not axes.fx.available + assert not axes.fy.available + assert "no fn curve" not in axes.fx.detail # the detail names the axis that is missing + assert "no fx curve" in axes.fx.detail + + +# ------------------------------------------------------------------ fingerprints + + +def test_relocating_a_bundle_preserves_the_processing_fingerprint(tmp_path: Path): + """Moving or renaming must not change what would be computed.""" + first = _write_bundle( + tmp_path / "one" / "mini", + descriptor=_descriptor(computational_assets={"map": "assets/map.npz"}), + extra={"assets/map.npz": b"shared-payload"}, + ) + second = tmp_path / "elsewhere" / "renamed-profile" + shutil.copytree(first, second) + + a = resolve_bundle(first) + b = resolve_bundle(second) + assert a.processing_fingerprint == b.processing_fingerprint + assert a.bundle_root != b.bundle_root + + +def test_changing_a_computational_asset_changes_the_fingerprint(tmp_path: Path): + root = _write_bundle( + tmp_path / "asset", + descriptor=_descriptor(computational_assets={"map": "assets/map.npz"}), + extra={"assets/map.npz": b"first"}, + ) + before = resolve_bundle(root).processing_fingerprint + (root / "assets" / "map.npz").write_bytes(b"second") + after = resolve_bundle(root).processing_fingerprint + assert before != after + + +def test_display_name_and_visualization_assets_stay_out_of_the_fingerprint(tmp_path: Path): + """A cosmetic change must not invalidate a comparable record.""" + def build(name: str, **descriptor_overrides) -> Path: + overrides = {"computational_assets": {"map": "assets/map.npz"}} + overrides.update(descriptor_overrides) + return _write_bundle( + tmp_path / name, + descriptor=_descriptor(**overrides), + extra={"assets/map.npz": b"map"}, + ) + + plain = build("plain", display_name="One") + fancy = build( + "fancy", + display_name="A Much Nicer Name", + assets={"scene": "scene/absent.oryscene"}, + ) + assert ( + resolve_bundle(plain).processing_fingerprint + == resolve_bundle(fancy).processing_fingerprint + ) + + +def test_changing_the_configuration_changes_the_fingerprint(tmp_path: Path): + root = _write_bundle(tmp_path / "cfg", descriptor=_descriptor()) + before = resolve_bundle(root).processing_fingerprint + changed = json.loads((root / "sensor.json").read_text(encoding="utf-8")) + changed["tracker"]["algorithm"] = "raft" + (root / "sensor.json").write_text(json.dumps(changed), encoding="utf-8") + after = resolve_bundle(root).processing_fingerprint + assert before != after + + +def test_renaming_the_config_does_not_change_the_fingerprint(tmp_path: Path): + """``name``/``version`` identify the document; they do not change the numbers.""" + root = _write_bundle(tmp_path / "named", descriptor=_descriptor()) + before = resolve_bundle(root).processing_fingerprint + renamed = json.loads((root / "sensor.json").read_text(encoding="utf-8")) + renamed["name"] = "something-else" + renamed["version"] = "9.9" + (root / "sensor.json").write_text(json.dumps(renamed), encoding="utf-8") + assert resolve_bundle(root).processing_fingerprint == before + + +def test_validate_profile_flags_assets_that_changed_since_resolution(tmp_path: Path): + root = _write_bundle( + tmp_path / "stale", + descriptor=_descriptor(computational_assets={"map": "assets/map.npz"}), + extra={"assets/map.npz": b"original"}, + ) + profile = resolve_bundle(root) + assert validate_profile(profile).state in ("ok", "degraded") + + (root / "assets" / "map.npz").write_bytes(b"tampered") + report = validate_profile(profile) + assert any(issue.code == "fingerprint.stale" for issue in report.errors) + + +def test_inventory_is_bundle_relative(tmp_path: Path): + profile = resolve_bundle(SDK_SENSORS / "box") + inventory = profile.inventory() + assert inventory["profile_id"] == profile.profile_id + assert inventory["processing_fingerprint"] == profile.processing_fingerprint + assert not str(inventory["sensor_config"]).startswith(str(profile.bundle_root)) + assert "/" in inventory["sensor_config"] or "\\" in inventory["sensor_config"] or inventory["sensor_config"]