Sensory Fusion¶

Sensory Fusion -- mvp.sensory_fusion

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

Overview¶

Simulated sensor inputs

sensory_fusion fuses data supplied by callers. It does not include real camera, lidar, IMU, force/torque, or ROS sensor drivers.

Not production robotics perception

Treat this component as a simulation/pre-adapted sensor fusion utility, not as a real hardware perception stack. It has useful validation, confidence weighting, EKF-style state estimation, filtering, outlier checks, and MCP persistence, but it does not handle live sensor streams, timestamp alignment, calibration, frame transforms, camera/LiDAR drivers, ROS bridges, or hardware safety. Use it for structured workflow reliability and reviewed planning artifacts; real robotics deployments need a separate validated perception and hardware integration layer.

Multi-modal sensor fusion block that combines caller-supplied structured readings (text, numeric, image summaries, point-cloud summaries, audio summaries) into a single confidence-weighted fused value. Implements attention-based source weighting, Extended Kalman Filter state estimation, signal filtering (low-pass/high-pass/bandpass/Kalman), outlier rejection, and active perception queries for simulated or pre-adapted perception workflows.

source_capabilities provides a read-only capability matrix for the 15 Tier-1 functional ops and 29 MCP functional ops, including optional dependency degradation, redacted agentic backend status, simulation-only maturity, and the production reviewer-signature gate. Outputs use only verified, qualified-draft, or blocked-escalated completion states with top-level warning/evidence fields.

When to use:

Fusing pre-adapted camera, LiDAR, and force-torque summaries into a unified simulated world-state estimate
Running an EKF over a noisy 8-dimensional state vector (position, velocity, contact force, confidence)
Generating active perception queries when sensor confidence falls below a threshold

Example:

from mvp.sensory_fusion import SensoryFusionBlock, SensoryFusionInput

block = SensoryFusionBlock(name="sensory_fusion")
result = block.infer(SensoryFusionInput(
    op="fuse",
    sensor_readings=[
        {"source": "camera", "value": 0.8, "confidence": 0.9, "timestamp": 1.0},
        {"source": "lidar", "value": 0.75, "confidence": 0.7, "timestamp": 1.0},
    ],
))
# result.value.fused_value -> dict; result.value.confidence -> float

Works well with: embodiment, tactile_fusion, motor_control, integration

Public API¶

`SensoryFusionInput(BaseModel)`¶

Input to SensoryFusionBlock.

Field	Type	Default
`op`	`Literal['fuse', 'attend', 'query_sensor', 'assess_confidence', 'active_perception', 'add_sensor_source', 'remove_sensor_source', 'list_sources', 'get_noise_model', 'filter_signal', 'get_info', 'reset', 'fuse_ekf', 'reject_outliers', 'configure_ekf', 'source_capabilities']`	'fuse'
`sensor_readings`	`list[dict]`	Field(default_factory=list)
`source_id`	`str`	''
`source_type`	`str`	''
`source_weight`	`float`	1.0
`goal`	`str`	''
`prior_observations`	`list[dict]`	Field(default_factory=list)
`signal`	`list[float]`	Field(default_factory=list)
`filter_type`	`str`	'lowpass'
`cutoff_freq`	`float`	0.1
`sample_rate`	`float`	1.0
`confidence_threshold`	`float`	0.5
`metadata`	`dict`	Field(default_factory=dict)
`state_vector`	`list[float]`	Field(default_factory=list)
`process_noise`	`float`	0.01
`measurement_noise`	`float`	0.1
`observation_model`	`list[list[float]]`	Field(default_factory=list)
`observation_vector`	`list[float]`	Field(default_factory=list)
`outlier_sigma`	`float`	3.0
`confidence_floor`	`float`	0.2
`run_mode`	`str`	'beta'
`reviewer_signature`	`str`	''
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`SensoryFusionOutput(BaseModel)`¶

Output from SensoryFusionBlock.

Field	Type	Default
`op`	`str`	required
`completion_state`	`CompletionState`	'qualified-draft'
`warning_card`	`dict`	Field(default_factory=dict)
`evidence`	`dict`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`maturity`	`str`	'simulation_only'
`prohibited_use`	`list[str]`	Field(default_factory=lambda: ['production_robotics', 'live_safety_critical'])
`permitted_use`	`list[str]`	Field(default_factory=lambda: ['simulation', 'pre_adapted_readings'])
`fused_value`	`dict`	Field(default_factory=dict)
`confidence`	`float`	0.0
`requires_review`	`bool`	False
`attention_weights`	`dict`	Field(default_factory=dict)
`active_queries`	`list[str]`	Field(default_factory=list)
`sources`	`list[dict]`	Field(default_factory=list)
`filtered_signal`	`list[float]`	Field(default_factory=list)
`noise_estimate`	`float`	0.0
`metadata`	`dict`	Field(default_factory=dict)
`state_estimate`	`list[float]`	Field(default_factory=list)
`covariance`	`list[list[float]]`	Field(default_factory=list)
`outliers_detected`	`list[dict]`	Field(default_factory=list)
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None
`agentic_evidence`	`dict`	Field(default_factory=dict)
`requires_human_review`	`bool`	False

