Embodiment¶

Embodiment -- mvp.embodiment

Cluster: Physical AI | Type: component | MCP Tools: 31

Overview¶

Simulation-only embodiment management layer for registering simulated actuators and sensors, tracking energy budgets per inference cycle, exposing current affordances based on base type, and persisting actuator/sensor logs, calibration records, and morphology configurations via the MCP block's SQLite store.

When to use:

• Building a software-only sensor/actuator sandbox for G6 workflows
• Tracking token, API-call, and wall-clock energy costs for simulated control cycles
• Recording morphology, calibration, and readiness state before a separate hardware integration exists
• Querying what actions a simulated body can currently perform given its base type

Do not use it for:

• Driving physical robots, PLCs, GPIO pins, serial devices, cameras, lidar, ROS/ROS2 topics, Modbus devices, or safety-critical actuators
• Certifying real-world motion plans or treating simulated readiness as operational safety approval

Example:

from mvp.embodiment import EmbodimentBlock, EmbodimentInput

block = EmbodimentBlock(name="embodiment")
block.infer(EmbodimentInput(op="register_actuator", actuator_id="arm_0", actuator_type="simulated"))
block.infer(EmbodimentInput(op="register_sensor", sensor_id="temp_0", sensor_type="simulated"))
block.infer(EmbodimentInput(op="set_base_type", base_type="mcp"))
result = block.infer(EmbodimentInput(op="get_affordances"))
# result.value.affordances -> ["tool_calling", "resource_access", ..., "closed_loop_control"]

Works well with: motor_control, sensory_fusion, realtime_bridge

The generated API reference below includes HAL protocol classes and safety-envelope types for future backend work. In the launch build, the supported end-user path is the in-memory/MCP simulation workflow.

Public API¶

EmbodimentBlock(AIBlock[EmbodimentInput, EmbodimentOutput, dict])¶

Physical embodiment management block (20 ops).

Methods:

infer(data: EmbodimentInput) -> Result[EmbodimentOutput]¶

SensorFrame¶

A single sensor reading with timestamp and metadata.

RGBFrame(SensorFrame)¶

RGB camera frame (width x height x 3 numpy array or list).

DepthFrame(SensorFrame)¶

Depth camera frame (width x height float array, meters).

PointCloud(SensorFrame)¶

LiDAR point cloud (Nx3 or Nx4 array).

JointState¶

State of a single joint.

GripperState¶

State of a gripper.

Odometry¶

Mobile base odometry.

Twist¶

Velocity command (linear + angular).

Pose¶

6-DOF pose goal.

SensorStream(Protocol)¶

Async iterator protocol for sensor data streams.

Methods:

start() -> None¶

Begin streaming sensor data.

stop() -> None¶

Stop streaming sensor data.

read() -> SensorFrame¶

Read a single frame.

CameraStream(Protocol)¶

RGB/Depth camera stream.

Methods:

start() -> None¶

stop() -> None¶

read_rgb() -> RGBFrame¶

read_depth() -> DepthFrame¶

read() -> SensorFrame¶

resolution() -> tuple[int, int]¶

fps() -> float¶

LidarStream(Protocol)¶

LiDAR point cloud stream.

Methods:

start() -> None¶

stop() -> None¶

read() -> PointCloud¶

max_range() -> float¶

num_beams() -> int¶

JointController(Protocol)¶

Control interface for robot joints.

Methods:

set_position(joint_name: str, position: float) -> None¶

Command a joint to a target position (radians).

set_velocity(joint_name: str, velocity: float) -> None¶

Command a joint to a target velocity (rad/s).

get_state(joint_name: str) -> JointState¶

Get current joint state.

get_all_states() -> dict[str, JointState]¶

Get all joint states.

joint_names() -> list[str]¶

Gripper(Protocol)¶

Gripper control interface.

Methods:

open(width: float = 1.0) -> None¶

Open gripper to specified width (0-1 normalized).

close(force: float = 0.5) -> None¶

Close gripper with specified force (0-1 normalized).

get_state() -> GripperState¶

Get current gripper state.

