Integration¶
Cross-modal integration adapters for Physical AI + Multimodal wiring.
Cluster: Core Infrastructure | Type: component | MCP Tools: None
Overview¶
Cross-modal wiring library that converts data between G6 Physical AI and Multimodal components without custom glue code. Provides pure functions for every common handoff — image/voice output to sensor readings, mesh toolpaths to G-code, G-code to motor actions, physics trajectories to scene updates, stats to chart data — so pipelines read as linear data flows rather than adapter spaghetti.
Scope and production caveat
integration is a lightweight adapter library, not a full production integration platform. It does not provide workflow orchestration, retries, observability, persistence, deployment controls, hardware validation, or compliance-grade audit trails. Use it to convert payloads between known G6 components in pilot and launch workflows, then run the receiving component's validation and safety checks before relying on the output. For unattended or customer-facing production workflows, wrap these adapters in an orchestrator with logging, error handling, rollback, and human approval gates where risk warrants it.
When to use:
- Feeding
adapt_imageoradapt_voiceoutput directly intosensory_fusionsensor readings - Converting a
mesh3dtoolpath to a G-code string and then to amotor_controlaction - Translating
physics_predictiontrajectory output into asynthetic_worldmove sequence
Example:
from mvp.integration import image_to_sensor_reading, toolpath_to_gcode, gcode_to_action
# Minimal dict-shaped inputs (in practice, supply real adapt_image / mesh3d output)
image_block_output = {"width": 64, "height": 64, "modality": "image"}
mesh3d_output = {"toolpath": [[0.0, 0.0, 0.0], [1.0, 0.0, 0.0], [1.0, 1.0, 0.0]]}
sensor_reading = image_to_sensor_reading(image_block_output)
gcode = toolpath_to_gcode(mesh3d_output, feed_rate=800.0)
result = gcode_to_action(gcode, device_id="cnc_0")
# result → an ActionRequest-shaped dict for motor_control
Works well with: mesh3d, motor_control, sensory_fusion, physics_prediction
Public API¶
IntegrationBlock(AIBlock)¶
Routes cross-modal integration operations to the appropriate adapter.
Methods:
infer(input: IntegrationInput) -> Result[IntegrationOutput]¶
IntegrationInput(BaseModel)¶
Unified input for cross-modal integration pipelines.
| Field | Type | Default |
|---|---|---|
op | INTEGRATION_OP | 'get_info' |
data | dict[str, Any] | Field(default_factory=dict) |
pipeline | str | '' |
metadata | dict[str, Any] | Field(default_factory=dict) |
IntegrationOutput(BaseModel)¶
Unified output from cross-modal integration pipelines.
| Field | Type | Default |
|---|---|---|
op | str | '' |
result | dict[str, Any] | Field(default_factory=dict) |
pipeline | str | '' |
error | str | '' |
completion_state | Literal['verified', 'qualified-draft', 'blocked-escalated'] | 'qualified-draft' |
warning_card | dict[str, Any] \| None | None |
evidence | dict[str, Any] \| None | None |
request_id | str | '' |
task_id | str | '' |
run_id | str | '' |
metadata | dict[str, Any] | Field(default_factory=dict) |
degraded | bool | False |
degradation_reason | str \| None | None |
Functions¶
toolpath_to_gcode(mesh3d_output: dict[str, Any], feed_rate: float = 1000.0, layer_height: float = 0.2) -> str¶
Serialize mesh3d toolpath output to G-code string.
gcode_to_action(gcode: str, device_id: str = '') -> dict[str, Any]¶
Wrap G-code string in a motor_control execute_action input.
heightmap_to_fabrication_chain(heightmap_output: dict[str, Any], feed_rate: float = 1000.0, layer_height: float = 0.2, device_id: str = '') -> dict[str, Any]¶
Full chain: heightmap mesh output -> gcode -> motor action.
mesh_to_slicing_input(vertices: list[list[float]], faces: list[list[int]], layer_height: float = 0.2) -> dict[str, Any]¶
Prepare mesh data for mesh3d.slice_for_printing.
image_to_sensor_reading(image_output: dict[str, Any]) -> dict[str, Any]¶
Convert adapt_image output to sensory_fusion sensor_readings format.
voice_to_sensor_reading(voice_output: dict[str, Any]) -> dict[str, Any]¶
Convert adapt_voice STT output to sensory_fusion sensor_readings format.
batch_to_sensor_readings(outputs: list[dict[str, Any]], modality: str = 'image') -> list[dict[str, Any]]¶
Convert a batch of image or voice outputs to sensor_readings list.
tactile_to_sensor_reading(tactile_output: dict[str, Any]) -> dict[str, Any]¶
Convert tactile fusion output to sensory_fusion sensor_readings format.
sim_to_scene(physical_ai_output: dict[str, Any]) -> dict[str, Any]¶
Convert physical_ai trajectory output to synthetic_world move_object input.
scene_to_sim(synthetic_world_output: dict[str, Any]) -> dict[str, Any]¶
Convert a synthetic_world scene into a physical_ai simulation input.
scene_to_obstacles(synthetic_world_output: dict[str, Any]) -> list[dict[str, Any]]¶
Extract object positions from synthetic_world scene as obstacle dicts.
trajectory_to_scene_path(physical_ai_output: dict[str, Any], object_id: str = 'agent') -> list[dict[str, Any]]¶
Convert a full trajectory to a sequence of move_object commands.
stats_to_chart_data(stats: dict[str, Any], preset: str = '') -> dict[str, Any]¶
Convert component stats dict to VisualisationInput format.
trajectory_to_chart_data(trajectory: list[list[float]]) -> dict[str, Any]¶
Convert a trajectory (list of [x,y,z]) to scatter chart input.
autonomy_history_to_chart_data(history: list[dict[str, Any]]) -> dict[str, Any]¶
Convert autonomy governor escalation history to line chart.
stats_to_dashboard_spec(component_stats: dict[str, dict[str, Any]]) -> dict[str, Any]¶
Convert multiple component stats to adapt_diagrams dashboard spec.
parse_voice_command(text: str, dialogue_buffer: list[dict[str, Any]] | None = None) -> dict[str, Any]¶
Parse transcribed voice text into a structured command dict.
checkpoint_to_speech(governor_output: dict[str, Any]) -> dict[str, Any]¶
Convert autonomy_governor checkpoint to adapt_voice TTS input.
build_safety_check_input(command: dict[str, Any]) -> dict[str, Any]¶
Wrap a parsed command in a motor_control safety_check request.