`SensoryFusionBlock(AIBlock[SensoryFusionInput, SensoryFusionOutput, dict])`¶

Multi-modal sensory fusion block (16 ops; 15 functional + capability discovery).

Field	Type	Default
`name`	`str`	'sensory_fusion'
`state`	`dict`	field(default_factory=dict)
`resource_bounds`	`ResourceBounds`	field(default_factory=ResourceBounds)
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`planner`	`SensoryFusionFloorPlanner \\| None`	None

Methods:

`infer(data: SensoryFusionInput) -> Result[SensoryFusionOutput]`¶

`MCPSensoryFusionInput(BaseModel)`¶

Input to SensoryFusionMCPBlock â€” 30 ops.

Field	Type	Default
`op`	Literal['fuse', 'attend', 'query_sensor', 'assess_confidence', 'active_perception', 'add_sensor_source', 'remove_sensor_source', 'list_sources', 'get_noise_model', 'filter_signal', 'get_info', 'reset', 'fuse_ekf', 'reject_outliers', 'configure_ekf', 'store_percept', 'load_percept', 'list_percepts', 'store_fusion_config', 'load_fusion_config', 'list_fusion_configs', 'search_percepts', 'analyze_sensor_health', 'calibrate_attention', 'get_fusion_stats', 'export_percepts', 'compare_percepts', 'batch_fuse', 'list_patterns', 'source_capabilities']	required
`sensor_readings`	`list[dict]`	Field(default_factory=list)
`source_id`	`str`	''
`source_type`	`str`	''
`source_weight`	`float`	1.0
`goal`	`str`	''
`prior_observations`	`list[dict]`	Field(default_factory=list)
`signal`	`list[float]`	Field(default_factory=list)
`filter_type`	`str`	'lowpass'
`cutoff_freq`	`float`	0.1
`sample_rate`	`float`	1.0
`confidence_threshold`	`float`	0.5
`metadata`	`dict`	Field(default_factory=dict)
`state_vector`	`list[float]`	Field(default_factory=list)
`process_noise`	`float`	0.01
`measurement_noise`	`float`	0.1
`observation_model`	`list[list[float]]`	Field(default_factory=list)
`observation_vector`	`list[float]`	Field(default_factory=list)
`outlier_sigma`	`float`	3.0
`confidence_floor`	`float`	0.2
`run_mode`	`str`	'beta'
`reviewer_signature`	`str`	''
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`percept_name`	`str`	''
`config_name`	`str`	''
`query`	`str`	''
`top_k`	`int`	10
`percept_json`	`str`	''
`compare_id`	`str`	''
`batch_readings`	`list[list[dict]]`	Field(default_factory=list)
`acknowledge_review`	`bool`	False

`MCPSensoryFusionOutput(BaseModel)`¶

Output from SensoryFusionMCPBlock.

Field	Type	Default
`op`	`str`	required
`completion_state`	`CompletionState`	'qualified-draft'
`warning_card`	`dict`	Field(default_factory=dict)
`evidence`	`dict`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`fused_value`	`dict`	Field(default_factory=dict)
`confidence`	`float`	0.0
`attention_weights`	`dict`	Field(default_factory=dict)
`active_queries`	`list[str]`	Field(default_factory=list)
`sources`	`list[dict]`	Field(default_factory=list)
`filtered_signal`	`list[float]`	Field(default_factory=list)
`noise_estimate`	`float`	0.0
`metadata`	`dict`	Field(default_factory=dict)
`state_estimate`	`list[float]`	Field(default_factory=list)
`covariance`	`list[list[float]]`	Field(default_factory=list)
`outliers_detected`	`list[dict]`	Field(default_factory=list)
`message`	`str`	''
`count`	`int`	0
`retrieved`	`list[dict]`	Field(default_factory=list)
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None
`requires_review`	`bool`	False
`agentic_evidence`	`dict`	Field(default_factory=dict)
`requires_human_review`	`bool`	False

`SensoryFusionMCPBlock(AIBlock[MCPSensoryFusionInput, MCPSensoryFusionOutput, dict])`¶

30-op sensory fusion MCP block with SQLite persistence.

Field	Type	Default
`name`	`str`	'sensory_fusion_mcp'
`state`	`dict`	field(default_factory=dict)
`db_path`	`str`	field(default_factory=lambda: os.environ.get('SENSORY_FUSION_DB_PATH', _DEFAULT_DB))
`resource_bounds`	`ResourceBounds`	field(default_factory=ResourceBounds)
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`infer(data: MCPSensoryFusionInput) -> Result[MCPSensoryFusionOutput]`¶

`SensoryFusionStore`¶

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

Constructor:

Parameter	Type	Default
`db_path`	`str`	':memory:'

Methods:

`save_percept(name: str, percept: dict, confidence: float = 0.0, tags: str = '', requires_review: bool = False) -> str`¶

`load_percept(name: str) -> dict | None`¶

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

`search_percepts(query: str, limit: int = 10) -> list[dict]`¶