MobileBase(Protocol)¶

Mobile base control interface.

Methods:

set_velocity(twist: Twist) -> None¶

Send velocity command.

set_pose_goal(pose: Pose) -> None¶

Set a pose goal for navigation.

get_odometry() -> Odometry¶

Get current odometry.

stop() -> None¶

Emergency stop the base.

HAL(Protocol)¶

Hardware Abstraction Layer — factory for sensors and actuators.

Methods:

backend() -> HALBackend¶

initialize() -> None¶

Initialize the HAL backend.

shutdown() -> None¶

Gracefully shutdown all hardware connections.

emergency_stop() -> None¶

Trigger emergency stop on all actuators.

get_camera(name: str) -> CameraStream¶

Get a camera stream by name.

get_lidar(name: str) -> LidarStream¶

Get a lidar stream by name.

get_joint_controller() -> JointController¶

Get the joint controller.

get_gripper(name: str) -> Gripper¶

Get a gripper by name.

get_mobile_base() -> MobileBase¶

Get the mobile base controller.

watchdog_kick() -> None¶

Kick the hardware watchdog (must be called periodically).

watchdog_timeout_s() -> float¶

is_estopped() -> bool¶

HALBackend(Enum)¶

Supported HAL backend types.

HALFactory¶

Creates HAL instances for the specified backend.

Methods:

create(backend: HALBackend, config: dict | None = None) -> HAL¶

Create a HAL instance for the given backend.

create_with_result(backend: HALBackend, config: dict | None = None) -> HALCreationResult¶

Create a HAL and report experimental fallback instead of hiding it.

SafetyIntegrityLevel(Enum)¶

IEC 61508 Safety Integrity Levels.

CollaborativeMode(Enum)¶

ISO/TS 15066 collaborative operation modes.

JointLimits¶

Position, velocity, and effort limits for a single joint.

CartesianLimits¶

Cartesian workspace limits.

ForceLimits¶

Force and power limits per ISO/TS 15066.

SafetyEnvelope¶

Defines the complete safety boundary for a robot system.

Methods:

validate_joint_position(joint_name: str, position: float) -> SafetyViolation | None¶

Check if a joint position command is within limits.

validate_joint_velocity(joint_name: str, velocity: float) -> SafetyViolation | None¶

Check if a joint velocity command is within limits.

validate_twist(twist: Twist) -> SafetyViolation | None¶

Check if a base velocity command is within Cartesian limits.

validate_pose(pose: Pose) -> SafetyViolation | None¶

Check if a pose goal is within the workspace.

validate_gripper_force(force: float) -> SafetyViolation | None¶

Check if gripper force is within limits.

ViolationType(Enum)¶

Types of safety violations.

SafetyViolation¶

Record of a safety limit violation.

SafetyError(RuntimeError)¶

Raised when a safety violation prevents command execution.

SafeHAL¶

Safety wrapper around any HAL implementation.

Methods:

backend() -> HALBackend¶

watchdog_timeout_s() -> float¶

is_estopped() -> bool¶

violations() -> list[SafetyViolation]¶

Get list of all recorded safety violations.

initialize() -> None¶

Initialize the underlying HAL.

shutdown() -> None¶

Shutdown the underlying HAL.

emergency_stop() -> None¶

Trigger emergency stop on the underlying HAL.

reset_estop() -> None¶

Reset e-stop state (requires explicit action).

get_camera(name: str) -> CameraStream¶

Cameras are read-only -- no safety wrapping needed.

get_lidar(name: str) -> LidarStream¶

LiDAR is read-only -- no safety wrapping needed.

get_joint_controller() -> SafeJointController¶

Get safety-wrapped joint controller.

get_gripper(name: str) -> SafeGripper¶

Get safety-wrapped gripper.

get_mobile_base() -> SafeMobileBase¶

Get safety-wrapped mobile base.

watchdog_kick() -> None¶

Kick the safety watchdog.

check_watchdog() -> bool¶

Check if watchdog is still alive. Triggers e-stop if expired.

EmbodimentInput(BaseModel)¶

Input to EmbodimentBlock.

