Spaces:
Paused
Paused
File size: 3,454 Bytes
31226fd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | """Typed data contracts for the MHRN 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,
}
|