930 lines
34 KiB
Python
930 lines
34 KiB
Python
"""Versioned semantic, state and advisory-risk projection for an obstacle.
|
|
|
|
The projection composes an immutable :class:`ObstacleObservation` instead of
|
|
changing the strict v1 geometry contract. Semantic identity, observed state,
|
|
class priors and advisory risk remain separate claims with explicit evidence.
|
|
None of them can create occupancy or acquire navigation, safety or actuation
|
|
authority.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
import re
|
|
from dataclasses import dataclass
|
|
from enum import StrEnum
|
|
from pathlib import Path
|
|
from typing import Final
|
|
|
|
from .contracts import FalseAuthority, MotionState, ObstacleObservation
|
|
|
|
OBJECT_UNDERSTANDING_SCHEMA: Final = "missioncore.object-understanding/v1"
|
|
OBJECT_SEMANTIC_VOCABULARY_SCHEMA: Final = "missioncore.object-semantic-vocabulary/v0"
|
|
MAX_SEMANTIC_HYPOTHESES: Final = 5
|
|
|
|
_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,159}$")
|
|
_NORMALIZED_LABEL = re.compile(r"^[a-z0-9][a-z0-9_]{0,79}$")
|
|
|
|
|
|
class ObjectUnderstandingError(ValueError):
|
|
"""An object-understanding document or vocabulary is incompatible."""
|
|
|
|
|
|
class EvidenceKind(StrEnum):
|
|
DETECTOR = "detector"
|
|
SEMANTIC_MASK = "semantic-mask"
|
|
HUMAN_REVIEW = "human-review"
|
|
GEOMETRY = "geometry"
|
|
TEMPORAL = "temporal"
|
|
POLICY = "policy"
|
|
|
|
|
|
class SemanticResolution(StrEnum):
|
|
UNRESOLVED = "unresolved"
|
|
SELECTED = "selected"
|
|
AMBIGUOUS = "ambiguous"
|
|
CONFLICT = "conflict"
|
|
|
|
|
|
class AgencyState(StrEnum):
|
|
UNKNOWN = "unknown"
|
|
INERT = "inert"
|
|
ANIMATE = "animate"
|
|
SELF_PROPELLED = "self-propelled"
|
|
|
|
|
|
class StateBasis(StrEnum):
|
|
UNKNOWN = "unknown"
|
|
OBSERVED = "observed"
|
|
CLASS_PRIOR = "class-prior"
|
|
FUSED = "fused"
|
|
|
|
|
|
class RiskLevel(StrEnum):
|
|
UNKNOWN = "unknown"
|
|
LOW = "low"
|
|
ELEVATED = "elevated"
|
|
HIGH = "high"
|
|
CRITICAL = "critical"
|
|
|
|
|
|
class RiskBasis(StrEnum):
|
|
UNKNOWN = "unknown"
|
|
SEMANTIC_PRIOR = "semantic-prior"
|
|
OBSERVED_STATE = "observed-state"
|
|
GEOMETRY = "geometry"
|
|
FUSED = "fused"
|
|
|
|
|
|
class AdvisoryResponse(StrEnum):
|
|
MONITOR = "monitor"
|
|
REDUCE_SPEED = "reduce-speed"
|
|
YIELD = "yield"
|
|
STOP = "stop"
|
|
ROUTE_AROUND = "route-around"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class EvidenceProvenance:
|
|
"""One source-bound evidence item used by semantic, state or risk claims."""
|
|
|
|
evidence_id: str
|
|
kind: EvidenceKind
|
|
source_id: str
|
|
frame_id: str
|
|
provider_id: str
|
|
model_id: str | None
|
|
model_revision: str | None
|
|
preprocess_id: str | None
|
|
prompt_set_id: str | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
if not isinstance(self.kind, EvidenceKind):
|
|
raise ObjectUnderstandingError("evidence kind is invalid")
|
|
for value, label in (
|
|
(self.evidence_id, "evidence id"),
|
|
(self.source_id, "evidence source id"),
|
|
(self.frame_id, "evidence frame id"),
|
|
(self.provider_id, "evidence provider id"),
|
|
):
|
|
_identifier(value, label)
|
|
for optional_value, label in (
|
|
(self.model_id, "evidence model id"),
|
|
(self.model_revision, "evidence model revision"),
|
|
(self.preprocess_id, "evidence preprocess id"),
|
|
(self.prompt_set_id, "evidence prompt-set id"),
|
|
):
|
|
_optional_identifier(optional_value, label)
|
|
if self.kind in {EvidenceKind.DETECTOR, EvidenceKind.SEMANTIC_MASK} and (
|
|
self.model_id is None or self.model_revision is None or self.preprocess_id is None
|
|
):
|
|
raise ObjectUnderstandingError(
|
|
"model evidence requires model, revision and preprocess identity"
|
|
)
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"evidence_id": self.evidence_id,
|
|
"kind": self.kind.value,
|
|
"source_id": self.source_id,
|
|
"frame_id": self.frame_id,
|
|
"provider_id": self.provider_id,
|
|
"model_id": self.model_id,
|
|
"model_revision": self.model_revision,
|
|
"preprocess_id": self.preprocess_id,
|
|
"prompt_set_id": self.prompt_set_id,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> EvidenceProvenance:
|
|
document = _object(value, "evidence provenance")
|
|
_exact_keys(
|
|
document,
|
|
{
|
|
"evidence_id",
|
|
"kind",
|
|
"source_id",
|
|
"frame_id",
|
|
"provider_id",
|
|
"model_id",
|
|
"model_revision",
|
|
"preprocess_id",
|
|
"prompt_set_id",
|
|
},
|
|
"evidence provenance",
|
|
)
|
|
return cls(
|
|
evidence_id=_string(document, "evidence_id"),
|
|
kind=_enum(EvidenceKind, document.get("kind"), "evidence kind"),
|
|
source_id=_string(document, "source_id"),
|
|
frame_id=_string(document, "frame_id"),
|
|
provider_id=_string(document, "provider_id"),
|
|
model_id=_optional_string(document.get("model_id"), "model id"),
|
|
model_revision=_optional_string(document.get("model_revision"), "model revision"),
|
|
preprocess_id=_optional_string(document.get("preprocess_id"), "preprocess id"),
|
|
prompt_set_id=_optional_string(document.get("prompt_set_id"), "prompt-set id"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SemanticHypothesis:
|
|
"""One ranked canonical class hypothesis, never an occupancy identity."""
|
|
|
|
rank: int
|
|
class_id: str
|
|
raw_label: str
|
|
confidence: float
|
|
evidence_ids: tuple[str, ...]
|
|
|
|
def __post_init__(self) -> None:
|
|
_positive_integer(self.rank, "semantic rank")
|
|
_identifier(self.class_id, "semantic class id")
|
|
_label(self.raw_label, "raw semantic label")
|
|
_confidence(self.confidence, "semantic confidence")
|
|
_unique_identifiers(self.evidence_ids, "semantic evidence ids")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"rank": self.rank,
|
|
"class_id": self.class_id,
|
|
"raw_label": self.raw_label,
|
|
"confidence": self.confidence,
|
|
"evidence_ids": list(self.evidence_ids),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> SemanticHypothesis:
|
|
document = _object(value, "semantic hypothesis")
|
|
_exact_keys(
|
|
document,
|
|
{"rank", "class_id", "raw_label", "confidence", "evidence_ids"},
|
|
"semantic hypothesis",
|
|
)
|
|
return cls(
|
|
rank=_integer(document, "rank"),
|
|
class_id=_string(document, "class_id"),
|
|
raw_label=_string(document, "raw_label"),
|
|
confidence=_number(document, "confidence"),
|
|
evidence_ids=_string_tuple(document.get("evidence_ids"), "evidence ids"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SemanticDecision:
|
|
"""Resolution over ranked hypotheses; ambiguity remains first-class."""
|
|
|
|
resolution: SemanticResolution
|
|
selected_class_id: str | None
|
|
selected_confidence: float | None
|
|
reason_codes: tuple[str, ...]
|
|
|
|
def __post_init__(self) -> None:
|
|
if not isinstance(self.resolution, SemanticResolution):
|
|
raise ObjectUnderstandingError("semantic resolution is invalid")
|
|
_optional_identifier(self.selected_class_id, "selected semantic class id")
|
|
if self.selected_confidence is not None:
|
|
_confidence(self.selected_confidence, "selected semantic confidence")
|
|
_unique_identifiers(self.reason_codes, "semantic decision reasons")
|
|
has_selection = self.selected_class_id is not None and self.selected_confidence is not None
|
|
if self.resolution is SemanticResolution.SELECTED:
|
|
if not has_selection:
|
|
raise ObjectUnderstandingError("selected semantics require class and confidence")
|
|
elif self.selected_class_id is not None or self.selected_confidence is not None:
|
|
raise ObjectUnderstandingError("non-selected semantics cannot publish a selected class")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"resolution": self.resolution.value,
|
|
"selected_class_id": self.selected_class_id,
|
|
"selected_confidence": self.selected_confidence,
|
|
"reason_codes": list(self.reason_codes),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> SemanticDecision:
|
|
document = _object(value, "semantic decision")
|
|
_exact_keys(
|
|
document,
|
|
{
|
|
"resolution",
|
|
"selected_class_id",
|
|
"selected_confidence",
|
|
"reason_codes",
|
|
},
|
|
"semantic decision",
|
|
)
|
|
return cls(
|
|
resolution=_enum(
|
|
SemanticResolution,
|
|
document.get("resolution"),
|
|
"semantic resolution",
|
|
),
|
|
selected_class_id=_optional_string(
|
|
document.get("selected_class_id"), "selected class id"
|
|
),
|
|
selected_confidence=_optional_number(
|
|
document.get("selected_confidence"), "selected confidence"
|
|
),
|
|
reason_codes=_string_tuple(document.get("reason_codes"), "reason codes"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ObjectStateEstimate:
|
|
"""Observed motion and agency prior, with their bases kept explicit."""
|
|
|
|
motion: MotionState
|
|
motion_confidence: float
|
|
agency: AgencyState
|
|
agency_basis: StateBasis
|
|
evidence_ids: tuple[str, ...]
|
|
reason_codes: tuple[str, ...]
|
|
|
|
def __post_init__(self) -> None:
|
|
if not isinstance(self.motion, MotionState):
|
|
raise ObjectUnderstandingError("motion state is invalid")
|
|
if not isinstance(self.agency, AgencyState):
|
|
raise ObjectUnderstandingError("agency state is invalid")
|
|
if not isinstance(self.agency_basis, StateBasis):
|
|
raise ObjectUnderstandingError("agency basis is invalid")
|
|
_confidence(self.motion_confidence, "motion confidence")
|
|
_unique_identifiers(self.evidence_ids, "state evidence ids", allow_empty=True)
|
|
_unique_identifiers(self.reason_codes, "state reason codes")
|
|
if self.agency is AgencyState.UNKNOWN:
|
|
if self.agency_basis is not StateBasis.UNKNOWN:
|
|
raise ObjectUnderstandingError("unknown agency must retain unknown evidence basis")
|
|
elif self.agency_basis is StateBasis.UNKNOWN:
|
|
raise ObjectUnderstandingError("agency claim requires an explicit basis")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"motion": self.motion.value,
|
|
"motion_confidence": self.motion_confidence,
|
|
"agency": self.agency.value,
|
|
"agency_basis": self.agency_basis.value,
|
|
"evidence_ids": list(self.evidence_ids),
|
|
"reason_codes": list(self.reason_codes),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> ObjectStateEstimate:
|
|
document = _object(value, "object state")
|
|
_exact_keys(
|
|
document,
|
|
{
|
|
"motion",
|
|
"motion_confidence",
|
|
"agency",
|
|
"agency_basis",
|
|
"evidence_ids",
|
|
"reason_codes",
|
|
},
|
|
"object state",
|
|
)
|
|
return cls(
|
|
motion=_enum(MotionState, document.get("motion"), "motion state"),
|
|
motion_confidence=_number(document, "motion_confidence"),
|
|
agency=_enum(AgencyState, document.get("agency"), "agency state"),
|
|
agency_basis=_enum(StateBasis, document.get("agency_basis"), "agency basis"),
|
|
evidence_ids=_string_tuple(document.get("evidence_ids"), "evidence ids"),
|
|
reason_codes=_string_tuple(document.get("reason_codes"), "reason codes"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class AdvisoryRiskAssessment:
|
|
"""Evidence-qualified risk hint that is never a planner command."""
|
|
|
|
policy_id: str
|
|
level: RiskLevel
|
|
confidence: float
|
|
basis: RiskBasis
|
|
responses: tuple[AdvisoryResponse, ...]
|
|
evidence_ids: tuple[str, ...]
|
|
reason_codes: tuple[str, ...]
|
|
|
|
def __post_init__(self) -> None:
|
|
if not isinstance(self.level, RiskLevel):
|
|
raise ObjectUnderstandingError("risk level is invalid")
|
|
if not isinstance(self.basis, RiskBasis):
|
|
raise ObjectUnderstandingError("risk basis is invalid")
|
|
_identifier(self.policy_id, "risk policy id")
|
|
_confidence(self.confidence, "risk confidence")
|
|
_unique_enum_values(self.responses, "advisory responses", allow_empty=True)
|
|
_unique_identifiers(self.evidence_ids, "risk evidence ids", allow_empty=True)
|
|
_unique_identifiers(self.reason_codes, "risk reason codes")
|
|
if self.level is RiskLevel.UNKNOWN:
|
|
if self.basis is not RiskBasis.UNKNOWN or self.confidence != 0.0:
|
|
raise ObjectUnderstandingError(
|
|
"unknown risk must retain unknown basis and zero confidence"
|
|
)
|
|
elif self.basis is RiskBasis.UNKNOWN:
|
|
raise ObjectUnderstandingError("risk claim requires an explicit basis")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"policy_id": self.policy_id,
|
|
"level": self.level.value,
|
|
"confidence": self.confidence,
|
|
"basis": self.basis.value,
|
|
"responses": [item.value for item in self.responses],
|
|
"evidence_ids": list(self.evidence_ids),
|
|
"reason_codes": list(self.reason_codes),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> AdvisoryRiskAssessment:
|
|
document = _object(value, "advisory risk")
|
|
_exact_keys(
|
|
document,
|
|
{
|
|
"policy_id",
|
|
"level",
|
|
"confidence",
|
|
"basis",
|
|
"responses",
|
|
"evidence_ids",
|
|
"reason_codes",
|
|
},
|
|
"advisory risk",
|
|
)
|
|
return cls(
|
|
policy_id=_string(document, "policy_id"),
|
|
level=_enum(RiskLevel, document.get("level"), "risk level"),
|
|
confidence=_number(document, "confidence"),
|
|
basis=_enum(RiskBasis, document.get("basis"), "risk basis"),
|
|
responses=tuple(
|
|
_enum(AdvisoryResponse, item, "advisory response")
|
|
for item in _array(document, "responses")
|
|
),
|
|
evidence_ids=_string_tuple(document.get("evidence_ids"), "evidence ids"),
|
|
reason_codes=_string_tuple(document.get("reason_codes"), "reason codes"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ObjectUnderstanding:
|
|
"""Complete machine projection: geometry, semantics, state, risk and lineage."""
|
|
|
|
understanding_id: str
|
|
vocabulary_id: str
|
|
generated_monotonic_ns: int
|
|
observation: ObstacleObservation
|
|
hypotheses: tuple[SemanticHypothesis, ...]
|
|
semantic: SemanticDecision
|
|
state: ObjectStateEstimate
|
|
risk: AdvisoryRiskAssessment
|
|
provenance: tuple[EvidenceProvenance, ...]
|
|
authority: FalseAuthority = FalseAuthority()
|
|
|
|
def __post_init__(self) -> None:
|
|
_identifier(self.understanding_id, "understanding id")
|
|
_identifier(self.vocabulary_id, "semantic vocabulary id")
|
|
_nonnegative_integer(self.generated_monotonic_ns, "generation time")
|
|
if not isinstance(self.observation, ObstacleObservation):
|
|
raise ObjectUnderstandingError("object geometry observation is invalid")
|
|
if (
|
|
not isinstance(self.hypotheses, tuple)
|
|
or any(not isinstance(item, SemanticHypothesis) for item in self.hypotheses)
|
|
or len(self.hypotheses) > MAX_SEMANTIC_HYPOTHESES
|
|
):
|
|
raise ObjectUnderstandingError("semantic hypothesis set is invalid")
|
|
if tuple(item.rank for item in self.hypotheses) != tuple(
|
|
range(1, len(self.hypotheses) + 1)
|
|
):
|
|
raise ObjectUnderstandingError("semantic hypotheses must have contiguous ranks")
|
|
class_ids = tuple(item.class_id for item in self.hypotheses)
|
|
if len(set(class_ids)) != len(class_ids):
|
|
raise ObjectUnderstandingError("semantic hypothesis classes must be unique")
|
|
if any(
|
|
self.hypotheses[index].confidence < self.hypotheses[index + 1].confidence
|
|
for index in range(len(self.hypotheses) - 1)
|
|
):
|
|
raise ObjectUnderstandingError("semantic hypotheses must be ordered by confidence")
|
|
if not isinstance(self.semantic, SemanticDecision):
|
|
raise ObjectUnderstandingError("semantic decision is invalid")
|
|
if not isinstance(self.state, ObjectStateEstimate):
|
|
raise ObjectUnderstandingError("object state is invalid")
|
|
if not isinstance(self.risk, AdvisoryRiskAssessment):
|
|
raise ObjectUnderstandingError("advisory risk is invalid")
|
|
if not isinstance(self.authority, FalseAuthority):
|
|
raise ObjectUnderstandingError("object understanding authority is invalid")
|
|
if not isinstance(self.provenance, tuple) or any(
|
|
not isinstance(item, EvidenceProvenance) for item in self.provenance
|
|
):
|
|
raise ObjectUnderstandingError("evidence provenance is invalid")
|
|
evidence_ids = tuple(item.evidence_id for item in self.provenance)
|
|
if len(set(evidence_ids)) != len(evidence_ids):
|
|
raise ObjectUnderstandingError("evidence provenance ids must be unique")
|
|
if any(
|
|
item.source_id != self.observation.source_id
|
|
or item.frame_id != self.observation.frame_id
|
|
for item in self.provenance
|
|
):
|
|
raise ObjectUnderstandingError("object evidence escaped its geometry source frame")
|
|
known_evidence = set(evidence_ids)
|
|
claimed_evidence = (
|
|
{evidence_id for item in self.hypotheses for evidence_id in item.evidence_ids}
|
|
| set(self.state.evidence_ids)
|
|
| set(self.risk.evidence_ids)
|
|
)
|
|
if claimed_evidence - known_evidence:
|
|
raise ObjectUnderstandingError("object claim references unknown evidence")
|
|
if (
|
|
self.semantic.resolution
|
|
in {
|
|
SemanticResolution.AMBIGUOUS,
|
|
SemanticResolution.CONFLICT,
|
|
}
|
|
and not self.hypotheses
|
|
):
|
|
raise ObjectUnderstandingError("ambiguous or conflicting semantics require hypotheses")
|
|
if self.semantic.resolution is SemanticResolution.CONFLICT and len(self.hypotheses) < 2:
|
|
raise ObjectUnderstandingError("semantic conflict requires two hypotheses")
|
|
if self.semantic.resolution is SemanticResolution.SELECTED:
|
|
selected = next(
|
|
(
|
|
item
|
|
for item in self.hypotheses
|
|
if item.class_id == self.semantic.selected_class_id
|
|
),
|
|
None,
|
|
)
|
|
if selected is None or selected.confidence != self.semantic.selected_confidence:
|
|
raise ObjectUnderstandingError(
|
|
"selected semantics must match one ranked hypothesis"
|
|
)
|
|
|
|
@property
|
|
def occupancy_identity(self) -> str:
|
|
"""Semantic or risk changes never replace the geometry-owned identity."""
|
|
|
|
return self.observation.occupancy_identity
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"schema_version": OBJECT_UNDERSTANDING_SCHEMA,
|
|
"understanding_id": self.understanding_id,
|
|
"vocabulary_id": self.vocabulary_id,
|
|
"generated_monotonic_ns": self.generated_monotonic_ns,
|
|
"observation": self.observation.to_dict(),
|
|
"hypotheses": [item.to_dict() for item in self.hypotheses],
|
|
"semantic": self.semantic.to_dict(),
|
|
"state": self.state.to_dict(),
|
|
"risk": self.risk.to_dict(),
|
|
"provenance": [item.to_dict() for item in self.provenance],
|
|
"authority": self.authority.to_dict(),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> ObjectUnderstanding:
|
|
document = _contract(
|
|
value,
|
|
OBJECT_UNDERSTANDING_SCHEMA,
|
|
{
|
|
"understanding_id",
|
|
"vocabulary_id",
|
|
"generated_monotonic_ns",
|
|
"observation",
|
|
"hypotheses",
|
|
"semantic",
|
|
"state",
|
|
"risk",
|
|
"provenance",
|
|
"authority",
|
|
},
|
|
"object understanding",
|
|
)
|
|
return cls(
|
|
understanding_id=_string(document, "understanding_id"),
|
|
vocabulary_id=_string(document, "vocabulary_id"),
|
|
generated_monotonic_ns=_integer(document, "generated_monotonic_ns"),
|
|
observation=ObstacleObservation.from_dict(document.get("observation")),
|
|
hypotheses=tuple(
|
|
SemanticHypothesis.from_dict(item) for item in _array(document, "hypotheses")
|
|
),
|
|
semantic=SemanticDecision.from_dict(document.get("semantic")),
|
|
state=ObjectStateEstimate.from_dict(document.get("state")),
|
|
risk=AdvisoryRiskAssessment.from_dict(document.get("risk")),
|
|
provenance=tuple(
|
|
EvidenceProvenance.from_dict(item) for item in _array(document, "provenance")
|
|
),
|
|
authority=FalseAuthority.from_dict(document.get("authority")),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CanonicalObjectClass:
|
|
"""One class in the bounded experimental object vocabulary."""
|
|
|
|
class_id: str
|
|
parent_id: str | None
|
|
aliases: tuple[str, ...]
|
|
agency_prior: AgencyState
|
|
risk_traits: tuple[str, ...]
|
|
|
|
def __post_init__(self) -> None:
|
|
if not isinstance(self.agency_prior, AgencyState):
|
|
raise ObjectUnderstandingError("canonical agency prior is invalid")
|
|
_identifier(self.class_id, "canonical class id")
|
|
_optional_identifier(self.parent_id, "canonical parent id")
|
|
if not self.aliases:
|
|
raise ObjectUnderstandingError("canonical class aliases must be nonempty")
|
|
normalized = tuple(normalize_raw_label(item) for item in self.aliases)
|
|
if len(set(normalized)) != len(normalized):
|
|
raise ObjectUnderstandingError("canonical class aliases must be unique")
|
|
_unique_identifiers(self.risk_traits, "class risk traits", allow_empty=True)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ObjectSemanticVocabulary:
|
|
"""Loaded executable vocabulary profile; not a runtime ontology service."""
|
|
|
|
vocabulary_id: str
|
|
status: str
|
|
scope: str
|
|
classes: tuple[CanonicalObjectClass, ...]
|
|
max_hypotheses: int
|
|
|
|
def __post_init__(self) -> None:
|
|
_identifier(self.vocabulary_id, "vocabulary id")
|
|
if self.status != "experimental":
|
|
raise ObjectUnderstandingError("object vocabulary must remain experimental")
|
|
_identifier(self.scope, "vocabulary scope")
|
|
if not 1 <= self.max_hypotheses <= MAX_SEMANTIC_HYPOTHESES:
|
|
raise ObjectUnderstandingError("vocabulary top-k bound is invalid")
|
|
if not self.classes:
|
|
raise ObjectUnderstandingError("object vocabulary must declare classes")
|
|
by_id = {item.class_id: item for item in self.classes}
|
|
if len(by_id) != len(self.classes):
|
|
raise ObjectUnderstandingError("canonical class ids must be unique")
|
|
for item in self.classes:
|
|
if item.parent_id is not None and item.parent_id not in by_id:
|
|
raise ObjectUnderstandingError("canonical class parent is undeclared")
|
|
seen = {item.class_id}
|
|
parent_id = item.parent_id
|
|
while parent_id is not None:
|
|
if parent_id in seen:
|
|
raise ObjectUnderstandingError("canonical class hierarchy is cyclic")
|
|
seen.add(parent_id)
|
|
parent_id = by_id[parent_id].parent_id
|
|
aliases = [normalize_raw_label(alias) for item in self.classes for alias in item.aliases]
|
|
if len(set(aliases)) != len(aliases):
|
|
raise ObjectUnderstandingError("canonical aliases must be globally unique")
|
|
|
|
def class_definition(self, class_id: str) -> CanonicalObjectClass:
|
|
for item in self.classes:
|
|
if item.class_id == class_id:
|
|
return item
|
|
raise ObjectUnderstandingError("canonical class is undeclared")
|
|
|
|
def resolve_label(self, raw_label: str) -> str | None:
|
|
normalized = normalize_raw_label(raw_label)
|
|
for item in self.classes:
|
|
if normalized in {normalize_raw_label(alias) for alias in item.aliases}:
|
|
return item.class_id
|
|
return None
|
|
|
|
def ancestors(self, class_id: str) -> tuple[str, ...]:
|
|
by_id = {item.class_id: item for item in self.classes}
|
|
current = self.class_definition(class_id)
|
|
result: list[str] = []
|
|
while current.parent_id is not None:
|
|
result.append(current.parent_id)
|
|
current = by_id[current.parent_id]
|
|
return tuple(result)
|
|
|
|
|
|
def load_object_semantic_vocabulary(path: Path) -> ObjectSemanticVocabulary:
|
|
"""Load and fail-close an executable vocabulary profile."""
|
|
|
|
try:
|
|
document = json.loads(path.expanduser().resolve(strict=True).read_text("utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
raise ObjectUnderstandingError("object vocabulary cannot be read") from exc
|
|
root = _object(document, "object semantic vocabulary")
|
|
_exact_keys(
|
|
root,
|
|
{
|
|
"schema_version",
|
|
"vocabulary_id",
|
|
"status",
|
|
"scope",
|
|
"classes",
|
|
"policies",
|
|
},
|
|
"object semantic vocabulary",
|
|
)
|
|
if root.get("schema_version") != OBJECT_SEMANTIC_VOCABULARY_SCHEMA:
|
|
raise ObjectUnderstandingError("object vocabulary schema is incompatible")
|
|
policies = _object(root.get("policies"), "object vocabulary policies")
|
|
_exact_keys(
|
|
policies,
|
|
{
|
|
"occupancy_independent_of_semantics",
|
|
"unknown_preserves_obstacle",
|
|
"class_prior_is_not_observed_state",
|
|
"risk_is_advisory_only",
|
|
"planner_command_authority",
|
|
"max_hypotheses",
|
|
},
|
|
"object vocabulary policies",
|
|
)
|
|
required_true = (
|
|
"occupancy_independent_of_semantics",
|
|
"unknown_preserves_obstacle",
|
|
"class_prior_is_not_observed_state",
|
|
"risk_is_advisory_only",
|
|
)
|
|
if (
|
|
any(policies.get(key) is not True for key in required_true)
|
|
or policies.get("planner_command_authority") is not False
|
|
):
|
|
raise ObjectUnderstandingError("object vocabulary authority policy changed")
|
|
classes: list[CanonicalObjectClass] = []
|
|
for raw in _array(root, "classes"):
|
|
row = _object(raw, "canonical object class")
|
|
_exact_keys(
|
|
row,
|
|
{"class_id", "parent_id", "aliases", "agency_prior", "risk_traits"},
|
|
"canonical object class",
|
|
)
|
|
classes.append(
|
|
CanonicalObjectClass(
|
|
class_id=_string(row, "class_id"),
|
|
parent_id=_optional_string(row.get("parent_id"), "parent id"),
|
|
aliases=_string_tuple(row.get("aliases"), "aliases"),
|
|
agency_prior=_enum(AgencyState, row.get("agency_prior"), "agency prior"),
|
|
risk_traits=_string_tuple(row.get("risk_traits"), "risk traits"),
|
|
)
|
|
)
|
|
return ObjectSemanticVocabulary(
|
|
vocabulary_id=_string(root, "vocabulary_id"),
|
|
status=_string(root, "status"),
|
|
scope=_string(root, "scope"),
|
|
classes=tuple(classes),
|
|
max_hypotheses=_integer(policies, "max_hypotheses"),
|
|
)
|
|
|
|
|
|
def validate_object_understanding(
|
|
value: ObjectUnderstanding,
|
|
vocabulary: ObjectSemanticVocabulary,
|
|
) -> None:
|
|
"""Validate canonical class references without changing the document."""
|
|
|
|
if not isinstance(value, ObjectUnderstanding):
|
|
raise ObjectUnderstandingError("object understanding is invalid")
|
|
if value.vocabulary_id != vocabulary.vocabulary_id:
|
|
raise ObjectUnderstandingError("object understanding vocabulary changed")
|
|
declared = {item.class_id for item in vocabulary.classes}
|
|
referenced = {item.class_id for item in value.hypotheses}
|
|
if value.semantic.selected_class_id is not None:
|
|
referenced.add(value.semantic.selected_class_id)
|
|
if referenced - declared:
|
|
raise ObjectUnderstandingError("object understanding uses undeclared classes")
|
|
if len(value.hypotheses) > vocabulary.max_hypotheses:
|
|
raise ObjectUnderstandingError("object understanding exceeds vocabulary top-k")
|
|
|
|
|
|
def normalize_raw_label(value: str) -> str:
|
|
"""Normalize a provider label only for alias lookup, never as class truth."""
|
|
|
|
_label(value, "raw semantic label")
|
|
normalized = re.sub(r"[_\s-]+", "_", value.strip().lower())
|
|
if _NORMALIZED_LABEL.fullmatch(normalized) is None:
|
|
raise ObjectUnderstandingError("raw semantic label cannot be normalized")
|
|
return normalized
|
|
|
|
|
|
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 ObjectUnderstandingError(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 ObjectUnderstandingError(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 ObjectUnderstandingError(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 ObjectUnderstandingError(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 ObjectUnderstandingError(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 _integer(document: dict[str, object], key: str) -> int:
|
|
value = document.get(key)
|
|
if not isinstance(value, int) or isinstance(value, bool):
|
|
raise ObjectUnderstandingError(f"{key} must be an integer")
|
|
return value
|
|
|
|
|
|
def _number(document: dict[str, object], key: str) -> float:
|
|
value = document.get(key)
|
|
if (
|
|
not isinstance(value, (int, float))
|
|
or isinstance(value, bool)
|
|
or not math.isfinite(float(value))
|
|
):
|
|
raise ObjectUnderstandingError(f"{key} must be finite")
|
|
return float(value)
|
|
|
|
|
|
def _optional_number(value: object, label: str) -> float | None:
|
|
if value is None:
|
|
return None
|
|
if (
|
|
not isinstance(value, (int, float))
|
|
or isinstance(value, bool)
|
|
or not math.isfinite(float(value))
|
|
):
|
|
raise ObjectUnderstandingError(f"{label} must be finite")
|
|
return float(value)
|
|
|
|
|
|
def _identifier(value: str, label: str) -> None:
|
|
if not isinstance(value, str) or _IDENTIFIER.fullmatch(value) is None:
|
|
raise ObjectUnderstandingError(f"{label} is not a safe identifier")
|
|
|
|
|
|
def _optional_identifier(value: str | None, label: str) -> None:
|
|
if value is not None:
|
|
_identifier(value, label)
|
|
|
|
|
|
def _label(value: str, label: str) -> None:
|
|
if (
|
|
not isinstance(value, str)
|
|
or not value
|
|
or value != value.strip()
|
|
or len(value) > 120
|
|
or any(ord(character) < 32 for character in value)
|
|
):
|
|
raise ObjectUnderstandingError(f"{label} is invalid")
|
|
|
|
|
|
def _nonnegative_integer(value: object, label: str) -> int:
|
|
if not isinstance(value, int) or isinstance(value, bool) or value < 0:
|
|
raise ObjectUnderstandingError(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 ObjectUnderstandingError(f"{label} must be positive")
|
|
return result
|
|
|
|
|
|
def _confidence(value: object, label: str) -> float:
|
|
if (
|
|
not isinstance(value, (int, float))
|
|
or isinstance(value, bool)
|
|
or not math.isfinite(float(value))
|
|
or not 0.0 <= float(value) <= 1.0
|
|
):
|
|
raise ObjectUnderstandingError(f"{label} must be within [0, 1]")
|
|
return float(value)
|
|
|
|
|
|
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 ObjectUnderstandingError(f"{label} must be unique")
|
|
for value in values:
|
|
_identifier(value, label)
|
|
|
|
|
|
def _unique_enum_values(
|
|
values: tuple[AdvisoryResponse, ...],
|
|
label: str,
|
|
*,
|
|
allow_empty: bool,
|
|
) -> None:
|
|
if (not values and not allow_empty) or len(set(values)) != len(values):
|
|
raise ObjectUnderstandingError(f"{label} must be unique")
|
|
if any(not isinstance(value, AdvisoryResponse) for value in values):
|
|
raise ObjectUnderstandingError(f"{label} are invalid")
|
|
|
|
|
|
def _string_tuple(value: object, label: str) -> tuple[str, ...]:
|
|
if not isinstance(value, list):
|
|
raise ObjectUnderstandingError(f"{label} must be an array")
|
|
return tuple(_string_value(item, label) for item in value)
|
|
|
|
|
|
def _enum[ENUM: StrEnum](
|
|
enum_type: type[ENUM],
|
|
value: object,
|
|
label: str,
|
|
) -> ENUM:
|
|
if not isinstance(value, str):
|
|
raise ObjectUnderstandingError(f"{label} must be a string")
|
|
try:
|
|
return enum_type(value)
|
|
except ValueError as exc:
|
|
raise ObjectUnderstandingError(f"{label} is incompatible") from exc
|
|
|
|
|
|
__all__ = [
|
|
"MAX_SEMANTIC_HYPOTHESES",
|
|
"OBJECT_SEMANTIC_VOCABULARY_SCHEMA",
|
|
"OBJECT_UNDERSTANDING_SCHEMA",
|
|
"AdvisoryResponse",
|
|
"AdvisoryRiskAssessment",
|
|
"AgencyState",
|
|
"CanonicalObjectClass",
|
|
"EvidenceKind",
|
|
"EvidenceProvenance",
|
|
"ObjectSemanticVocabulary",
|
|
"ObjectStateEstimate",
|
|
"ObjectUnderstanding",
|
|
"ObjectUnderstandingError",
|
|
"RiskBasis",
|
|
"RiskLevel",
|
|
"SemanticDecision",
|
|
"SemanticHypothesis",
|
|
"SemanticResolution",
|
|
"StateBasis",
|
|
"load_object_semantic_vocabulary",
|
|
"normalize_raw_label",
|
|
"validate_object_understanding",
|
|
]
|