EmbodimentOutput(BaseModel)¶

Output from EmbodimentBlock.

EmbodimentMCPBlock(AIBlock[MCPEmbodimentInput, MCPEmbodimentOutput, dict])¶

31-op embodiment MCP block with SQLite persistence.

Methods:

infer(data: MCPEmbodimentInput) -> Result[MCPEmbodimentOutput]¶

MCPEmbodimentInput(BaseModel)¶

Input to EmbodimentMCPBlock — 31 ops.

MCPEmbodimentOutput(BaseModel)¶

Output from EmbodimentMCPBlock.

EmbodimentStore¶

SQLite-backed store for the embodiment MCP sub-package.

Constructor:

Methods:

store_config(name: str, config_json: str, tags: str = '') -> str¶

load_config(name: str) -> dict | None¶

list_configs(limit: int = 50) -> list[dict]¶

search_configs(query: str, top_k: int = 10) -> list[dict]¶

store_actuator_log(actuator_id: str, value: float = 0.0, action: str = '', metadata: dict | None = None) -> str¶

list_actuator_logs(actuator_id: str = '', limit: int = 50) -> list[dict]¶

store_sensor_log(sensor_id: str, value: float = 0.0, action: str = '', metadata: dict | None = None) -> str¶

list_sensor_logs(sensor_id: str = '', limit: int = 50) -> list[dict]¶

store_calibration(sensor_id: str, params: dict, result: dict | None = None) -> str¶

get_calibration(sensor_id: str) -> dict | None¶

store_morphology(morphology_id: str, spec: dict, tags: str = '') -> str¶

get_morphology(morphology_id: str) -> dict | None¶

list_morphologies(limit: int = 50) -> list[dict]¶

log_error(op: str, error_message: str, params: dict | None = None) -> None¶

store_energy_budget_log(energy: float, computational: float = 0.0, communication: float = 0.0, storage: float = 0.0, budget_ceiling: float = 0.0, budget_remaining: float = 0.0) -> str¶

list_energy_budget_logs(limit: int = 50) -> list[dict]¶

store_control_log(actuator_id: str, setpoint: float = 0.0, measured: float = 0.0, output: float = 0.0, error: float = 0.0, p_term: float = 0.0, i_term: float = 0.0, d_term: float = 0.0, was_rate_limited: bool = False) -> str¶

list_control_logs(actuator_id: str = '', limit: int = 50) -> list[dict]¶

count_all() -> dict[str, int]¶

health() -> dict¶

Functions¶

derive_affordances(base_type: str, sensor_ids: list[str], actuator_ids: list[str], base_affordances: dict[str, list[str]]) -> list[str]¶

score_affordance(affordance: str, sensor_health: dict[str, float], calibration_status: dict[str, bool], energy_ok: bool) -> float¶

morphology_similarity(spec_a: dict, spec_b: dict) -> dict¶

health_readiness(energy_budget_ok: bool, sensor_health: dict[str, float], health_threshold: float = 0.5, min_healthy_sensors: int = 0) -> dict¶

pid_step(setpoint: float, measured: float, prev_error: float, integral: float, kp: float, ki: float, kd: float, dt: float, integral_min: float = -100.0, integral_max: float = 100.0) -> tuple[float, float, float]¶

ramp_limit(current: float, target: float, max_rate: float, dt: float) -> float¶

control_step(setpoint: float, measured: float, prev_error: float, integral: float, current_output: float, kp: float, ki: float, kd: float, dt: float, max_rate: float = 0.0, integral_min: float = -100.0, integral_max: float = 100.0) -> dict¶

compute_energy(tokens: int, api_calls: int, wall_clock_s: float, weights: dict | None = None) -> dict¶

burn_rate(log: list[tuple[float, float]], window_seconds: float, now: float) -> float¶

depletion_forecast(budget_remaining: float, current_burn_rate: float) -> float¶

process_sensor_reading(raw: float, offset: float, scale: float, prev_ema: float, alpha: float, count: int, mean: float, m2: float, n_sigma: float = 3.0) -> dict¶

MCP Tools¶