feat(perception): define semantic object understanding
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,454 @@
|
||||
"""Separate semantic-class evaluation over geometry-bound object projections.
|
||||
|
||||
This companion contour does not mutate the class-free M4.8 truth or score
|
||||
object presence. It evaluates canonical semantic resolution only after a
|
||||
separate two-reviewer, adjudicated semantic label set exists.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import math
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
from k1link.perception.object_understanding import (
|
||||
ObjectSemanticVocabulary,
|
||||
ObjectUnderstanding,
|
||||
SemanticResolution,
|
||||
validate_object_understanding,
|
||||
)
|
||||
|
||||
SEMANTIC_TRUTH_LABEL_SCHEMA: Final = "missioncore.semantic-object-truth-label/v0"
|
||||
SEMANTIC_QUALITY_PROFILE_SCHEMA: Final = "missioncore.semantic-object-quality-profile/v0"
|
||||
SEMANTIC_QUALITY_REPORT_SCHEMA: Final = "missioncore.semantic-object-quality-report/v0"
|
||||
|
||||
_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,159}$")
|
||||
_FALSE_AUTHORITY: Final = {
|
||||
"physical_live": False,
|
||||
"commands_enabled": False,
|
||||
"actuation_allowed": False,
|
||||
"navigation_or_safety_accepted": False,
|
||||
}
|
||||
|
||||
|
||||
class SemanticObjectQualityError(ValueError):
|
||||
"""A semantic truth, profile or evaluation input is incompatible."""
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SemanticTruthLabel:
|
||||
"""One independently reviewed canonical class bound to existing geometry."""
|
||||
|
||||
label_id: str
|
||||
observation_id: str
|
||||
source_id: str
|
||||
frame_id: str
|
||||
class_id: str
|
||||
reviewer_count: int
|
||||
adjudicated: bool
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
for value, label in (
|
||||
(self.label_id, "semantic truth label id"),
|
||||
(self.observation_id, "semantic truth observation id"),
|
||||
(self.source_id, "semantic truth source id"),
|
||||
(self.frame_id, "semantic truth frame id"),
|
||||
(self.class_id, "semantic truth class id"),
|
||||
):
|
||||
_identifier(value, label)
|
||||
if (
|
||||
not isinstance(self.reviewer_count, int)
|
||||
or isinstance(self.reviewer_count, bool)
|
||||
or self.reviewer_count < 2
|
||||
):
|
||||
raise SemanticObjectQualityError("semantic truth requires two independent reviewers")
|
||||
if self.adjudicated is not True:
|
||||
raise SemanticObjectQualityError("semantic truth must be adjudicated")
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {
|
||||
"schema_version": SEMANTIC_TRUTH_LABEL_SCHEMA,
|
||||
"label_id": self.label_id,
|
||||
"observation_id": self.observation_id,
|
||||
"source_id": self.source_id,
|
||||
"frame_id": self.frame_id,
|
||||
"class_id": self.class_id,
|
||||
"reviewer_count": self.reviewer_count,
|
||||
"adjudicated": self.adjudicated,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, value: object) -> SemanticTruthLabel:
|
||||
document = _object(value, "semantic truth label")
|
||||
_exact_keys(
|
||||
document,
|
||||
{
|
||||
"schema_version",
|
||||
"label_id",
|
||||
"observation_id",
|
||||
"source_id",
|
||||
"frame_id",
|
||||
"class_id",
|
||||
"reviewer_count",
|
||||
"adjudicated",
|
||||
},
|
||||
"semantic truth label",
|
||||
)
|
||||
if document.get("schema_version") != SEMANTIC_TRUTH_LABEL_SCHEMA:
|
||||
raise SemanticObjectQualityError("semantic truth label schema is incompatible")
|
||||
return cls(
|
||||
label_id=_string(document, "label_id"),
|
||||
observation_id=_string(document, "observation_id"),
|
||||
source_id=_string(document, "source_id"),
|
||||
frame_id=_string(document, "frame_id"),
|
||||
class_id=_string(document, "class_id"),
|
||||
reviewer_count=_integer(document, "reviewer_count"),
|
||||
adjudicated=_boolean(document, "adjudicated"),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SemanticObjectQualityProfile:
|
||||
"""Bounded evaluation policy for a frozen semantic provider candidate."""
|
||||
|
||||
profile_id: str
|
||||
vocabulary_id: str
|
||||
top_k: int
|
||||
minimum_prediction_coverage: float
|
||||
minimum_exact_top1_accuracy: float
|
||||
minimum_coarse_group_accuracy: float
|
||||
minimum_exact_top_k_recall: float
|
||||
maximum_unresolved_fraction: float
|
||||
maximum_conflict_fraction: float
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_identifier(self.profile_id, "semantic quality profile id")
|
||||
_identifier(self.vocabulary_id, "semantic quality vocabulary id")
|
||||
if (
|
||||
not isinstance(self.top_k, int)
|
||||
or isinstance(self.top_k, bool)
|
||||
or not 1 <= self.top_k <= 5
|
||||
):
|
||||
raise SemanticObjectQualityError("semantic quality top-k is invalid")
|
||||
for value, label in (
|
||||
(self.minimum_prediction_coverage, "minimum prediction coverage"),
|
||||
(self.minimum_exact_top1_accuracy, "minimum exact top-1 accuracy"),
|
||||
(self.minimum_coarse_group_accuracy, "minimum coarse group accuracy"),
|
||||
(self.minimum_exact_top_k_recall, "minimum exact top-k recall"),
|
||||
(self.maximum_unresolved_fraction, "maximum unresolved fraction"),
|
||||
(self.maximum_conflict_fraction, "maximum conflict fraction"),
|
||||
):
|
||||
_fraction(value, label)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SemanticQualityCase:
|
||||
"""One deterministic truth/prediction comparison row."""
|
||||
|
||||
label_id: str
|
||||
observation_id: str
|
||||
truth_class_id: str
|
||||
resolution: str
|
||||
selected_class_id: str | None
|
||||
exact_top1_correct: bool
|
||||
coarse_group_correct: bool
|
||||
exact_top_k_hit: bool
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {
|
||||
"label_id": self.label_id,
|
||||
"observation_id": self.observation_id,
|
||||
"truth_class_id": self.truth_class_id,
|
||||
"resolution": self.resolution,
|
||||
"selected_class_id": self.selected_class_id,
|
||||
"exact_top1_correct": self.exact_top1_correct,
|
||||
"coarse_group_correct": self.coarse_group_correct,
|
||||
"exact_top_k_hit": self.exact_top_k_hit,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SemanticObjectQualityResult:
|
||||
"""In-memory deterministic semantic report and comparison ledger."""
|
||||
|
||||
report: dict[str, object]
|
||||
cases: tuple[SemanticQualityCase, ...]
|
||||
|
||||
|
||||
def load_semantic_object_quality_profile(path: Path) -> SemanticObjectQualityProfile:
|
||||
"""Read a strict semantic evaluation profile."""
|
||||
|
||||
try:
|
||||
document = json.loads(path.expanduser().resolve(strict=True).read_text("utf-8"))
|
||||
except (OSError, json.JSONDecodeError) as exc:
|
||||
raise SemanticObjectQualityError("semantic quality profile cannot be read") from exc
|
||||
root = _object(document, "semantic quality profile")
|
||||
_exact_keys(
|
||||
root,
|
||||
{
|
||||
"schema_version",
|
||||
"profile_id",
|
||||
"vocabulary_id",
|
||||
"top_k",
|
||||
"thresholds",
|
||||
"scope",
|
||||
"authority",
|
||||
},
|
||||
"semantic quality profile",
|
||||
)
|
||||
if root.get("schema_version") != SEMANTIC_QUALITY_PROFILE_SCHEMA:
|
||||
raise SemanticObjectQualityError("semantic quality profile schema is incompatible")
|
||||
scope = _object(root.get("scope"), "semantic quality scope")
|
||||
_exact_keys(
|
||||
scope,
|
||||
{
|
||||
"object_presence_scored",
|
||||
"semantic_class_scored",
|
||||
"risk_policy_scored",
|
||||
"requires_separate_adjudicated_semantic_truth",
|
||||
},
|
||||
"semantic quality scope",
|
||||
)
|
||||
if scope != {
|
||||
"object_presence_scored": False,
|
||||
"semantic_class_scored": True,
|
||||
"risk_policy_scored": False,
|
||||
"requires_separate_adjudicated_semantic_truth": True,
|
||||
}:
|
||||
raise SemanticObjectQualityError("semantic quality scope changed")
|
||||
if _object(root.get("authority"), "semantic quality authority") != _FALSE_AUTHORITY:
|
||||
raise SemanticObjectQualityError("semantic quality authority changed")
|
||||
thresholds = _object(root.get("thresholds"), "semantic quality thresholds")
|
||||
_exact_keys(
|
||||
thresholds,
|
||||
{
|
||||
"minimum_prediction_coverage",
|
||||
"minimum_exact_top1_accuracy",
|
||||
"minimum_coarse_group_accuracy",
|
||||
"minimum_exact_top_k_recall",
|
||||
"maximum_unresolved_fraction",
|
||||
"maximum_conflict_fraction",
|
||||
},
|
||||
"semantic quality thresholds",
|
||||
)
|
||||
return SemanticObjectQualityProfile(
|
||||
profile_id=_string(root, "profile_id"),
|
||||
vocabulary_id=_string(root, "vocabulary_id"),
|
||||
top_k=_integer(root, "top_k"),
|
||||
minimum_prediction_coverage=_number(thresholds, "minimum_prediction_coverage"),
|
||||
minimum_exact_top1_accuracy=_number(thresholds, "minimum_exact_top1_accuracy"),
|
||||
minimum_coarse_group_accuracy=_number(thresholds, "minimum_coarse_group_accuracy"),
|
||||
minimum_exact_top_k_recall=_number(thresholds, "minimum_exact_top_k_recall"),
|
||||
maximum_unresolved_fraction=_number(thresholds, "maximum_unresolved_fraction"),
|
||||
maximum_conflict_fraction=_number(thresholds, "maximum_conflict_fraction"),
|
||||
)
|
||||
|
||||
|
||||
def score_semantic_object_quality(
|
||||
*,
|
||||
predictions: tuple[ObjectUnderstanding, ...],
|
||||
truth: tuple[SemanticTruthLabel, ...],
|
||||
vocabulary: ObjectSemanticVocabulary,
|
||||
profile: SemanticObjectQualityProfile,
|
||||
) -> SemanticObjectQualityResult:
|
||||
"""Score semantics on geometry-bound truth without scoring object presence."""
|
||||
|
||||
if not truth or any(not isinstance(item, SemanticTruthLabel) for item in truth):
|
||||
raise SemanticObjectQualityError("semantic truth set is invalid")
|
||||
if any(not isinstance(item, ObjectUnderstanding) for item in predictions):
|
||||
raise SemanticObjectQualityError("semantic prediction set is invalid")
|
||||
if profile.vocabulary_id != vocabulary.vocabulary_id:
|
||||
raise SemanticObjectQualityError("semantic quality vocabulary changed")
|
||||
truth_ids = tuple(item.observation_id for item in truth)
|
||||
if len(set(truth_ids)) != len(truth_ids):
|
||||
raise SemanticObjectQualityError("semantic truth observations must be unique")
|
||||
prediction_ids = tuple(item.observation.observation_id for item in predictions)
|
||||
if len(set(prediction_ids)) != len(prediction_ids):
|
||||
raise SemanticObjectQualityError("semantic prediction observations must be unique")
|
||||
declared_classes = {item.class_id for item in vocabulary.classes}
|
||||
if any(item.class_id not in declared_classes for item in truth):
|
||||
raise SemanticObjectQualityError("semantic truth uses an undeclared class")
|
||||
for candidate in predictions:
|
||||
validate_object_understanding(candidate, vocabulary)
|
||||
|
||||
by_observation = {item.observation.observation_id: item for item in predictions}
|
||||
cases: list[SemanticQualityCase] = []
|
||||
predicted_count = 0
|
||||
selected_count = 0
|
||||
conflict_count = 0
|
||||
for label in truth:
|
||||
prediction = by_observation.get(label.observation_id)
|
||||
if prediction is None:
|
||||
cases.append(
|
||||
SemanticQualityCase(
|
||||
label_id=label.label_id,
|
||||
observation_id=label.observation_id,
|
||||
truth_class_id=label.class_id,
|
||||
resolution="unavailable",
|
||||
selected_class_id=None,
|
||||
exact_top1_correct=False,
|
||||
coarse_group_correct=False,
|
||||
exact_top_k_hit=False,
|
||||
)
|
||||
)
|
||||
continue
|
||||
predicted_count += 1
|
||||
observation = prediction.observation
|
||||
if observation.source_id != label.source_id or observation.frame_id != label.frame_id:
|
||||
raise SemanticObjectQualityError("semantic prediction escaped its truth source frame")
|
||||
selected = prediction.semantic.selected_class_id
|
||||
if prediction.semantic.resolution is SemanticResolution.SELECTED:
|
||||
selected_count += 1
|
||||
if prediction.semantic.resolution is SemanticResolution.CONFLICT:
|
||||
conflict_count += 1
|
||||
exact = selected == label.class_id
|
||||
coarse = selected is not None and _coarse_group(selected, vocabulary) == _coarse_group(
|
||||
label.class_id, vocabulary
|
||||
)
|
||||
top_k_ids = {item.class_id for item in prediction.hypotheses[: profile.top_k]}
|
||||
cases.append(
|
||||
SemanticQualityCase(
|
||||
label_id=label.label_id,
|
||||
observation_id=label.observation_id,
|
||||
truth_class_id=label.class_id,
|
||||
resolution=prediction.semantic.resolution.value,
|
||||
selected_class_id=selected,
|
||||
exact_top1_correct=exact,
|
||||
coarse_group_correct=coarse,
|
||||
exact_top_k_hit=label.class_id in top_k_ids,
|
||||
)
|
||||
)
|
||||
|
||||
truth_count = len(truth)
|
||||
exact_accuracy = sum(item.exact_top1_correct for item in cases) / truth_count
|
||||
coarse_accuracy = sum(item.coarse_group_correct for item in cases) / truth_count
|
||||
top_k_recall = sum(item.exact_top_k_hit for item in cases) / truth_count
|
||||
prediction_coverage = predicted_count / truth_count
|
||||
unresolved_fraction = (truth_count - selected_count) / truth_count
|
||||
conflict_fraction = conflict_count / truth_count
|
||||
gates = {
|
||||
"prediction_coverage": (prediction_coverage >= profile.minimum_prediction_coverage),
|
||||
"exact_top1_accuracy": (exact_accuracy >= profile.minimum_exact_top1_accuracy),
|
||||
"coarse_group_accuracy": (coarse_accuracy >= profile.minimum_coarse_group_accuracy),
|
||||
"exact_top_k_recall": (top_k_recall >= profile.minimum_exact_top_k_recall),
|
||||
"unresolved_fraction": (unresolved_fraction <= profile.maximum_unresolved_fraction),
|
||||
"conflict_fraction": (conflict_fraction <= profile.maximum_conflict_fraction),
|
||||
}
|
||||
report: dict[str, object] = {
|
||||
"schema_version": SEMANTIC_QUALITY_REPORT_SCHEMA,
|
||||
"profile_id": profile.profile_id,
|
||||
"vocabulary_id": vocabulary.vocabulary_id,
|
||||
"scope": {
|
||||
"object_presence_scored": False,
|
||||
"semantic_class_scored": True,
|
||||
"risk_policy_scored": False,
|
||||
},
|
||||
"metrics": {
|
||||
"truth_count": truth_count,
|
||||
"prediction_count": predicted_count,
|
||||
"selected_count": selected_count,
|
||||
"conflict_count": conflict_count,
|
||||
"prediction_coverage": prediction_coverage,
|
||||
"exact_top1_accuracy": exact_accuracy,
|
||||
"coarse_group_accuracy": coarse_accuracy,
|
||||
"exact_top_k_recall": top_k_recall,
|
||||
"unresolved_fraction": unresolved_fraction,
|
||||
"conflict_fraction": conflict_fraction,
|
||||
},
|
||||
"gates": gates,
|
||||
"candidate_semantic_gate_passed": all(gates.values()),
|
||||
"limitations": [
|
||||
"Object presence and geometry quality remain owned by the class-free contour.",
|
||||
"Semantic qualification does not qualify risk policy or physical motion.",
|
||||
"No navigation, safety, command or actuation authority is granted.",
|
||||
],
|
||||
"authority": dict(_FALSE_AUTHORITY),
|
||||
}
|
||||
return SemanticObjectQualityResult(report=report, cases=tuple(cases))
|
||||
|
||||
|
||||
def _coarse_group(
|
||||
class_id: str,
|
||||
vocabulary: ObjectSemanticVocabulary,
|
||||
) -> str:
|
||||
ancestors = vocabulary.ancestors(class_id)
|
||||
if not ancestors:
|
||||
return class_id
|
||||
non_root = tuple(item for item in ancestors if item != "object.unknown")
|
||||
return non_root[-1] if non_root else class_id
|
||||
|
||||
|
||||
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 SemanticObjectQualityError(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 SemanticObjectQualityError(f"{label} fields are incompatible")
|
||||
|
||||
|
||||
def _string(document: dict[str, object], key: str) -> str:
|
||||
value = document.get(key)
|
||||
if not isinstance(value, str) or not value:
|
||||
raise SemanticObjectQualityError(f"{key} must be a nonempty string")
|
||||
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 SemanticObjectQualityError(f"{key} must be an integer")
|
||||
return value
|
||||
|
||||
|
||||
def _boolean(document: dict[str, object], key: str) -> bool:
|
||||
value = document.get(key)
|
||||
if not isinstance(value, bool):
|
||||
raise SemanticObjectQualityError(f"{key} must be boolean")
|
||||
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 SemanticObjectQualityError(f"{key} must be finite")
|
||||
return float(value)
|
||||
|
||||
|
||||
def _fraction(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 SemanticObjectQualityError(f"{label} must be within [0, 1]")
|
||||
return float(value)
|
||||
|
||||
|
||||
def _identifier(value: str, label: str) -> None:
|
||||
if not isinstance(value, str) or _IDENTIFIER.fullmatch(value) is None:
|
||||
raise SemanticObjectQualityError(f"{label} is not a safe identifier")
|
||||
|
||||
|
||||
__all__ = [
|
||||
"SEMANTIC_QUALITY_PROFILE_SCHEMA",
|
||||
"SEMANTIC_QUALITY_REPORT_SCHEMA",
|
||||
"SEMANTIC_TRUTH_LABEL_SCHEMA",
|
||||
"SemanticObjectQualityError",
|
||||
"SemanticObjectQualityProfile",
|
||||
"SemanticObjectQualityResult",
|
||||
"SemanticQualityCase",
|
||||
"SemanticTruthLabel",
|
||||
"load_semantic_object_quality_profile",
|
||||
"score_semantic_object_quality",
|
||||
]
|
||||
Reference in New Issue
Block a user