orisys-tech/orisys-sdk
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add a sensor bundle contract separating family, backend and profile

Browse Source 
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.
This commit is contained in:
ykcheungab
2026-10-04 20:55:39 +08:00
parent 6e3d509239
commit 230b04a79d
10 changed files with 2679 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/orisys/sensors/__init__.py
+146
View File
@@ -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",
]
src/orisys/sensors/backends.py
+285
View File
@@ -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,
)
src/orisys/sensors/calibration.py
+280
View File
@@ -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()
src/orisys/sensors/capabilities.py
+209
View File
@@ -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,
)
src/orisys/sensors/descriptor.py
+216
View File
@@ -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",
)
src/orisys/sensors/fingerprint.py
+118
View File
@@ -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,
}
src/orisys/sensors/legacy.py
+221
View File
@@ -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()
src/orisys/sensors/resolve.py
+585
View File
@@ -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)