1130 lines
44 KiB
Python
1130 lines
44 KiB
Python
"""Strict object-centric perception contracts for Mission Core Milestone 4.
|
|
|
|
These documents are the product boundary. They intentionally contain no LAB,
|
|
experiment, provider or web identity and fail closed on unknown JSON fields.
|
|
Semantic labels and provider tracklets are optional diagnostics; neither is an
|
|
occupancy identity or a downstream persistence contract.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import re
|
|
from dataclasses import dataclass
|
|
from enum import StrEnum
|
|
from typing import Final
|
|
|
|
SOURCE_ENVELOPE_SCHEMA: Final = "missioncore.source-envelope/v1"
|
|
OBJECT_PROPOSAL_SCHEMA: Final = "missioncore.object-proposal-2d/v1"
|
|
OBSTACLE_OBSERVATION_SCHEMA: Final = "missioncore.obstacle-observation/v1"
|
|
TEMPORAL_OBSTACLE_SCHEMA: Final = "missioncore.temporal-obstacle/v1"
|
|
LOCAL_OBSTACLE_MAP_SCHEMA: Final = "missioncore.local-obstacle-map/v1"
|
|
THREAT_ASSESSMENT_SCHEMA: Final = "missioncore.threat-assessment/v1"
|
|
|
|
_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,159}$")
|
|
class PerceptionContractError(ValueError):
|
|
"""A value violates the source-neutral object perception contract."""
|
|
|
|
|
|
class ClockBasis(StrEnum):
|
|
RECORDED_HOST = "recorded-host"
|
|
LIVE_HOST = "live-host"
|
|
DEVICE = "device"
|
|
|
|
|
|
class ModalityOutcome(StrEnum):
|
|
AVAILABLE = "available"
|
|
UNAVAILABLE = "unavailable"
|
|
DROPPED = "dropped"
|
|
STALE = "stale"
|
|
NOT_EXPECTED = "not-expected"
|
|
|
|
|
|
class EvidenceBasis(StrEnum):
|
|
CAMERA = "camera"
|
|
LIDAR = "lidar"
|
|
FUSED = "fused"
|
|
CONFLICT = "conflict"
|
|
|
|
|
|
class EvidenceCurrentness(StrEnum):
|
|
CURRENT = "current"
|
|
HELD = "held"
|
|
STALE = "stale"
|
|
UNAVAILABLE = "unavailable"
|
|
|
|
|
|
class TemporalState(StrEnum):
|
|
CURRENT = "current"
|
|
RETAINED = "retained"
|
|
HELD = "held"
|
|
EXPIRED = "expired"
|
|
|
|
|
|
class MotionState(StrEnum):
|
|
MOVING = "moving"
|
|
STATIONARY = "stationary"
|
|
UNKNOWN = "unknown"
|
|
|
|
|
|
class QualificationState(StrEnum):
|
|
QUALIFIED = "qualified"
|
|
UNQUALIFIED = "unqualified"
|
|
|
|
|
|
class CorridorIntersection(StrEnum):
|
|
INTERSECTS = "intersects"
|
|
CLEAR = "clear"
|
|
UNKNOWN = "unknown"
|
|
|
|
|
|
class ThreatDecision(StrEnum):
|
|
THREAT = "threat"
|
|
NOT_THREAT = "not-threat"
|
|
UNKNOWN = "unknown"
|
|
|
|
|
|
class ThreatAuthority(StrEnum):
|
|
REPLAY_SIMULATED = "replay-simulated"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class TimestampBundle:
|
|
utc_ns: int
|
|
monotonic_ns: int
|
|
source_ns: int
|
|
clock_basis: ClockBasis
|
|
|
|
def __post_init__(self) -> None:
|
|
_nonnegative_integer(self.utc_ns, "UTC timestamp")
|
|
_nonnegative_integer(self.monotonic_ns, "monotonic timestamp")
|
|
_nonnegative_integer(self.source_ns, "source timestamp")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"utc_ns": self.utc_ns,
|
|
"monotonic_ns": self.monotonic_ns,
|
|
"source_ns": self.source_ns,
|
|
"clock_basis": self.clock_basis.value,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> TimestampBundle:
|
|
document = _object(value, "timestamps")
|
|
_exact_keys(
|
|
document,
|
|
{"utc_ns", "monotonic_ns", "source_ns", "clock_basis"},
|
|
"timestamps",
|
|
)
|
|
return cls(
|
|
utc_ns=_integer(document, "utc_ns"),
|
|
monotonic_ns=_integer(document, "monotonic_ns"),
|
|
source_ns=_integer(document, "source_ns"),
|
|
clock_basis=_enum(ClockBasis, document.get("clock_basis"), "clock basis"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ModalityStatus:
|
|
available: bool
|
|
outcome: ModalityOutcome
|
|
reason: str
|
|
|
|
def __post_init__(self) -> None:
|
|
_identifier(self.reason, "modality reason")
|
|
if self.available is not (self.outcome is ModalityOutcome.AVAILABLE):
|
|
raise PerceptionContractError("modality availability and outcome disagree")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"available": self.available,
|
|
"outcome": self.outcome.value,
|
|
"reason": self.reason,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> ModalityStatus:
|
|
document = _object(value, "modality status")
|
|
_exact_keys(document, {"available", "outcome", "reason"}, "modality status")
|
|
return cls(
|
|
available=_boolean(document, "available"),
|
|
outcome=_enum(ModalityOutcome, document.get("outcome"), "modality outcome"),
|
|
reason=_string(document, "reason"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class BoundingRegion2D:
|
|
x_min: float
|
|
y_min: float
|
|
x_max: float
|
|
y_max: float
|
|
|
|
def __post_init__(self) -> None:
|
|
values = tuple(_finite_number(value, "bounding region") for value in self.as_tuple())
|
|
if values[0] < 0.0 or values[1] < 0.0 or values[2] <= values[0] or values[3] <= values[1]:
|
|
raise PerceptionContractError("bounding region is invalid")
|
|
|
|
def as_tuple(self) -> tuple[float, float, float, float]:
|
|
return (self.x_min, self.y_min, self.x_max, self.y_max)
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"x_min": self.x_min,
|
|
"y_min": self.y_min,
|
|
"x_max": self.x_max,
|
|
"y_max": self.y_max,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> BoundingRegion2D:
|
|
document = _object(value, "bounding region")
|
|
_exact_keys(document, {"x_min", "y_min", "x_max", "y_max"}, "bounding region")
|
|
return cls(
|
|
x_min=_number(document, "x_min"),
|
|
y_min=_number(document, "y_min"),
|
|
x_max=_number(document, "x_max"),
|
|
y_max=_number(document, "y_max"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class MetricGeometry:
|
|
coordinate_frame: str
|
|
centroid_xyz_m: tuple[float, float, float]
|
|
range_m: float
|
|
covariance_diagonal_m2: tuple[float, float, float]
|
|
|
|
def __post_init__(self) -> None:
|
|
_identifier(self.coordinate_frame, "coordinate frame")
|
|
_vector3(self.centroid_xyz_m, "metric centroid")
|
|
if _finite_number(self.range_m, "metric range") <= 0.0:
|
|
raise PerceptionContractError("metric range must be positive")
|
|
covariance = _vector3(self.covariance_diagonal_m2, "metric covariance")
|
|
if any(value < 0.0 for value in covariance):
|
|
raise PerceptionContractError("metric covariance must be nonnegative")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"coordinate_frame": self.coordinate_frame,
|
|
"centroid_xyz_m": list(self.centroid_xyz_m),
|
|
"range_m": self.range_m,
|
|
"covariance_diagonal_m2": list(self.covariance_diagonal_m2),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> MetricGeometry:
|
|
document = _object(value, "metric geometry")
|
|
_exact_keys(
|
|
document,
|
|
{"coordinate_frame", "centroid_xyz_m", "range_m", "covariance_diagonal_m2"},
|
|
"metric geometry",
|
|
)
|
|
return cls(
|
|
coordinate_frame=_string(document, "coordinate_frame"),
|
|
centroid_xyz_m=_number_vector3(document.get("centroid_xyz_m"), "metric centroid"),
|
|
range_m=_number(document, "range_m"),
|
|
covariance_diagonal_m2=_number_vector3(
|
|
document.get("covariance_diagonal_m2"),
|
|
"metric covariance",
|
|
),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class FalseAuthority:
|
|
ground_truth: bool = False
|
|
physical_live: bool = False
|
|
physical_collision_accepted: bool = False
|
|
commands_enabled: bool = False
|
|
actuation_allowed: bool = False
|
|
navigation_or_safety_accepted: bool = False
|
|
|
|
def __post_init__(self) -> None:
|
|
if any(
|
|
(
|
|
self.ground_truth,
|
|
self.physical_live,
|
|
self.physical_collision_accepted,
|
|
self.commands_enabled,
|
|
self.actuation_allowed,
|
|
self.navigation_or_safety_accepted,
|
|
)
|
|
):
|
|
raise PerceptionContractError(
|
|
"Milestone 4 cannot publish physical or command authority"
|
|
)
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"ground_truth": self.ground_truth,
|
|
"physical_live": self.physical_live,
|
|
"physical_collision_accepted": self.physical_collision_accepted,
|
|
"commands_enabled": self.commands_enabled,
|
|
"actuation_allowed": self.actuation_allowed,
|
|
"navigation_or_safety_accepted": self.navigation_or_safety_accepted,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> FalseAuthority:
|
|
document = _object(value, "authority")
|
|
keys = {
|
|
"ground_truth",
|
|
"physical_live",
|
|
"physical_collision_accepted",
|
|
"commands_enabled",
|
|
"actuation_allowed",
|
|
"navigation_or_safety_accepted",
|
|
}
|
|
_exact_keys(document, keys, "authority")
|
|
return cls(**{key: _boolean(document, key) for key in keys})
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SourceEnvelope:
|
|
source_id: str
|
|
session_id: str
|
|
frame_id: str
|
|
sequence: int
|
|
timestamps: TimestampBundle
|
|
source_age_ns: int
|
|
binding_reason: str
|
|
calibration_id: str
|
|
representation_id: str
|
|
image: ModalityStatus
|
|
registered_point_increment: ModalityStatus
|
|
pose: ModalityStatus
|
|
|
|
def __post_init__(self) -> None:
|
|
for value, label in (
|
|
(self.source_id, "source id"),
|
|
(self.session_id, "session id"),
|
|
(self.frame_id, "frame id"),
|
|
(self.binding_reason, "binding reason"),
|
|
(self.calibration_id, "calibration id"),
|
|
(self.representation_id, "representation id"),
|
|
):
|
|
_identifier(value, label)
|
|
_nonnegative_integer(self.sequence, "source sequence")
|
|
_nonnegative_integer(self.source_age_ns, "source age")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"schema_version": SOURCE_ENVELOPE_SCHEMA,
|
|
"source_id": self.source_id,
|
|
"session_id": self.session_id,
|
|
"frame_id": self.frame_id,
|
|
"sequence": self.sequence,
|
|
"timestamps": self.timestamps.to_dict(),
|
|
"source_age_ns": self.source_age_ns,
|
|
"binding_reason": self.binding_reason,
|
|
"calibration_id": self.calibration_id,
|
|
"representation_id": self.representation_id,
|
|
"image": self.image.to_dict(),
|
|
"registered_point_increment": self.registered_point_increment.to_dict(),
|
|
"pose": self.pose.to_dict(),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> SourceEnvelope:
|
|
document = _contract(value, SOURCE_ENVELOPE_SCHEMA, {
|
|
"source_id", "session_id", "frame_id", "sequence", "timestamps",
|
|
"source_age_ns", "binding_reason", "calibration_id", "representation_id",
|
|
"image", "registered_point_increment", "pose",
|
|
}, "source envelope")
|
|
return cls(
|
|
source_id=_string(document, "source_id"),
|
|
session_id=_string(document, "session_id"),
|
|
frame_id=_string(document, "frame_id"),
|
|
sequence=_integer(document, "sequence"),
|
|
timestamps=TimestampBundle.from_dict(document.get("timestamps")),
|
|
source_age_ns=_integer(document, "source_age_ns"),
|
|
binding_reason=_string(document, "binding_reason"),
|
|
calibration_id=_string(document, "calibration_id"),
|
|
representation_id=_string(document, "representation_id"),
|
|
image=ModalityStatus.from_dict(document.get("image")),
|
|
registered_point_increment=ModalityStatus.from_dict(
|
|
document.get("registered_point_increment")
|
|
),
|
|
pose=ModalityStatus.from_dict(document.get("pose")),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ObjectProposal2D:
|
|
proposal_id: str
|
|
source_id: str
|
|
frame_id: str
|
|
region: BoundingRegion2D
|
|
objectness: float
|
|
provider_id: str
|
|
model_id: str
|
|
preprocess_id: str
|
|
semantic_hint: str | None = None
|
|
provider_tracklet: str | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
for value, label in (
|
|
(self.proposal_id, "proposal id"),
|
|
(self.source_id, "proposal source id"),
|
|
(self.frame_id, "proposal frame id"),
|
|
(self.provider_id, "detector provider id"),
|
|
(self.model_id, "detector model id"),
|
|
(self.preprocess_id, "preprocess id"),
|
|
):
|
|
_identifier(value, label)
|
|
confidence = _finite_number(self.objectness, "objectness")
|
|
if not 0.0 <= confidence <= 1.0:
|
|
raise PerceptionContractError("objectness must be within [0, 1]")
|
|
_optional_identifier(self.semantic_hint, "semantic hint")
|
|
_optional_identifier(self.provider_tracklet, "provider tracklet")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"schema_version": OBJECT_PROPOSAL_SCHEMA,
|
|
"proposal_id": self.proposal_id,
|
|
"source_id": self.source_id,
|
|
"frame_id": self.frame_id,
|
|
"region": self.region.to_dict(),
|
|
"objectness": self.objectness,
|
|
"provider_id": self.provider_id,
|
|
"model_id": self.model_id,
|
|
"preprocess_id": self.preprocess_id,
|
|
"semantic_hint": self.semantic_hint,
|
|
"provider_tracklet": self.provider_tracklet,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> ObjectProposal2D:
|
|
document = _contract(value, OBJECT_PROPOSAL_SCHEMA, {
|
|
"proposal_id", "source_id", "frame_id", "region", "objectness",
|
|
"provider_id", "model_id", "preprocess_id", "semantic_hint",
|
|
"provider_tracklet",
|
|
}, "object proposal")
|
|
return cls(
|
|
proposal_id=_string(document, "proposal_id"),
|
|
source_id=_string(document, "source_id"),
|
|
frame_id=_string(document, "frame_id"),
|
|
region=BoundingRegion2D.from_dict(document.get("region")),
|
|
objectness=_number(document, "objectness"),
|
|
provider_id=_string(document, "provider_id"),
|
|
model_id=_string(document, "model_id"),
|
|
preprocess_id=_string(document, "preprocess_id"),
|
|
semantic_hint=_optional_string(document.get("semantic_hint"), "semantic hint"),
|
|
provider_tracklet=_optional_string(
|
|
document.get("provider_tracklet"), "provider tracklet"
|
|
),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ObstacleObservation:
|
|
observation_id: str
|
|
occupancy_key: str
|
|
source_id: str
|
|
frame_id: str
|
|
evidence_time_ns: int
|
|
basis: EvidenceBasis
|
|
currentness: EvidenceCurrentness
|
|
occupied_support: bool
|
|
source_point_ids: tuple[int, ...]
|
|
metric_geometry: MetricGeometry | None
|
|
proposal_ids: tuple[str, ...]
|
|
semantic_hint: str | None
|
|
reason_codes: tuple[str, ...]
|
|
authority: FalseAuthority = FalseAuthority()
|
|
|
|
def __post_init__(self) -> None:
|
|
for value, label in (
|
|
(self.observation_id, "observation id"),
|
|
(self.occupancy_key, "occupancy key"),
|
|
(self.source_id, "observation source id"),
|
|
(self.frame_id, "observation frame id"),
|
|
):
|
|
_identifier(value, label)
|
|
_nonnegative_integer(self.evidence_time_ns, "evidence time")
|
|
_unique_nonnegative_integers(self.source_point_ids, "source point ids")
|
|
_unique_identifiers(self.proposal_ids, "proposal ids", allow_empty=True)
|
|
_unique_identifiers(self.reason_codes, "observation reason codes")
|
|
_optional_identifier(self.semantic_hint, "semantic hint")
|
|
if not isinstance(self.occupied_support, bool):
|
|
raise PerceptionContractError("occupied support must be boolean")
|
|
if self.currentness is not EvidenceCurrentness.CURRENT and (
|
|
self.occupied_support or self.source_point_ids or self.metric_geometry is not None
|
|
):
|
|
raise PerceptionContractError(
|
|
"non-current evidence cannot publish current geometry"
|
|
)
|
|
if self.basis is EvidenceBasis.CAMERA:
|
|
if self.occupied_support or self.source_point_ids or self.metric_geometry is not None:
|
|
raise PerceptionContractError("camera-only evidence is non-metric uncertainty")
|
|
if not self.proposal_ids:
|
|
raise PerceptionContractError("camera evidence requires an object proposal")
|
|
elif self.basis is EvidenceBasis.CONFLICT:
|
|
if self.occupied_support or self.source_point_ids or self.metric_geometry is not None:
|
|
raise PerceptionContractError(
|
|
"conflicting evidence cannot publish metric occupancy"
|
|
)
|
|
elif self.currentness is EvidenceCurrentness.CURRENT and (
|
|
not self.occupied_support
|
|
or not self.source_point_ids
|
|
or self.metric_geometry is None
|
|
):
|
|
raise PerceptionContractError(
|
|
"current LiDAR/fused occupancy requires qualified points"
|
|
)
|
|
|
|
@property
|
|
def occupancy_identity(self) -> str:
|
|
"""Identity deliberately excludes semantic labels and provider tracklets."""
|
|
|
|
return self.occupancy_key
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"schema_version": OBSTACLE_OBSERVATION_SCHEMA,
|
|
"observation_id": self.observation_id,
|
|
"occupancy_key": self.occupancy_key,
|
|
"source_id": self.source_id,
|
|
"frame_id": self.frame_id,
|
|
"evidence_time_ns": self.evidence_time_ns,
|
|
"basis": self.basis.value,
|
|
"currentness": self.currentness.value,
|
|
"occupied_support": self.occupied_support,
|
|
"source_point_ids": list(self.source_point_ids),
|
|
"metric_geometry": (
|
|
None if self.metric_geometry is None else self.metric_geometry.to_dict()
|
|
),
|
|
"proposal_ids": list(self.proposal_ids),
|
|
"semantic_hint": self.semantic_hint,
|
|
"reason_codes": list(self.reason_codes),
|
|
"authority": self.authority.to_dict(),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> ObstacleObservation:
|
|
document = _contract(value, OBSTACLE_OBSERVATION_SCHEMA, {
|
|
"observation_id", "occupancy_key", "source_id", "frame_id",
|
|
"evidence_time_ns", "basis", "currentness", "occupied_support",
|
|
"source_point_ids", "metric_geometry", "proposal_ids", "semantic_hint",
|
|
"reason_codes", "authority",
|
|
}, "obstacle observation")
|
|
metric = document.get("metric_geometry")
|
|
return cls(
|
|
observation_id=_string(document, "observation_id"),
|
|
occupancy_key=_string(document, "occupancy_key"),
|
|
source_id=_string(document, "source_id"),
|
|
frame_id=_string(document, "frame_id"),
|
|
evidence_time_ns=_integer(document, "evidence_time_ns"),
|
|
basis=_enum(EvidenceBasis, document.get("basis"), "evidence basis"),
|
|
currentness=_enum(
|
|
EvidenceCurrentness, document.get("currentness"), "evidence currentness"
|
|
),
|
|
occupied_support=_boolean(document, "occupied_support"),
|
|
source_point_ids=_integer_tuple(document.get("source_point_ids"), "source point ids"),
|
|
metric_geometry=None if metric is None else MetricGeometry.from_dict(metric),
|
|
proposal_ids=_string_tuple(document.get("proposal_ids"), "proposal ids"),
|
|
semantic_hint=_optional_string(document.get("semantic_hint"), "semantic hint"),
|
|
reason_codes=_string_tuple(document.get("reason_codes"), "reason codes"),
|
|
authority=FalseAuthority.from_dict(document.get("authority")),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class HistorySample:
|
|
frame_id: str
|
|
evidence_time_ns: int
|
|
centroid_xyz_m: tuple[float, float, float]
|
|
|
|
def __post_init__(self) -> None:
|
|
_identifier(self.frame_id, "history frame id")
|
|
_nonnegative_integer(self.evidence_time_ns, "history evidence time")
|
|
_vector3(self.centroid_xyz_m, "history centroid")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"frame_id": self.frame_id,
|
|
"evidence_time_ns": self.evidence_time_ns,
|
|
"centroid_xyz_m": list(self.centroid_xyz_m),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> HistorySample:
|
|
document = _object(value, "history sample")
|
|
_exact_keys(document, {"frame_id", "evidence_time_ns", "centroid_xyz_m"}, "history sample")
|
|
return cls(
|
|
frame_id=_string(document, "frame_id"),
|
|
evidence_time_ns=_integer(document, "evidence_time_ns"),
|
|
centroid_xyz_m=_number_vector3(document.get("centroid_xyz_m"), "history centroid"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class GridCell:
|
|
x: int
|
|
y: int
|
|
z: int
|
|
|
|
def __post_init__(self) -> None:
|
|
for value in (self.x, self.y, self.z):
|
|
if not isinstance(value, int) or isinstance(value, bool):
|
|
raise PerceptionContractError("grid cell coordinates must be integers")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {"x": self.x, "y": self.y, "z": self.z}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> GridCell:
|
|
document = _object(value, "grid cell")
|
|
_exact_keys(document, {"x", "y", "z"}, "grid cell")
|
|
return cls(x=_integer(document, "x"), y=_integer(document, "y"), z=_integer(document, "z"))
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class TemporalObstacle:
|
|
component_id: str
|
|
identity_scope: str
|
|
state: TemporalState
|
|
ttl_ns: int
|
|
last_hit_ns: int
|
|
age_ns: int
|
|
association_basis: str
|
|
history: tuple[HistorySample, ...]
|
|
cells: tuple[GridCell, ...]
|
|
coordinate_frame: str | None
|
|
last_centroid_xyz_m: tuple[float, float, float] | None
|
|
motion: MotionState
|
|
motion_confidence: float
|
|
motion_reason: str
|
|
semantic_hint: str | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
_identifier(self.component_id, "temporal component id")
|
|
if self.identity_scope != "ephemeral":
|
|
raise PerceptionContractError("temporal component identity must remain ephemeral")
|
|
_positive_integer(self.ttl_ns, "temporal TTL")
|
|
_nonnegative_integer(self.last_hit_ns, "last hit time")
|
|
_nonnegative_integer(self.age_ns, "temporal age")
|
|
_identifier(self.association_basis, "association basis")
|
|
_identifier(self.motion_reason, "motion reason")
|
|
_optional_identifier(self.semantic_hint, "semantic hint")
|
|
confidence = _finite_number(self.motion_confidence, "motion confidence")
|
|
if not 0.0 <= confidence <= 1.0:
|
|
raise PerceptionContractError("motion confidence must be within [0, 1]")
|
|
if not self.history or len(self.history) > 32:
|
|
raise PerceptionContractError("temporal history must contain 1..32 samples")
|
|
if len(set(self.cells)) != len(self.cells):
|
|
raise PerceptionContractError("temporal cells must be unique")
|
|
if self.state is TemporalState.CURRENT:
|
|
if self.age_ns > self.ttl_ns or not self.cells:
|
|
raise PerceptionContractError("current temporal occupancy requires bounded cells")
|
|
elif self.state is TemporalState.RETAINED:
|
|
if not 0 < self.age_ns <= self.ttl_ns or not self.cells:
|
|
raise PerceptionContractError(
|
|
"retained rolling-map occupancy must remain within its bound"
|
|
)
|
|
if self.motion is not MotionState.UNKNOWN:
|
|
raise PerceptionContractError(
|
|
"retained rolling-map occupancy cannot claim object motion"
|
|
)
|
|
elif self.state is TemporalState.HELD:
|
|
if not 0 < self.age_ns <= self.ttl_ns or not self.cells:
|
|
raise PerceptionContractError("held temporal state must remain within TTL")
|
|
elif self.cells:
|
|
raise PerceptionContractError(
|
|
"expired temporal components cannot publish occupied cells"
|
|
)
|
|
if self.cells:
|
|
if self.coordinate_frame is None or self.last_centroid_xyz_m is None:
|
|
raise PerceptionContractError("temporal cells require their last metric geometry")
|
|
_identifier(self.coordinate_frame, "temporal coordinate frame")
|
|
_vector3(self.last_centroid_xyz_m, "temporal centroid")
|
|
elif self.coordinate_frame is not None or self.last_centroid_xyz_m is not None:
|
|
raise PerceptionContractError("temporal metric geometry requires occupied cells")
|
|
if self.motion is MotionState.UNKNOWN and confidence != 0.0:
|
|
raise PerceptionContractError("unknown motion must have zero confidence")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"schema_version": TEMPORAL_OBSTACLE_SCHEMA,
|
|
"component_id": self.component_id,
|
|
"identity_scope": self.identity_scope,
|
|
"state": self.state.value,
|
|
"ttl_ns": self.ttl_ns,
|
|
"last_hit_ns": self.last_hit_ns,
|
|
"age_ns": self.age_ns,
|
|
"association_basis": self.association_basis,
|
|
"history": [item.to_dict() for item in self.history],
|
|
"cells": [item.to_dict() for item in self.cells],
|
|
"coordinate_frame": self.coordinate_frame,
|
|
"last_centroid_xyz_m": (
|
|
None if self.last_centroid_xyz_m is None else list(self.last_centroid_xyz_m)
|
|
),
|
|
"motion": self.motion.value,
|
|
"motion_confidence": self.motion_confidence,
|
|
"motion_reason": self.motion_reason,
|
|
"semantic_hint": self.semantic_hint,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> TemporalObstacle:
|
|
document = _contract(value, TEMPORAL_OBSTACLE_SCHEMA, {
|
|
"component_id", "identity_scope", "state", "ttl_ns", "last_hit_ns",
|
|
"age_ns", "association_basis", "history", "cells", "coordinate_frame",
|
|
"last_centroid_xyz_m", "motion", "motion_confidence", "motion_reason",
|
|
"semantic_hint",
|
|
}, "temporal obstacle")
|
|
centroid = document.get("last_centroid_xyz_m")
|
|
return cls(
|
|
component_id=_string(document, "component_id"),
|
|
identity_scope=_string(document, "identity_scope"),
|
|
state=_enum(TemporalState, document.get("state"), "temporal state"),
|
|
ttl_ns=_integer(document, "ttl_ns"),
|
|
last_hit_ns=_integer(document, "last_hit_ns"),
|
|
age_ns=_integer(document, "age_ns"),
|
|
association_basis=_string(document, "association_basis"),
|
|
history=tuple(HistorySample.from_dict(item) for item in _array(document, "history")),
|
|
cells=tuple(GridCell.from_dict(item) for item in _array(document, "cells")),
|
|
coordinate_frame=_optional_string(document.get("coordinate_frame"), "coordinate frame"),
|
|
last_centroid_xyz_m=(
|
|
None if centroid is None else _number_vector3(centroid, "temporal centroid")
|
|
),
|
|
motion=_enum(MotionState, document.get("motion"), "motion state"),
|
|
motion_confidence=_number(document, "motion_confidence"),
|
|
motion_reason=_string(document, "motion_reason"),
|
|
semantic_hint=_optional_string(document.get("semantic_hint"), "semantic hint"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SourceAccounting:
|
|
source_envelopes: int
|
|
terminal_outcomes: int
|
|
dropped: int
|
|
failed: int
|
|
|
|
def __post_init__(self) -> None:
|
|
for value, label in (
|
|
(self.source_envelopes, "source envelope count"),
|
|
(self.terminal_outcomes, "terminal outcome count"),
|
|
(self.dropped, "drop count"),
|
|
(self.failed, "failure count"),
|
|
):
|
|
_nonnegative_integer(value, label)
|
|
if self.terminal_outcomes + self.dropped + self.failed != self.source_envelopes:
|
|
raise PerceptionContractError("source accounting is not closed")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"source_envelopes": self.source_envelopes,
|
|
"terminal_outcomes": self.terminal_outcomes,
|
|
"dropped": self.dropped,
|
|
"failed": self.failed,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> SourceAccounting:
|
|
document = _object(value, "source accounting")
|
|
_exact_keys(
|
|
document,
|
|
{"source_envelopes", "terminal_outcomes", "dropped", "failed"},
|
|
"source accounting",
|
|
)
|
|
return cls(
|
|
source_envelopes=_integer(document, "source_envelopes"),
|
|
terminal_outcomes=_integer(document, "terminal_outcomes"),
|
|
dropped=_integer(document, "dropped"),
|
|
failed=_integer(document, "failed"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class LocalObstacleMap:
|
|
source_id: str
|
|
session_id: str
|
|
frame_id: str
|
|
graph_id: str
|
|
generated_monotonic_ns: int
|
|
output_age_ns: int
|
|
occupied: tuple[TemporalObstacle, ...]
|
|
unknown: tuple[TemporalObstacle, ...]
|
|
camera_uncertainty: tuple[ObjectProposal2D, ...]
|
|
accounting: SourceAccounting
|
|
free_space_claimed: bool = False
|
|
authority: FalseAuthority = FalseAuthority()
|
|
|
|
def __post_init__(self) -> None:
|
|
for value, label in (
|
|
(self.source_id, "map source id"),
|
|
(self.session_id, "map session id"),
|
|
(self.frame_id, "map frame id"),
|
|
(self.graph_id, "graph id"),
|
|
):
|
|
_identifier(value, label)
|
|
_nonnegative_integer(self.generated_monotonic_ns, "map generation time")
|
|
_nonnegative_integer(self.output_age_ns, "map output age")
|
|
if self.free_space_claimed:
|
|
raise PerceptionContractError("Milestone 4 cannot publish implicit free space")
|
|
if any(
|
|
item.state not in {TemporalState.CURRENT, TemporalState.RETAINED}
|
|
for item in self.occupied
|
|
):
|
|
raise PerceptionContractError(
|
|
"occupied map entries must be current hits or bounded rolling-map retention"
|
|
)
|
|
if any(
|
|
item.state in {TemporalState.CURRENT, TemporalState.RETAINED}
|
|
for item in self.unknown
|
|
):
|
|
raise PerceptionContractError(
|
|
"held or expired temporal entries must remain unknown"
|
|
)
|
|
component_ids = [item.component_id for item in (*self.occupied, *self.unknown)]
|
|
if len(set(component_ids)) != len(component_ids):
|
|
raise PerceptionContractError("map component identities must be unique")
|
|
proposal_ids = [item.proposal_id for item in self.camera_uncertainty]
|
|
if len(set(proposal_ids)) != len(proposal_ids):
|
|
raise PerceptionContractError("camera uncertainty proposals must be unique")
|
|
if any(
|
|
item.source_id != self.source_id or item.frame_id != self.frame_id
|
|
for item in self.camera_uncertainty
|
|
):
|
|
raise PerceptionContractError("camera uncertainty is bound to another source frame")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"schema_version": LOCAL_OBSTACLE_MAP_SCHEMA,
|
|
"source_id": self.source_id,
|
|
"session_id": self.session_id,
|
|
"frame_id": self.frame_id,
|
|
"graph_id": self.graph_id,
|
|
"generated_monotonic_ns": self.generated_monotonic_ns,
|
|
"output_age_ns": self.output_age_ns,
|
|
"occupied": [item.to_dict() for item in self.occupied],
|
|
"unknown": [item.to_dict() for item in self.unknown],
|
|
"camera_uncertainty": [item.to_dict() for item in self.camera_uncertainty],
|
|
"accounting": self.accounting.to_dict(),
|
|
"free_space_claimed": self.free_space_claimed,
|
|
"authority": self.authority.to_dict(),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> LocalObstacleMap:
|
|
document = _contract(value, LOCAL_OBSTACLE_MAP_SCHEMA, {
|
|
"source_id", "session_id", "frame_id", "graph_id",
|
|
"generated_monotonic_ns", "output_age_ns", "occupied", "unknown",
|
|
"camera_uncertainty", "accounting", "free_space_claimed", "authority",
|
|
}, "local obstacle map")
|
|
return cls(
|
|
source_id=_string(document, "source_id"),
|
|
session_id=_string(document, "session_id"),
|
|
frame_id=_string(document, "frame_id"),
|
|
graph_id=_string(document, "graph_id"),
|
|
generated_monotonic_ns=_integer(document, "generated_monotonic_ns"),
|
|
output_age_ns=_integer(document, "output_age_ns"),
|
|
occupied=tuple(
|
|
TemporalObstacle.from_dict(item) for item in _array(document, "occupied")
|
|
),
|
|
unknown=tuple(TemporalObstacle.from_dict(item) for item in _array(document, "unknown")),
|
|
camera_uncertainty=tuple(
|
|
ObjectProposal2D.from_dict(item) for item in _array(document, "camera_uncertainty")
|
|
),
|
|
accounting=SourceAccounting.from_dict(document.get("accounting")),
|
|
free_space_claimed=_boolean(document, "free_space_claimed"),
|
|
authority=FalseAuthority.from_dict(document.get("authority")),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ThreatAssessment:
|
|
assessment_id: str
|
|
component_id: str
|
|
rig_profile_id: str
|
|
corridor_profile_id: str
|
|
qualification: QualificationState
|
|
relative_speed_mps: float | None
|
|
closest_approach_m: float | None
|
|
ttc_seconds: float | None
|
|
corridor_intersection: CorridorIntersection
|
|
decision: ThreatDecision
|
|
reason_codes: tuple[str, ...]
|
|
authority: ThreatAuthority = ThreatAuthority.REPLAY_SIMULATED
|
|
physical_collision_accepted: bool = False
|
|
actuation_allowed: bool = False
|
|
|
|
def __post_init__(self) -> None:
|
|
for value, label in (
|
|
(self.assessment_id, "assessment id"),
|
|
(self.component_id, "assessed component id"),
|
|
(self.rig_profile_id, "rig profile id"),
|
|
(self.corridor_profile_id, "corridor profile id"),
|
|
):
|
|
_identifier(value, label)
|
|
_unique_identifiers(self.reason_codes, "threat reason codes")
|
|
relative_speed = _optional_finite(self.relative_speed_mps, "relative speed")
|
|
closest = _optional_finite(self.closest_approach_m, "closest approach")
|
|
ttc = _optional_finite(self.ttc_seconds, "TTC")
|
|
if closest is not None and closest < 0.0:
|
|
raise PerceptionContractError("closest approach must be nonnegative")
|
|
if ttc is not None and ttc < 0.0:
|
|
raise PerceptionContractError("TTC must be nonnegative")
|
|
if self.physical_collision_accepted or self.actuation_allowed:
|
|
raise PerceptionContractError("replay threat cannot authorize collision or actuation")
|
|
if self.authority is not ThreatAuthority.REPLAY_SIMULATED:
|
|
raise PerceptionContractError("Milestone 4 threat authority must be replay-simulated")
|
|
if self.qualification is QualificationState.UNQUALIFIED:
|
|
if self.decision is not ThreatDecision.UNKNOWN:
|
|
raise PerceptionContractError("unqualified threat evidence must remain unknown")
|
|
if any(value is not None for value in (relative_speed, closest, ttc)):
|
|
raise PerceptionContractError("unqualified threat cannot publish derived metrics")
|
|
if self.decision is ThreatDecision.THREAT and (
|
|
self.qualification is not QualificationState.QUALIFIED
|
|
or self.corridor_intersection is not CorridorIntersection.INTERSECTS
|
|
):
|
|
raise PerceptionContractError("threat requires qualified corridor intersection")
|
|
if self.decision is ThreatDecision.NOT_THREAT and (
|
|
self.qualification is not QualificationState.QUALIFIED
|
|
or self.corridor_intersection is not CorridorIntersection.CLEAR
|
|
):
|
|
raise PerceptionContractError("not-threat requires qualified corridor clearance")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"schema_version": THREAT_ASSESSMENT_SCHEMA,
|
|
"assessment_id": self.assessment_id,
|
|
"component_id": self.component_id,
|
|
"rig_profile_id": self.rig_profile_id,
|
|
"corridor_profile_id": self.corridor_profile_id,
|
|
"qualification": self.qualification.value,
|
|
"relative_speed_mps": self.relative_speed_mps,
|
|
"closest_approach_m": self.closest_approach_m,
|
|
"ttc_seconds": self.ttc_seconds,
|
|
"corridor_intersection": self.corridor_intersection.value,
|
|
"decision": self.decision.value,
|
|
"reason_codes": list(self.reason_codes),
|
|
"authority": self.authority.value,
|
|
"physical_collision_accepted": self.physical_collision_accepted,
|
|
"actuation_allowed": self.actuation_allowed,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> ThreatAssessment:
|
|
document = _contract(value, THREAT_ASSESSMENT_SCHEMA, {
|
|
"assessment_id", "component_id", "rig_profile_id", "corridor_profile_id",
|
|
"qualification", "relative_speed_mps", "closest_approach_m", "ttc_seconds",
|
|
"corridor_intersection", "decision", "reason_codes", "authority",
|
|
"physical_collision_accepted", "actuation_allowed",
|
|
}, "threat assessment")
|
|
return cls(
|
|
assessment_id=_string(document, "assessment_id"),
|
|
component_id=_string(document, "component_id"),
|
|
rig_profile_id=_string(document, "rig_profile_id"),
|
|
corridor_profile_id=_string(document, "corridor_profile_id"),
|
|
qualification=_enum(
|
|
QualificationState, document.get("qualification"), "qualification state"
|
|
),
|
|
relative_speed_mps=_optional_number(
|
|
document.get("relative_speed_mps"), "relative speed"
|
|
),
|
|
closest_approach_m=_optional_number(
|
|
document.get("closest_approach_m"), "closest approach"
|
|
),
|
|
ttc_seconds=_optional_number(document.get("ttc_seconds"), "TTC"),
|
|
corridor_intersection=_enum(
|
|
CorridorIntersection,
|
|
document.get("corridor_intersection"),
|
|
"corridor intersection",
|
|
),
|
|
decision=_enum(ThreatDecision, document.get("decision"), "threat decision"),
|
|
reason_codes=_string_tuple(document.get("reason_codes"), "reason codes"),
|
|
authority=_enum(ThreatAuthority, document.get("authority"), "threat authority"),
|
|
physical_collision_accepted=_boolean(document, "physical_collision_accepted"),
|
|
actuation_allowed=_boolean(document, "actuation_allowed"),
|
|
)
|
|
|
|
|
|
def validate_exclusive_point_ownership(
|
|
observations: tuple[ObstacleObservation, ...],
|
|
) -> None:
|
|
"""Reject one source point being claimed by more than one observation."""
|
|
|
|
owners: dict[int, str] = {}
|
|
for observation in observations:
|
|
for point_id in observation.source_point_ids:
|
|
previous = owners.setdefault(point_id, observation.observation_id)
|
|
if previous != observation.observation_id:
|
|
raise PerceptionContractError(
|
|
f"source point {point_id} has duplicate observation ownership"
|
|
)
|
|
|
|
|
|
def _contract(
|
|
value: object,
|
|
schema: str,
|
|
fields: set[str],
|
|
label: str,
|
|
) -> dict[str, object]:
|
|
document = _object(value, label)
|
|
_exact_keys(document, {"schema_version", *fields}, label)
|
|
if document.get("schema_version") != schema:
|
|
raise PerceptionContractError(f"{label} schema is incompatible")
|
|
return document
|
|
|
|
|
|
def _object(value: object, label: str) -> dict[str, object]:
|
|
if not isinstance(value, dict) or any(not isinstance(key, str) for key in value):
|
|
raise PerceptionContractError(f"{label} must be an object")
|
|
return value
|
|
|
|
|
|
def _exact_keys(document: dict[str, object], expected: set[str], label: str) -> None:
|
|
if set(document) != expected:
|
|
raise PerceptionContractError(f"{label} fields are incompatible")
|
|
|
|
|
|
def _array(document: dict[str, object], key: str) -> list[object]:
|
|
value = document.get(key)
|
|
if not isinstance(value, list):
|
|
raise PerceptionContractError(f"{key} must be an array")
|
|
return value
|
|
|
|
|
|
def _string(document: dict[str, object], key: str) -> str:
|
|
return _string_value(document.get(key), key)
|
|
|
|
|
|
def _string_value(value: object, label: str) -> str:
|
|
if not isinstance(value, str) or not value:
|
|
raise PerceptionContractError(f"{label} must be a nonempty string")
|
|
return value
|
|
|
|
|
|
def _optional_string(value: object, label: str) -> str | None:
|
|
if value is None:
|
|
return None
|
|
return _string_value(value, label)
|
|
|
|
|
|
def _boolean(document: dict[str, object], key: str) -> bool:
|
|
value = document.get(key)
|
|
if not isinstance(value, bool):
|
|
raise PerceptionContractError(f"{key} must be boolean")
|
|
return value
|
|
|
|
|
|
def _integer(document: dict[str, object], key: str) -> int:
|
|
value = document.get(key)
|
|
if not isinstance(value, int) or isinstance(value, bool):
|
|
raise PerceptionContractError(f"{key} must be an integer")
|
|
return value
|
|
|
|
|
|
def _number(document: dict[str, object], key: str) -> float:
|
|
return _finite_number(document.get(key), key)
|
|
|
|
|
|
def _optional_number(value: object, label: str) -> float | None:
|
|
if value is None:
|
|
return None
|
|
return _finite_number(value, label)
|
|
|
|
|
|
def _finite_number(value: object, label: str) -> float:
|
|
if (
|
|
not isinstance(value, (int, float))
|
|
or isinstance(value, bool)
|
|
or not math.isfinite(float(value))
|
|
):
|
|
raise PerceptionContractError(f"{label} must be finite")
|
|
return float(value)
|
|
|
|
|
|
def _optional_finite(value: float | None, label: str) -> float | None:
|
|
return None if value is None else _finite_number(value, label)
|
|
|
|
|
|
def _nonnegative_integer(value: object, label: str) -> int:
|
|
if not isinstance(value, int) or isinstance(value, bool) or value < 0:
|
|
raise PerceptionContractError(f"{label} must be a nonnegative integer")
|
|
return value
|
|
|
|
|
|
def _positive_integer(value: object, label: str) -> int:
|
|
result = _nonnegative_integer(value, label)
|
|
if result == 0:
|
|
raise PerceptionContractError(f"{label} must be positive")
|
|
return result
|
|
|
|
|
|
def _identifier(value: str, label: str) -> str:
|
|
if not isinstance(value, str) or _IDENTIFIER.fullmatch(value) is None:
|
|
raise PerceptionContractError(f"{label} is not a safe identifier")
|
|
return value
|
|
|
|
|
|
def _optional_identifier(value: str | None, label: str) -> None:
|
|
if value is not None:
|
|
_identifier(value, label)
|
|
|
|
|
|
def _unique_identifiers(
|
|
values: tuple[str, ...],
|
|
label: str,
|
|
*,
|
|
allow_empty: bool = False,
|
|
) -> None:
|
|
if (not values and not allow_empty) or len(set(values)) != len(values):
|
|
raise PerceptionContractError(f"{label} must be nonempty and unique")
|
|
for value in values:
|
|
_identifier(value, label)
|
|
|
|
|
|
def _unique_nonnegative_integers(values: tuple[int, ...], label: str) -> None:
|
|
if len(set(values)) != len(values):
|
|
raise PerceptionContractError(f"{label} must be unique")
|
|
for value in values:
|
|
_nonnegative_integer(value, label)
|
|
|
|
|
|
def _string_tuple(value: object, label: str) -> tuple[str, ...]:
|
|
if not isinstance(value, list):
|
|
raise PerceptionContractError(f"{label} must be an array")
|
|
return tuple(_string_value(item, label) for item in value)
|
|
|
|
|
|
def _integer_tuple(value: object, label: str) -> tuple[int, ...]:
|
|
if not isinstance(value, list):
|
|
raise PerceptionContractError(f"{label} must be an array")
|
|
result: list[int] = []
|
|
for item in value:
|
|
if not isinstance(item, int) or isinstance(item, bool):
|
|
raise PerceptionContractError(f"{label} must contain integers")
|
|
result.append(item)
|
|
return tuple(result)
|
|
|
|
|
|
def _vector3(values: tuple[float, float, float], label: str) -> tuple[float, float, float]:
|
|
if len(values) != 3:
|
|
raise PerceptionContractError(f"{label} must contain three values")
|
|
return tuple(_finite_number(value, label) for value in values) # type: ignore[return-value]
|
|
|
|
|
|
def _number_vector3(value: object, label: str) -> tuple[float, float, float]:
|
|
if not isinstance(value, list) or len(value) != 3:
|
|
raise PerceptionContractError(f"{label} must contain three values")
|
|
values = tuple(_finite_number(item, label) for item in value)
|
|
return (values[0], values[1], values[2])
|
|
|
|
|
|
def _enum[ENUM: StrEnum](
|
|
enum_type: type[ENUM],
|
|
value: object,
|
|
label: str,
|
|
) -> ENUM:
|
|
if not isinstance(value, str):
|
|
raise PerceptionContractError(f"{label} must be a string")
|
|
try:
|
|
return enum_type(value)
|
|
except ValueError as exc:
|
|
raise PerceptionContractError(f"{label} is incompatible") from exc
|