# ----------------------------------------------------------------------------
# Copyright (c) 2021-2026 DexForce Technology Co., Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ----------------------------------------------------------------------------
"""Immutable task-registration catalog for declarative Task Programs."""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass, field, fields, is_dataclass
from enum import Enum
import hashlib
import json
import math
from _thread import LockType
from threading import Lock
from types import MappingProxyType
import torch
from embodichain.lab.gym.envs.settling import DynamicSettleMonitorCfg
from embodichain.lab.sim.atomic_actions import (
Affordance,
AtomicActionEngine,
EndpointTrackingFeedbackAddress,
GRASP_CAPABILITY,
JOINT_POSITION_CHANNEL,
SceneProvider,
SkillDescriptor,
)
from embodichain.lab.sim.atomic_actions.primitives import BUILTIN_ACTION_TYPES
from embodichain.lab.sim.atomic_actions.tracking import (
FeedbackTerminalAcceptance,
TrackingRuntime,
)
from embodichain.lab.task_program.semantics import (
CONTACT_EFFECT_CHANNEL,
CONSTRAINT_EFFECT_CHANNEL,
ControlPartEndpoint,
ControlPartEvidenceAddress,
FORCE_EFFECT_CHANNEL,
JOINT_STATE_EFFECT_CHANNEL,
PLACE_IN_AFFORDANCE_CAPABILITY,
PLACE_ON_AFFORDANCE_CAPABILITY,
POSE_RELATION_EFFECT_CHANNEL,
BoundRobotSkillProfile,
EffectEvidenceProvider,
Place,
RobotSkillProfile,
RegisteredSemanticCall,
ResourceEndpoint,
ResourceEndpointAdapter,
SceneAffordanceRef,
SceneArticulationRef,
SceneEntityRef,
SceneManifest,
SceneObjectRef,
SceneRegistry,
SemanticCallCatalog,
SemanticIntegrationManifest,
builtin_semantic_call_catalog,
)
from embodichain.lab.task_program.compiler.lowering import (
ContainerRelationTargetGrounder,
HandOverPoseProvider,
RegisteredSemanticLowerer,
RelationTargetGrounder,
SupportSurfaceRelationTargetGrounder,
)
from embodichain.lab.task_program.semantics.effects import (
COMPOSITE_EFFECT_MONITOR_ID,
COMPOSITE_EFFECT_MONITOR_REVISION,
CompositeEffectMonitorFactory,
EffectMonitorRegistry,
)
from embodichain.lab.task_program.runtime.parallel_executor import (
ParallelCommandSafetyValidator,
)
from embodichain.lab.task_program.language.schema import (
TaskProgramCfg,
TaskProgramIntegrationCfg,
PostPolicyCfg,
RegisteredSemanticCallCfg,
SemanticCallCfg,
ValidatorCfg,
)
from embodichain.lab.task_program.compiler import (
CompiledTaskProgram,
TaskProgramCompiler,
)
from embodichain.lab.task_program.language.decoder import (
ConfigPath,
TaskProgramValidationError,
SceneReferenceRole,
)
from embodichain.lab.gym.envs.task_program.bridge import RuntimeTransportActionEncoder
from .extensions import (
ControlPartEvidenceProviderDeclaration,
ControlPartEvidenceProviderFactory,
EndpointAdapterDeclaration,
ParallelCommandSafetyValidatorFactory,
ParallelSafetyDeclaration,
RegisteredSemanticLowererDeclaration,
RegisteredSemanticLowererFactory,
RuntimeTransportDeclaration,
StandardExtensionDeclarations,
build_standard_extension_declarations,
validate_immutable_extension_declaration,
)
from .simulation import SimulationRobotSkillProfileBinding, SimulationSceneBinding
from .simulation.policies import default_simulation_settle_presets
_CATALOG_FINGERPRINT_SCHEMA_VERSION = 4
_POST_POLICY_KINDS = frozenset({"wait_stable"})
_VALIDATOR_KINDS = frozenset({"articulation_joint_position", "object_near_target"})
[docs]
class IntegrationFingerprintMismatch(RuntimeError):
"""Raised when a live integration no longer matches its registration."""
def _qualified_name(value: type[object] | object) -> str:
"""Return a stable fully-qualified type name."""
value_type = value if isinstance(value, type) else type(value)
return f"{value_type.__module__}.{value_type.__qualname__}"
def _canonical_value(value: object) -> object:
"""Convert provider-free declarations to deterministic JSON values."""
if value is None or type(value) in (bool, int, str):
return value
if type(value) is float:
if not math.isfinite(value):
raise ValueError("Fingerprint metadata cannot contain non-finite floats.")
return value
if isinstance(value, Enum):
return {
"type": _qualified_name(value),
"value": _canonical_value(value.value),
}
if isinstance(value, type):
return {"type": _qualified_name(value)}
if isinstance(value, torch.Tensor):
tensor = value.detach().cpu()
return {
"tensor_dtype": str(tensor.dtype),
"tensor_shape": list(tensor.shape),
"tensor_value": tensor.tolist(),
}
if isinstance(value, Mapping):
normalized: dict[str, object] = {}
for key, nested in value.items():
if type(key) is not str:
raise TypeError("Fingerprint mapping keys must be exact strings.")
normalized[key] = _canonical_value(nested)
return {key: normalized[key] for key in sorted(normalized)}
if isinstance(value, (tuple, list)):
return [_canonical_value(nested) for nested in value]
if isinstance(value, (set, frozenset)):
normalized = [_canonical_value(nested) for nested in value]
return sorted(
normalized,
key=lambda item: json.dumps(
item,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
),
)
if is_dataclass(value):
metadata = {
data_field.name: _canonical_value(getattr(value, data_field.name))
for data_field in fields(value)
}
return {"type": _qualified_name(value), "fields": metadata}
raise TypeError(
"Registration fingerprint metadata contains unsupported value type "
f"{_qualified_name(value)!r}. Values must be complete declarative data; "
"live or opaque objects cannot be fingerprinted by type alone."
)
def _provider_fingerprint_declaration(provider: object) -> object:
"""Return the complete canonical declaration for one validated provider."""
if is_dataclass(provider):
return provider
return {"provider_type": _qualified_name(provider)}
def _canonical_json(value: object) -> str:
"""Encode one declaration using the versioned canonical JSON form."""
return json.dumps(
_canonical_value(value),
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
allow_nan=False,
)
def _digest(payload: object) -> str:
"""Return the SHA-256 digest for one canonical declaration payload."""
return hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest()
def _snapshot_settle_presets(
values: Mapping[str, DynamicSettleMonitorCfg],
) -> Mapping[str, DynamicSettleMonitorCfg]:
"""Own one strict named settle-preset table."""
if not isinstance(values, Mapping) or not values:
raise ValueError("settle_presets must be a non-empty mapping.")
normalized: dict[str, DynamicSettleMonitorCfg] = {}
for preset_id, preset in values.items():
if (
type(preset_id) is not str
or not preset_id
or preset_id != preset_id.strip()
):
raise ValueError(
"Settle preset IDs must be non-empty strings without outer "
"whitespace."
)
if not isinstance(preset, DynamicSettleMonitorCfg):
raise TypeError(
"settle_presets values must be DynamicSettleMonitorCfg values."
)
normalized[preset_id] = preset.snapshot()
return MappingProxyType(normalized)
def _exact_identifier(value: object, *, field_name: str) -> str:
"""Validate one exact catalog identifier."""
if type(value) is not str or not value or value != value.strip():
raise ValueError(
f"{field_name} must be a non-empty string without outer whitespace."
)
return value
def _relation_grounder_key(
grounder: RelationTargetGrounder,
) -> tuple[str, type[Affordance], str]:
"""Return the compiler-compatible exact key for one relation grounder."""
grounder_type = type(grounder)
capability = _exact_identifier(
getattr(grounder_type, "capability", None),
field_name="RelationTargetGrounder.capability",
)
affordance_type = getattr(grounder_type, "affordance_type", None)
if not isinstance(affordance_type, type) or not issubclass(
affordance_type,
Affordance,
):
raise TypeError(
"RelationTargetGrounder.affordance_type must be an Affordance subclass."
)
revision = _exact_identifier(
getattr(grounder_type, "affordance_revision", None),
field_name="RelationTargetGrounder.affordance_revision",
)
return capability, affordance_type, revision
def _relation_grounder_order_key(
grounder: RelationTargetGrounder,
) -> tuple[str, str, str]:
"""Return one totally ordered rendering of a relation-grounder key."""
capability, affordance_type, revision = _relation_grounder_key(grounder)
return capability, _qualified_name(affordance_type), revision
def _snapshot_relation_grounders(
values: tuple[RelationTargetGrounder, ...],
) -> tuple[RelationTargetGrounder, ...]:
"""Validate and own one immutable relation-grounder tuple."""
if type(values) is not tuple:
raise TypeError("relation_grounders must be an exact tuple.")
seen: set[tuple[str, type[Affordance], str]] = set()
for grounder in values:
if not isinstance(grounder, RelationTargetGrounder):
raise TypeError(
"relation_grounders must contain RelationTargetGrounder instances."
)
validate_immutable_extension_declaration(
grounder,
field_name="relation_grounders",
)
key = _relation_grounder_key(grounder)
if key in seen:
raise ValueError(f"Duplicate relation grounder key {key!r}.")
seen.add(key)
return tuple(values)
def _builtin_relation_grounders(
scene_binding: SimulationSceneBinding,
) -> tuple[RelationTargetGrounder, ...]:
"""Install standard grounders for declared production relation bindings."""
values: list[RelationTargetGrounder] = []
if scene_binding.support_surfaces:
values.append(SupportSurfaceRelationTargetGrounder())
if scene_binding.containers:
values.append(ContainerRelationTargetGrounder())
return tuple(values)
def _snapshot_relation_grounder_keys(
values: frozenset[tuple[str, type[Affordance], str]],
) -> frozenset[tuple[str, type[Affordance], str]]:
"""Validate immutable provider-free relation-grounder lookup keys."""
if type(values) is not frozenset:
raise TypeError("relation_grounder_keys must be an exact frozenset.")
normalized: set[tuple[str, type[Affordance], str]] = set()
for key in values:
if type(key) is not tuple or len(key) != 3:
raise TypeError("relation_grounder_keys must contain exact 3-tuple values.")
capability, affordance_type, revision = key
_exact_identifier(capability, field_name="relation grounder capability")
if not isinstance(affordance_type, type) or not issubclass(
affordance_type,
Affordance,
):
raise TypeError(
"relation grounder affordance types must be Affordance subclasses."
)
_exact_identifier(revision, field_name="relation grounder revision")
normalized.add((capability, affordance_type, revision))
return frozenset(normalized)
def _handover_pose_provider_id(provider: HandOverPoseProvider) -> str:
"""Return the compiler-compatible class ID for one hand-over provider."""
return _exact_identifier(
getattr(type(provider), "provider_id", None),
field_name="HandOverPoseProvider.provider_id",
)
def _snapshot_handover_pose_providers(
values: tuple[HandOverPoseProvider, ...],
) -> tuple[HandOverPoseProvider, ...]:
"""Validate and own one immutable hand-over-provider tuple."""
if type(values) is not tuple:
raise TypeError("handover_pose_providers must be an exact tuple.")
seen: set[str] = set()
for provider in values:
if not isinstance(provider, HandOverPoseProvider):
raise TypeError(
"handover_pose_providers must contain HandOverPoseProvider instances."
)
validate_immutable_extension_declaration(
provider,
field_name="handover_pose_providers",
)
provider_id = _handover_pose_provider_id(provider)
if provider_id in seen:
raise ValueError(f"Duplicate handover pose provider {provider_id!r}.")
seen.add(provider_id)
return tuple(values)
def _validate_standard_call_catalog(
call_catalog: SemanticCallCatalog,
lowerer_declarations: Mapping[str, RegisteredSemanticLowererDeclaration],
) -> None:
"""Require exact built-ins and factory coverage for every registered call."""
builtins = builtin_semantic_call_catalog().descriptors
registered_call_ids: set[str] = set()
for descriptor in call_catalog.descriptors.values():
if descriptor.spec_type is RegisteredSemanticCall:
registered_call_ids.add(descriptor.call_id)
continue
expected = builtins.get(descriptor.call_id)
if expected != descriptor:
raise ValueError(
f"Semantic call {descriptor.call_id!r} does not match its exact "
"curated descriptor."
)
declared_call_ids = set(lowerer_declarations)
if registered_call_ids != declared_call_ids:
raise ValueError(
"Registered semantic call descriptors and lowerer factories must "
"cover exactly the same call IDs; "
f"missing={sorted(registered_call_ids - declared_call_ids)}, "
f"unused={sorted(declared_call_ids - registered_call_ids)}."
)
def _validate_standard_effect_monitors(profile: RobotSkillProfile) -> None:
"""Require every preset to use the exact built-in effect-monitor factory."""
registry = EffectMonitorRegistry((CompositeEffectMonitorFactory(),))
builtin_key = (
COMPOSITE_EFFECT_MONITOR_ID,
COMPOSITE_EFFECT_MONITOR_REVISION,
)
for preset_id, preset in profile.presets.items():
for semantic_id, monitor_ref in preset.effect_monitors.items():
key = monitor_ref.monitor_id, monitor_ref.revision
if key != builtin_key:
raise ValueError(
f"Preset {preset_id!r} semantic call {semantic_id!r} selects "
f"non-built-in effect monitor {key!r}; the standard "
"simulation registration supports only {builtin_key!r}."
)
try:
registry.validate_ref(monitor_ref)
except (KeyError, TypeError, ValueError) as exc:
raise ValueError(
f"Preset {preset_id!r} semantic call {semantic_id!r} has an "
"invalid built-in effect-monitor declaration."
) from exc
def _validate_standard_tracking_metrics(profile: RobotSkillProfile) -> None:
"""Resolve every reachable metric through the exact built-in evaluator table."""
evaluators = TrackingRuntime.with_builtins().evaluators
for preset_id, preset in profile.presets.items():
policy = preset.tracking_policy
metric_groups = []
if policy.in_flight is not None:
metric_groups.append(("in_flight", policy.in_flight.metrics))
if isinstance(policy.terminal, FeedbackTerminalAcceptance):
metric_groups.append(("terminal", policy.terminal.metrics))
for phase, metrics in metric_groups:
for metric in metrics:
try:
evaluators.resolve(metric)
except (KeyError, TypeError, ValueError) as exc:
key = metric.metric_id, metric.revision, _qualified_name(metric)
raise ValueError(
f"Preset {preset_id!r} {phase} tracking metric {key!r} "
"has no exact built-in evaluator in the standard "
"simulation registration."
) from exc
[docs]
@dataclass(frozen=True, slots=True)
class TaskProgramIntegrationCatalog:
"""Provider-free integration directory owned by one task registration."""
scene_registry_id: str
robot_profile_id: str
scene: SceneManifest
robot_profile: RobotSkillProfile
call_catalog: SemanticCallCatalog
relation_grounder_keys: frozenset[tuple[str, type[Affordance], str]]
settle_preset_ids: frozenset[str]
endpoint_adapter_declarations: Mapping[
type[ResourceEndpoint], EndpointAdapterDeclaration
]
runtime_transport_declarations: tuple[RuntimeTransportDeclaration, ...]
parallel_safety_declaration: ParallelSafetyDeclaration | None
control_part_evidence_declaration: ControlPartEvidenceProviderDeclaration | None
registered_semantic_lowerer_declarations: Mapping[
str, RegisteredSemanticLowererDeclaration
]
fingerprint: str
_required_skills: Mapping[str, SkillDescriptor] = field(
repr=False,
compare=False,
)
def __post_init__(self) -> None:
for field_name in ("scene_registry_id", "robot_profile_id"):
value = getattr(self, field_name)
if type(value) is not str or not value or value != value.strip():
raise ValueError(f"{field_name} must be an exact identifier.")
if type(self.scene) is not SceneManifest:
raise TypeError("scene must be exactly SceneManifest.")
if type(self.robot_profile) is not RobotSkillProfile:
raise TypeError("robot_profile must be exactly RobotSkillProfile.")
if type(self.call_catalog) is not SemanticCallCatalog:
raise TypeError("call_catalog must be exactly SemanticCallCatalog.")
object.__setattr__(
self,
"relation_grounder_keys",
_snapshot_relation_grounder_keys(self.relation_grounder_keys),
)
extensions = StandardExtensionDeclarations(
endpoint_adapters=self.endpoint_adapter_declarations,
runtime_transports=self.runtime_transport_declarations,
parallel_safety=self.parallel_safety_declaration,
control_part_evidence=self.control_part_evidence_declaration,
registered_semantic_lowerers=(
self.registered_semantic_lowerer_declarations
),
)
profile_endpoint_types = frozenset(
type(endpoint)
for resource in self.robot_profile.resources.values()
for endpoint in resource.endpoints.values()
)
if profile_endpoint_types != frozenset(extensions.endpoint_adapters):
raise ValueError(
"endpoint_adapter_declarations must cover every exact robot "
"profile endpoint type and no others."
)
object.__setattr__(
self,
"endpoint_adapter_declarations",
extensions.endpoint_adapters,
)
object.__setattr__(
self,
"runtime_transport_declarations",
extensions.runtime_transports,
)
object.__setattr__(
self,
"parallel_safety_declaration",
extensions.parallel_safety,
)
object.__setattr__(
self,
"control_part_evidence_declaration",
extensions.control_part_evidence,
)
object.__setattr__(
self,
"registered_semantic_lowerer_declarations",
extensions.registered_semantic_lowerers,
)
if self.robot_profile.profile_id != self.robot_profile_id:
raise ValueError("robot_profile_id must match robot_profile.profile_id.")
preset_ids = frozenset(self.settle_preset_ids)
if not preset_ids:
raise ValueError("settle_preset_ids must not be empty.")
object.__setattr__(self, "settle_preset_ids", preset_ids)
if (
type(self.fingerprint) is not str
or len(self.fingerprint) != 64
or any(
character not in "0123456789abcdef" for character in self.fingerprint
)
):
raise ValueError("fingerprint must be a lowercase SHA-256 digest.")
object.__setattr__(
self,
"_required_skills",
MappingProxyType(dict(self._required_skills)),
)
[docs]
def validate_integration(
self,
integration: TaskProgramIntegrationCfg,
*,
path: ConfigPath,
) -> None:
"""Validate exact scene, profile, and runtime-preset selection."""
del path
if integration.scene_registry != self.scene_registry_id:
raise ValueError(
f"Expected scene_registry {self.scene_registry_id!r}, got "
f"{integration.scene_registry!r}."
)
if integration.robot_profile != self.robot_profile_id:
raise ValueError(
f"Expected robot_profile {self.robot_profile_id!r}, got "
f"{integration.robot_profile!r}."
)
if integration.runtime_preset not in self.robot_profile.presets:
raise KeyError(
f"Unknown runtime preset {integration.runtime_preset!r}; available "
f"presets are {sorted(self.robot_profile.presets)}."
)
[docs]
def validate_semantic_call(
self,
call: SemanticCallCfg,
*,
path: ConfigPath,
) -> None:
"""Validate that the semantic call is present in the integration catalog."""
call_id = call.call_id if type(call) is RegisteredSemanticCallCfg else call.kind
self.call_catalog.discover(call_id)
def _validate_place_relation_grounder(
self,
call: Place,
*,
affordance: SceneAffordanceRef,
path: ConfigPath,
) -> None:
"""Require the exact linked relation-affordance grounder pre-sim."""
if call.on is not None:
capability = PLACE_ON_AFFORDANCE_CAPABILITY
relation_field = "on"
elif call.inside is not None:
capability = PLACE_IN_AFFORDANCE_CAPABILITY
relation_field = "inside"
else:
return
entry = self.scene.lookup(
affordance,
expected_type=SceneAffordanceRef,
path=(*path, relation_field),
)
payload_type = entry.affordance_payload_type
revision = entry.affordance_revision
if payload_type is None or revision is None:
raise TaskProgramValidationError(
"incomplete_relation_affordance_declaration",
(*path, relation_field),
f"Relation affordance {affordance.entity_id!r} must declare an "
"exact payload type and revision.",
)
key = (capability, payload_type, revision)
if key not in self.relation_grounder_keys:
rendered_key = (
capability,
_qualified_name(payload_type),
revision,
)
raise TaskProgramValidationError(
"relation_grounder_not_registered",
(*path, relation_field),
f"No task-registration relation grounder matches linked "
f"affordance {affordance.entity_id!r} with key {rendered_key!r}.",
)
[docs]
def validate_scene_reference(
self,
reference: str,
*,
role: SceneReferenceRole,
path: ConfigPath,
) -> None:
"""Validate one typed scene reference against its declared role."""
expected: dict[str, tuple[type[SceneEntityRef], ...]] = {
"entity": (SceneEntityRef,),
"object": (SceneObjectRef,),
"articulation": (SceneArticulationRef,),
"affordance": (SceneAffordanceRef,),
"object_or_affordance": (SceneObjectRef, SceneAffordanceRef),
}
expected_types = expected.get(role)
if expected_types is None:
raise ValueError(f"Unsupported scene reference role {role!r}.")
resolved = self.scene.resolve(reference, path=path)
if not isinstance(resolved, expected_types):
raise TypeError(
f"Scene reference {reference!r} is {type(resolved).__name__}, "
f"not one of {tuple(value.__name__ for value in expected_types)}."
)
[docs]
def validate_post_policy(
self,
policy: PostPolicyCfg,
*,
path: ConfigPath,
) -> None:
"""Validate one registered post-policy kind and named preset."""
del path
if policy.kind not in _POST_POLICY_KINDS:
raise KeyError(
f"Unknown post-policy kind {policy.kind!r}; available kinds are "
f"{sorted(_POST_POLICY_KINDS)}."
)
if policy.preset not in self.settle_preset_ids:
raise KeyError(
f"Unknown settle preset {policy.preset!r}; available presets are "
f"{sorted(self.settle_preset_ids)}."
)
[docs]
def validate_validator(
self,
validator: ValidatorCfg,
*,
path: ConfigPath,
) -> None:
"""Validate one registered segment-validator kind."""
del path
if validator.kind not in _VALIDATOR_KINDS:
raise KeyError(
f"Unknown validator kind {validator.kind!r}; available kinds are "
f"{sorted(_VALIDATOR_KINDS)}."
)
[docs]
def preflight(self, program: TaskProgramCfg) -> CompiledTaskProgram:
"""Compile and statically link every expanded semantic call."""
self.validate_integration(program.integration, path=("integration",))
compiled = TaskProgramCompiler(self.scene).compile(program)
manifest = SemanticIntegrationManifest(
scene=self.scene,
robot_profile=self.robot_profile,
call_catalog=self.call_catalog,
runtime_preset=program.integration.runtime_preset,
)
for segment in compiled.iter_segments():
if (
segment.parallel_block is not None
and self.parallel_safety_declaration is None
):
raise TaskProgramValidationError(
"parallel_safety_factory_not_registered",
segment.parallel_block.source_path,
"Parallel execution requires a task-registration-owned "
"physical safety-validator factory.",
)
for call in segment.calls:
linked = manifest.link_call(call.call, path=call.source_path)
if type(linked.call) is Place and linked.call.at is None:
destination = linked.affordances.get("destination")
if destination is None:
raise AssertionError(
"Linked relation Place call lacks a destination "
"affordance."
)
self._validate_place_relation_grounder(
linked.call,
affordance=destination,
path=call.source_path,
)
return compiled
[docs]
def validate_engine(self, engine: AtomicActionEngine) -> None:
"""Require the live engine to expose every statically selected skill."""
if not isinstance(engine, AtomicActionEngine):
raise TypeError("engine must be an AtomicActionEngine.")
for skill_id, expected in self._required_skills.items():
actual = engine.skills.get(skill_id)
if actual != expected:
raise IntegrationFingerprintMismatch(
f"Live skill {skill_id!r} differs from the registered "
"semantic target descriptor."
)
[docs]
def validate_bound_endpoint_extensions(
self,
bound_profile: BoundRobotSkillProfile,
) -> None:
"""Match every live resolved endpoint to its fingerprinted declaration."""
if type(bound_profile) is not BoundRobotSkillProfile:
raise TypeError("bound_profile must be exactly BoundRobotSkillProfile.")
bound_profile.assert_current()
if bound_profile.profile_id != self.robot_profile_id:
raise IntegrationFingerprintMismatch(
"The bound robot profile ID differs from the registered profile."
)
transport_owner_by_target_type = {
target_type: transport
for transport in self.runtime_transport_declarations
for target_type in transport.target_types
}
expected_resource_ids = frozenset(self.robot_profile.resources)
live_resource_ids = frozenset(bound_profile.resources)
if live_resource_ids != expected_resource_ids:
raise IntegrationFingerprintMismatch(
"Bound robot resource IDs differ from the registered profile; "
f"expected {sorted(expected_resource_ids)}, "
f"got {sorted(live_resource_ids)}."
)
for resource_id, resource in bound_profile.resources.items():
expected_resource = self.robot_profile.resources[resource_id]
if resource.resource_id != expected_resource.resource_id:
raise IntegrationFingerprintMismatch(
f"Bound resource {resource_id!r} declaration ID differs from "
"the registered profile."
)
if resource.members != expected_resource.members:
raise IntegrationFingerprintMismatch(
f"Bound resource {resource_id!r} members differ from the "
"registered profile."
)
expected_endpoint_ids = frozenset(expected_resource.endpoints)
live_endpoint_ids = frozenset(resource.endpoints)
if live_endpoint_ids != expected_endpoint_ids:
raise IntegrationFingerprintMismatch(
f"Bound resource {resource_id!r} endpoint IDs differ from the "
f"registered profile; expected {sorted(expected_endpoint_ids)}, "
f"got {sorted(live_endpoint_ids)}."
)
for endpoint_id, endpoint in resource.endpoints.items():
location = f"{resource_id}.{endpoint_id}"
expected_endpoint = expected_resource.endpoints[endpoint_id]
if type(endpoint.endpoint) is not type(expected_endpoint) or (
_canonical_json(endpoint.endpoint)
!= _canonical_json(expected_endpoint)
):
raise IntegrationFingerprintMismatch(
f"Bound endpoint {location!r} declaration differs from the "
"registered robot profile."
)
endpoint_type = type(endpoint.endpoint)
declaration = self.endpoint_adapter_declarations.get(endpoint_type)
if declaration is None:
raise IntegrationFingerprintMismatch(
f"Bound endpoint {location!r} has undeclared exact type "
f"{_qualified_name(endpoint_type)!r}."
)
if endpoint.adapter_id != declaration.adapter_id:
raise IntegrationFingerprintMismatch(
f"Bound endpoint {location!r} adapter ID "
f"{endpoint.adapter_id!r} differs from registered "
f"{declaration.adapter_id!r}."
)
target = endpoint.runtime_target
target_type = type(target)
if target_type not in declaration.runtime_target_types:
raise IntegrationFingerprintMismatch(
f"Bound endpoint {location!r} resolved undeclared exact "
f"runtime target type {_qualified_name(target_type)!r}."
)
owner = transport_owner_by_target_type.get(target_type)
if owner is None or owner.transport_id not in (
declaration.runtime_transport_ids
):
raise IntegrationFingerprintMismatch(
f"Bound endpoint {location!r} target type has no registered "
"adapter transport owner."
)
if target.transport_id != owner.transport_id:
raise IntegrationFingerprintMismatch(
f"Bound endpoint {location!r} live transport "
f"{target.transport_id!r} differs from target type owner "
f"{owner.transport_id!r}."
)
feedback_keys = frozenset(
(binding.source.provider_id, binding.source.revision)
for binding in endpoint.tracking_channels.values()
)
if feedback_keys != declaration.tracking_feedback_source_keys:
raise IntegrationFingerprintMismatch(
f"Bound endpoint {location!r} tracking-feedback routes "
"differ from its registered adapter declaration."
)
projector_keys = frozenset(
(binding.projector.projector_id, binding.projector.revision)
for binding in endpoint.tracking_channels.values()
)
if projector_keys != declaration.tracking_projector_keys:
raise IntegrationFingerprintMismatch(
f"Bound endpoint {location!r} tracking-projector routes "
"differ from its registered adapter declaration."
)
evidence_keys = frozenset(
(source.provider_id, source.revision)
for source in endpoint.effect_sources.values()
)
if evidence_keys != declaration.effect_evidence_source_keys:
raise IntegrationFingerprintMismatch(
f"Bound endpoint {location!r} effect-evidence routes "
"differ from its registered adapter declaration."
)
if endpoint_type is ControlPartEndpoint:
control_part = endpoint.endpoint.control_part
if getattr(target, "control_part", None) != control_part:
raise IntegrationFingerprintMismatch(
f"Bound endpoint {location!r} runtime target addresses "
"a different control part."
)
if frozenset(endpoint.tracking_channels) != frozenset(
{JOINT_POSITION_CHANNEL}
):
raise IntegrationFingerprintMismatch(
f"Bound endpoint {location!r} must expose exactly the "
"built-in joint-position tracking channel."
)
tracking = endpoint.tracking_channels[JOINT_POSITION_CHANNEL]
feedback_address = tracking.source.address
if type(feedback_address) is not EndpointTrackingFeedbackAddress:
raise IntegrationFingerprintMismatch(
f"Bound endpoint {location!r} must use the exact "
"built-in endpoint tracking address."
)
if (
feedback_address.channel_id != JOINT_POSITION_CHANNEL
or type(feedback_address.target) is not target_type
or _canonical_json(feedback_address.target)
!= _canonical_json(target)
):
raise IntegrationFingerprintMismatch(
f"Bound endpoint {location!r} tracking address differs "
"from its runtime target or channel."
)
expected_effect_channels = {
POSE_RELATION_EFFECT_CHANNEL,
JOINT_STATE_EFFECT_CHANNEL,
}
if GRASP_CAPABILITY in endpoint.endpoint.capabilities:
expected_effect_channels.update(
{
CONTACT_EFFECT_CHANNEL,
CONSTRAINT_EFFECT_CHANNEL,
FORCE_EFFECT_CHANNEL,
}
)
if frozenset(endpoint.effect_sources) != frozenset(
expected_effect_channels
):
raise IntegrationFingerprintMismatch(
f"Bound endpoint {location!r} effect-evidence channels "
"differ from the exact built-in control-part routes."
)
for channel, source in endpoint.effect_sources.items():
address = source.address
if (
type(address) is not ControlPartEvidenceAddress
or address.control_part != control_part
or address.channel != channel
):
raise IntegrationFingerprintMismatch(
f"Bound endpoint {location!r} effect-evidence "
f"address for channel {channel!r} differs from its "
"control part or channel."
)
elif endpoint.tracking_channels or endpoint.effect_sources:
raise IntegrationFingerprintMismatch(
f"Bound custom endpoint {location!r} exposes closed-loop "
"routes forbidden by the C1 standard runtime."
)
def _registration_payload(
*,
scene_binding: SimulationSceneBinding,
scene: SceneManifest,
robot_profile_binding: SimulationRobotSkillProfileBinding,
robot_profile: RobotSkillProfile,
call_catalog: SemanticCallCatalog,
settle_presets: Mapping[str, DynamicSettleMonitorCfg],
relation_grounder_keys: frozenset[tuple[str, type[Affordance], str]],
relation_grounders: tuple[RelationTargetGrounder, ...],
handover_pose_providers: tuple[HandOverPoseProvider, ...],
extensions: StandardExtensionDeclarations,
endpoint_adapters: tuple[ResourceEndpointAdapter, ...],
runtime_transports: tuple[RuntimeTransportActionEncoder, ...],
parallel_safety_factory: ParallelCommandSafetyValidatorFactory | None,
control_part_evidence_factory: ControlPartEvidenceProviderFactory | None,
registered_semantic_lowerer_factories: tuple[RegisteredSemanticLowererFactory, ...],
) -> dict[str, object]:
"""Build the versioned canonical fingerprint payload."""
return {
"schema_version": _CATALOG_FINGERPRINT_SCHEMA_VERSION,
"scene_binding": scene_binding,
"scene_manifest": scene.entries,
"robot_profile_binding": robot_profile_binding,
"robot_profile": robot_profile,
"call_descriptors": tuple(
sorted(
call_catalog.descriptors.values(),
key=lambda descriptor: descriptor.call_id,
)
),
"relation_grounder_keys": relation_grounder_keys,
"relation_grounders": tuple(
{
"key": _relation_grounder_key(grounder),
"provider": _provider_fingerprint_declaration(grounder),
}
for grounder in sorted(
relation_grounders,
key=_relation_grounder_order_key,
)
),
"handover_pose_providers": tuple(
{
"provider_id": _handover_pose_provider_id(provider),
"provider": _provider_fingerprint_declaration(provider),
}
for provider in sorted(
handover_pose_providers,
key=_handover_pose_provider_id,
)
),
"standard_extensions": {
"endpoint_adapters": tuple(
sorted(
extensions.endpoint_adapters.values(),
key=lambda declaration: declaration.adapter_id,
)
),
"runtime_transports": extensions.runtime_transports,
"parallel_safety": extensions.parallel_safety,
"control_part_evidence": extensions.control_part_evidence,
"registered_semantic_lowerers": tuple(
sorted(
extensions.registered_semantic_lowerers.values(),
key=lambda declaration: declaration.call_id,
)
),
},
"endpoint_adapters": tuple(
{
"declaration": extensions.endpoint_adapters[
getattr(type(adapter), "endpoint_type")
],
"provider": _provider_fingerprint_declaration(adapter),
}
for adapter in sorted(
endpoint_adapters,
key=lambda value: getattr(type(value), "adapter_id"),
)
),
"runtime_transports": tuple(
{
"declaration": next(
declaration
for declaration in extensions.runtime_transports
if declaration.transport_id
== getattr(type(transport), "transport_id")
),
"provider": _provider_fingerprint_declaration(transport),
}
for transport in runtime_transports
),
"parallel_safety_factory": (
None
if parallel_safety_factory is None
else {
"declaration": extensions.parallel_safety,
"provider": _provider_fingerprint_declaration(parallel_safety_factory),
}
),
"control_part_evidence_factory": (
None
if control_part_evidence_factory is None
else {
"declaration": extensions.control_part_evidence,
"provider": _provider_fingerprint_declaration(
control_part_evidence_factory
),
}
),
"registered_semantic_lowerer_factories": tuple(
{
"declaration": extensions.registered_semantic_lowerers[
getattr(type(factory), "call_id")
],
"provider": _provider_fingerprint_declaration(factory),
}
for factory in sorted(
registered_semantic_lowerer_factories,
key=lambda value: getattr(type(value), "call_id"),
)
),
"post_policy_kinds": _POST_POLICY_KINDS,
"settle_presets": settle_presets,
"validator_kinds": _VALIDATOR_KINDS,
}
[docs]
@dataclass(frozen=True, slots=True)
class SimulationTaskProgramRegistration:
"""Exact immutable task-owned simulation integration registration."""
scene_binding: SimulationSceneBinding
robot_profile_binding: SimulationRobotSkillProfileBinding
call_catalog: SemanticCallCatalog = field(
default_factory=builtin_semantic_call_catalog
)
settle_presets: Mapping[str, DynamicSettleMonitorCfg] = field(
default_factory=default_simulation_settle_presets
)
relation_grounders: tuple[RelationTargetGrounder, ...] = ()
handover_pose_providers: tuple[HandOverPoseProvider, ...] = ()
endpoint_adapters: tuple[ResourceEndpointAdapter, ...] = ()
runtime_transports: tuple[RuntimeTransportActionEncoder, ...] = ()
parallel_safety_factory: ParallelCommandSafetyValidatorFactory | None = None
control_part_evidence_factory: ControlPartEvidenceProviderFactory | None = None
registered_semantic_lowerer_factories: tuple[
RegisteredSemanticLowererFactory, ...
] = ()
catalog: TaskProgramIntegrationCatalog = field(init=False)
_parallel_safety_validator_history: list[ParallelCommandSafetyValidator] = field(
init=False,
repr=False,
compare=False,
)
_parallel_safety_validator_lock: LockType = field(
init=False,
repr=False,
compare=False,
)
_control_part_evidence_provider_history: list[EffectEvidenceProvider] = field(
init=False,
repr=False,
compare=False,
)
_control_part_evidence_provider_lock: LockType = field(
init=False,
repr=False,
compare=False,
)
_registered_semantic_lowerer_history: list[RegisteredSemanticLowerer] = field(
init=False,
repr=False,
compare=False,
)
_registered_semantic_lowerer_lock: LockType = field(
init=False,
repr=False,
compare=False,
)
def __post_init__(self) -> None:
if type(self.scene_binding) is not SimulationSceneBinding:
raise TypeError("scene_binding must be exactly SimulationSceneBinding.")
if type(self.robot_profile_binding) is not SimulationRobotSkillProfileBinding:
raise TypeError(
"robot_profile_binding must be exactly "
"SimulationRobotSkillProfileBinding."
)
if type(self.call_catalog) is not SemanticCallCatalog:
raise TypeError("call_catalog must be exactly SemanticCallCatalog.")
settle_presets = _snapshot_settle_presets(self.settle_presets)
object.__setattr__(self, "settle_presets", settle_presets)
configured_relation_grounders = _snapshot_relation_grounders(
self.relation_grounders
)
relation_grounders = _snapshot_relation_grounders(
(
*_builtin_relation_grounders(self.scene_binding),
*configured_relation_grounders,
)
)
object.__setattr__(self, "relation_grounders", relation_grounders)
relation_grounder_keys = frozenset(
_relation_grounder_key(grounder) for grounder in relation_grounders
)
handover_pose_providers = _snapshot_handover_pose_providers(
self.handover_pose_providers
)
object.__setattr__(
self,
"handover_pose_providers",
handover_pose_providers,
)
scene = self.scene_binding.declare()
profile = self.robot_profile_binding.declare()
_validate_standard_effect_monitors(profile)
_validate_standard_tracking_metrics(profile)
extensions = build_standard_extension_declarations(
profile=profile,
endpoint_adapters=self.endpoint_adapters,
runtime_transports=self.runtime_transports,
parallel_safety_factory=self.parallel_safety_factory,
control_part_evidence_factory=self.control_part_evidence_factory,
registered_semantic_lowerer_factories=(
self.registered_semantic_lowerer_factories
),
)
_validate_standard_call_catalog(
self.call_catalog,
extensions.registered_semantic_lowerers,
)
selected_handover_provider = profile.grounding_providers.get("hand_over")
registered_handover_provider_ids = {
_handover_pose_provider_id(provider) for provider in handover_pose_providers
}
if (
selected_handover_provider is not None
and selected_handover_provider not in registered_handover_provider_ids
):
raise ValueError(
"Robot profile selects handover pose provider "
f"{selected_handover_provider!r}, but the task registration did "
"not install it."
)
builtin_skills = {
descriptor.skill_id: descriptor
for action_type in BUILTIN_ACTION_TYPES
if (descriptor := action_type.descriptor()).agent_visible
and descriptor.binding_contract is not None
}
required_skills: dict[str, SkillDescriptor] = {}
for descriptor in self.call_catalog.descriptors.values():
target = descriptor.target_descriptor
installed = builtin_skills.get(descriptor.skill_id)
if target is None or installed != target:
raise ValueError(
f"Semantic call {descriptor.call_id!r} targets skill "
f"{descriptor.skill_id!r}, which is not installed by the "
"standard simulation factory."
)
required_skills[descriptor.skill_id] = target
fingerprint = _digest(
_registration_payload(
scene_binding=self.scene_binding,
scene=scene,
robot_profile_binding=self.robot_profile_binding,
robot_profile=profile,
call_catalog=self.call_catalog,
settle_presets=settle_presets,
relation_grounder_keys=relation_grounder_keys,
relation_grounders=relation_grounders,
handover_pose_providers=handover_pose_providers,
extensions=extensions,
endpoint_adapters=self.endpoint_adapters,
runtime_transports=self.runtime_transports,
parallel_safety_factory=self.parallel_safety_factory,
control_part_evidence_factory=self.control_part_evidence_factory,
registered_semantic_lowerer_factories=(
self.registered_semantic_lowerer_factories
),
)
)
object.__setattr__(
self,
"catalog",
TaskProgramIntegrationCatalog(
scene_registry_id=self.scene_binding.registry_id,
robot_profile_id=self.robot_profile_binding.profile_id,
scene=scene,
robot_profile=profile,
call_catalog=self.call_catalog,
relation_grounder_keys=relation_grounder_keys,
settle_preset_ids=frozenset(settle_presets),
endpoint_adapter_declarations=extensions.endpoint_adapters,
runtime_transport_declarations=extensions.runtime_transports,
parallel_safety_declaration=extensions.parallel_safety,
control_part_evidence_declaration=extensions.control_part_evidence,
registered_semantic_lowerer_declarations=(
extensions.registered_semantic_lowerers
),
fingerprint=fingerprint,
_required_skills=required_skills,
),
)
object.__setattr__(self, "_parallel_safety_validator_history", [])
object.__setattr__(self, "_parallel_safety_validator_lock", Lock())
object.__setattr__(self, "_control_part_evidence_provider_history", [])
object.__setattr__(self, "_control_part_evidence_provider_lock", Lock())
object.__setattr__(self, "_registered_semantic_lowerer_history", [])
object.__setattr__(self, "_registered_semantic_lowerer_lock", Lock())
@property
def fingerprint(self) -> str:
"""Return the canonical registration fingerprint."""
return self.catalog.fingerprint
[docs]
def assert_unchanged(self) -> None:
"""Reject nested declaration drift before live component creation."""
scene = self.scene_binding.declare()
profile = self.robot_profile_binding.declare()
try:
_validate_standard_effect_monitors(profile)
_validate_standard_tracking_metrics(profile)
extensions = build_standard_extension_declarations(
profile=profile,
endpoint_adapters=self.endpoint_adapters,
runtime_transports=self.runtime_transports,
parallel_safety_factory=self.parallel_safety_factory,
control_part_evidence_factory=self.control_part_evidence_factory,
registered_semantic_lowerer_factories=(
self.registered_semantic_lowerer_factories
),
)
_validate_standard_call_catalog(
self.call_catalog,
extensions.registered_semantic_lowerers,
)
relation_grounders = _snapshot_relation_grounders(self.relation_grounders)
relation_grounder_keys = frozenset(
_relation_grounder_key(grounder) for grounder in relation_grounders
)
handover_pose_providers = _snapshot_handover_pose_providers(
self.handover_pose_providers
)
current = _digest(
_registration_payload(
scene_binding=self.scene_binding,
scene=scene,
robot_profile_binding=self.robot_profile_binding,
robot_profile=profile,
call_catalog=self.call_catalog,
settle_presets=self.settle_presets,
relation_grounder_keys=relation_grounder_keys,
relation_grounders=relation_grounders,
handover_pose_providers=handover_pose_providers,
extensions=extensions,
endpoint_adapters=self.endpoint_adapters,
runtime_transports=self.runtime_transports,
parallel_safety_factory=self.parallel_safety_factory,
control_part_evidence_factory=self.control_part_evidence_factory,
registered_semantic_lowerer_factories=(
self.registered_semantic_lowerer_factories
),
)
)
except (TypeError, ValueError) as exc:
raise IntegrationFingerprintMismatch(
"Task Program integration provider declaration changed after "
"task registration."
) from exc
if current != self.fingerprint:
raise IntegrationFingerprintMismatch(
"Task Program integration declaration changed after task "
"registration."
)
@property
def endpoint_adapter_map(
self,
) -> Mapping[type[ResourceEndpoint], ResourceEndpointAdapter]:
"""Return custom live adapters keyed by their exact endpoint type."""
return MappingProxyType(
{
getattr(type(adapter), "endpoint_type"): adapter
for adapter in self.endpoint_adapters
}
)
[docs]
def create_registered_semantic_lowerers(
self,
*,
simulation: object,
robot: object,
scene_registry: SceneRegistry,
engine: AtomicActionEngine,
) -> tuple[RegisteredSemanticLowerer, ...]:
"""Create fresh live lowerers declared by this registration."""
self.assert_unchanged()
if type(scene_registry) is not SceneRegistry:
raise TypeError("scene_registry must be exactly SceneRegistry.")
if not isinstance(engine, AtomicActionEngine):
raise TypeError("engine must be an AtomicActionEngine.")
if engine.robot is not robot:
raise ValueError("engine and factory must reference the exact same robot.")
created: list[RegisteredSemanticLowerer] = []
with self._registered_semantic_lowerer_lock:
for factory in self.registered_semantic_lowerer_factories:
call_id = _exact_identifier(
getattr(type(factory), "call_id", None),
field_name="RegisteredSemanticLowererFactory.call_id",
)
lowerer = factory.create(
simulation=simulation,
robot=robot,
scene_registry=scene_registry,
engine=engine,
)
if not isinstance(lowerer, RegisteredSemanticLowerer):
raise TypeError(
"RegisteredSemanticLowererFactory.create() must return a "
"RegisteredSemanticLowerer."
)
if any(
lowerer is previous
for previous in self._registered_semantic_lowerer_history
):
raise ValueError(
"RegisteredSemanticLowererFactory.create() must return a "
"fresh lowerer for every runtime assembly owned by this "
"registration."
)
descriptor = self.call_catalog.discover(call_id)
lowerer_type = type(lowerer)
if getattr(lowerer_type, "call_id", None) != call_id:
raise ValueError(
"The live registered lowerer call_id must match its "
"factory declaration."
)
if (
getattr(lowerer_type, "target_descriptor", None)
!= descriptor.target_descriptor
):
raise ValueError(
f"Live lowerer {call_id!r} target_descriptor must exactly "
"match the registered catalog target."
)
self._registered_semantic_lowerer_history.append(lowerer)
created.append(lowerer)
return tuple(created)
[docs]
def create_control_part_evidence_provider(
self,
*,
simulation: object,
robot: object,
scene_registry: SceneRegistry,
engine: AtomicActionEngine,
scene_provider: SceneProvider,
) -> EffectEvidenceProvider | None:
"""Create the registration-owned live control-part evidence provider.
Args:
simulation: Simulation that owns the selected robot and sensors.
robot: Exact robot selected by the environment factory.
scene_registry: Exact live semantic scene registry.
engine: Atomic-action engine assembled for the robot.
scene_provider: Shared synchronized live scene provider.
Returns:
Fresh registered provider, or ``None`` when no factory is declared.
"""
self.assert_unchanged()
factory = self.control_part_evidence_factory
if factory is None:
return None
if type(scene_registry) is not SceneRegistry:
raise TypeError("scene_registry must be exactly SceneRegistry.")
if not isinstance(engine, AtomicActionEngine):
raise TypeError("engine must be an AtomicActionEngine.")
if engine.robot is not robot:
raise ValueError("engine and factory must reference the exact same robot.")
if not isinstance(scene_provider, SceneProvider):
raise TypeError("scene_provider must implement SceneProvider.")
declaration = self.catalog.control_part_evidence_declaration
if declaration is None:
raise AssertionError(
"A registered control-part evidence factory lost its declaration."
)
with self._control_part_evidence_provider_lock:
provider = factory.create(
simulation=simulation,
robot=robot,
scene_registry=scene_registry,
engine=engine,
scene_provider=scene_provider,
)
if not isinstance(provider, EffectEvidenceProvider):
raise TypeError(
"control_part_evidence_factory.create() must return an "
"EffectEvidenceProvider."
)
if (provider.provider_id, provider.revision) != (
declaration.provider_id,
declaration.revision,
):
raise ValueError(
"The live control-part evidence provider identity must match "
"its exact registration declaration."
)
if any(
provider is previous
for previous in self._control_part_evidence_provider_history
):
raise ValueError(
"ControlPartEvidenceProviderFactory.create() must return a "
"fresh provider for every runtime assembly owned by this "
"registration."
)
self._control_part_evidence_provider_history.append(provider)
return provider
[docs]
def create_parallel_safety_validator(
self,
*,
simulation: object,
robot: object,
scene_registry: SceneRegistry,
engine: AtomicActionEngine,
) -> ParallelCommandSafetyValidator | None:
"""Create and strictly validate the registration-owned live safety gate."""
self.assert_unchanged()
factory = self.parallel_safety_factory
if factory is None:
return None
if type(scene_registry) is not SceneRegistry:
raise TypeError("scene_registry must be exactly SceneRegistry.")
if not isinstance(engine, AtomicActionEngine):
raise TypeError("engine must be an AtomicActionEngine.")
if engine.robot is not robot:
raise ValueError("engine and factory must reference the exact same robot.")
with self._parallel_safety_validator_lock:
validator = factory.create(
simulation=simulation,
robot=robot,
scene_registry=scene_registry,
engine=engine,
)
if not isinstance(validator, ParallelCommandSafetyValidator):
raise TypeError(
"parallel_safety_factory.create() must return a "
"ParallelCommandSafetyValidator."
)
if any(
validator is previous
for previous in self._parallel_safety_validator_history
):
raise ValueError(
"ParallelCommandSafetyValidatorFactory.create() must return a "
"fresh validator for every runtime assembly owned by this "
"registration."
)
self._parallel_safety_validator_history.append(validator)
return validator
[docs]
def validate_scene_registry(self, registry: SceneRegistry) -> None:
"""Validate a live registry against the registered scene declaration."""
self.assert_unchanged()
self.catalog.scene.validate_registry(registry)
[docs]
def validate_engine(self, engine: AtomicActionEngine) -> None:
"""Validate live atomic skills against this registration."""
self.assert_unchanged()
self.catalog.validate_engine(engine)
[docs]
def validate_bound_profile(
self,
bound_profile: BoundRobotSkillProfile,
) -> None:
"""Validate resolved semantic endpoints against this registration."""
self.assert_unchanged()
self.catalog.validate_bound_endpoint_extensions(bound_profile)
[docs]
def validate_robot_profile(
self,
profile: RobotSkillProfile,
) -> None:
"""Validate a live profile against its registered declaration."""
self.assert_unchanged()
if type(profile) is not RobotSkillProfile:
raise TypeError("profile must be exactly RobotSkillProfile.")
if _canonical_json(profile) != _canonical_json(self.catalog.robot_profile):
raise IntegrationFingerprintMismatch(
"Live robot skill profile differs from the registered declaration."
)
__all__ = [
"TaskProgramIntegrationCatalog",
"IntegrationFingerprintMismatch",
"SimulationTaskProgramRegistration",
]