"""Typed data contracts for the Brain-5D embodiment layer.""" from __future__ import annotations from dataclasses import dataclass from enum import StrEnum JSONScalar = str | int | float | bool | None JSONValue = JSONScalar | list["JSONValue"] | dict[str, "JSONValue"] class EnvironmentKind(StrEnum): """Supported environment categories.""" SIMULATED = "simulated" PHYSICAL = "physical" DIGITAL = "digital" HYBRID = "hybrid" @dataclass(frozen=True, slots=True) class SensorFrame: """One timestamped sensor observation.""" sensor_id: str tick: int modality: str payload: JSONValue @dataclass(frozen=True, slots=True) class ActionCommand: """One typed command produced for an actuator or environment.""" actuator_id: str tick: int action: str payload: JSONValue = None @dataclass(frozen=True, slots=True) class ActuatorResult: """Result returned by one actuator invocation.""" accepted: bool message: str = "" @dataclass(frozen=True, slots=True) class ActionReceipt: """Execution receipt separating acceptance from observed environment effect.""" command_id: str accepted: bool started: bool completed: bool failed: bool latency: int | None = None error: str | None = None safety_clamp: str | None = None effect_observed: bool | None = None def to_json(self) -> dict[str, JSONValue]: return { "command_id": self.command_id, "accepted": self.accepted, "started": self.started, "completed": self.completed, "failed": self.failed, "latency": self.latency, "error": self.error, "safety_clamp": self.safety_clamp, "effect_observed": self.effect_observed, } @dataclass(frozen=True, slots=True) class EnvironmentObservation: """Observation returned by an environment after reset or action.""" tick: int state: dict[str, JSONValue] reward: float = 0.0 terminated: bool = False truncated: bool = False @dataclass(frozen=True, slots=True) class EmbodimentMetrics: """Read-only operating metrics for dashboard publication.""" environment_kind: str = "unconfigured" active_sensors: int = 0 active_actuators: int = 0 episode: int = 0 episode_reward: float = 0.0 last_reward: float = 0.0 last_action: str = "" last_text_input: str = "" last_observation_state: dict[str, JSONValue] | None = None last_observation_tick: int | None = None last_observation_terminated: bool | None = None last_observation_truncated: bool | None = None def to_json(self) -> dict[str, JSONValue]: """Return a JSON-ready representation.""" return { "environment_kind": self.environment_kind, "active_sensors": self.active_sensors, "active_actuators": self.active_actuators, "episode": self.episode, "episode_reward": self.episode_reward, "last_reward": self.last_reward, "last_action": self.last_action, "last_text_input": self.last_text_input, "last_observation_state": self.last_observation_state, "last_observation_tick": self.last_observation_tick, "last_observation_terminated": self.last_observation_terminated, "last_observation_truncated": self.last_observation_truncated, }