`export_percepts(limit: int = 100) -> list[dict]`¶

`save_fusion_config(name: str, config: dict, tags: str = '') -> str`¶

`load_fusion_config(name: str) -> dict | None`¶

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

`log_sensor_health(source_id: str, status: str = 'ok', metrics: dict | None = None) -> str`¶

`get_sensor_health(source_id: str) -> dict | None`¶

`log_attention(goal: str, weights: dict, source_count: int) -> str`¶

`log_filter(filter_type: str, params: dict, input_len: int, output_len: int) -> str`¶

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

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

MCP Tools¶

Operation	Source
`fuse`	sensory_fusion_mcp
`attend`	sensory_fusion_mcp
`query_sensor`	sensory_fusion_mcp
`assess_confidence`	sensory_fusion_mcp
`active_perception`	sensory_fusion_mcp
`add_sensor_source`	sensory_fusion_mcp
`remove_sensor_source`	sensory_fusion_mcp
`list_sources`	sensory_fusion_mcp
`get_noise_model`	sensory_fusion_mcp
`filter_signal`	sensory_fusion_mcp
`get_info`	sensory_fusion_mcp
`reset`	sensory_fusion_mcp
`fuse_ekf`	sensory_fusion_mcp
`reject_outliers`	sensory_fusion_mcp
`configure_ekf`	sensory_fusion_mcp
`store_percept`	sensory_fusion_mcp
`load_percept`	sensory_fusion_mcp
`list_percepts`	sensory_fusion_mcp
`store_fusion_config`	sensory_fusion_mcp
`load_fusion_config`	sensory_fusion_mcp
`list_fusion_configs`	sensory_fusion_mcp
`search_percepts`	sensory_fusion_mcp
`analyze_sensor_health`	sensory_fusion_mcp
`calibrate_attention`	sensory_fusion_mcp
`get_fusion_stats`	sensory_fusion_mcp
`export_percepts`	sensory_fusion_mcp
`compare_percepts`	sensory_fusion_mcp
`batch_fuse`	sensory_fusion_mcp
`list_patterns`	sensory_fusion_mcp
`source_capabilities`	sensory_fusion_mcp

Sensory Fusion¶

Overview¶

Public API¶

SensoryFusionInput(BaseModel)¶

SensoryFusionOutput(BaseModel)¶

SensoryFusionBlock(AIBlock[SensoryFusionInput, SensoryFusionOutput, dict])¶

infer(data: SensoryFusionInput) -> Result[SensoryFusionOutput]¶

MCPSensoryFusionInput(BaseModel)¶

MCPSensoryFusionOutput(BaseModel)¶

SensoryFusionMCPBlock(AIBlock[MCPSensoryFusionInput, MCPSensoryFusionOutput, dict])¶

infer(data: MCPSensoryFusionInput) -> Result[MCPSensoryFusionOutput]¶

SensoryFusionStore¶

save_percept(name: str, percept: dict, confidence: float = 0.0, tags: str = '', requires_review: bool = False) -> str¶

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

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

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

export_percepts(limit: int = 100) -> list[dict]¶

save_fusion_config(name: str, config: dict, tags: str = '') -> str¶

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

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

log_sensor_health(source_id: str, status: str = 'ok', metrics: dict | None = None) -> str¶

get_sensor_health(source_id: str) -> dict | None¶

log_attention(goal: str, weights: dict, source_count: int) -> str¶

log_filter(filter_type: str, params: dict, input_len: int, output_len: int) -> str¶

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

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

MCP Tools¶

`SensoryFusionInput(BaseModel)`¶

`SensoryFusionOutput(BaseModel)`¶

`SensoryFusionBlock(AIBlock[SensoryFusionInput, SensoryFusionOutput, dict])`¶

`infer(data: SensoryFusionInput) -> Result[SensoryFusionOutput]`¶

`MCPSensoryFusionInput(BaseModel)`¶

`MCPSensoryFusionOutput(BaseModel)`¶

`SensoryFusionMCPBlock(AIBlock[MCPSensoryFusionInput, MCPSensoryFusionOutput, dict])`¶

`infer(data: MCPSensoryFusionInput) -> Result[MCPSensoryFusionOutput]`¶

`SensoryFusionStore`¶

`save_percept(name: str, percept: dict, confidence: float = 0.0, tags: str = '', requires_review: bool = False) -> str`¶

`load_percept(name: str) -> dict | None`¶

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

`search_percepts(query: str, limit: int = 10) -> list[dict]`¶

`export_percepts(limit: int = 100) -> list[dict]`¶

`save_fusion_config(name: str, config: dict, tags: str = '') -> str`¶

`load_fusion_config(name: str) -> dict | None`¶

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

`log_sensor_health(source_id: str, status: str = 'ok', metrics: dict | None = None) -> str`¶

`get_sensor_health(source_id: str) -> dict | None`¶

`log_attention(goal: str, weights: dict, source_count: int) -> str`¶

`log_filter(filter_type: str, params: dict, input_len: int, output_len: int) -> str`¶

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

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