INTENT Pattern Library Architecture

2.4 INTENT Pattern Library Architecture

While the core Spaxiom DSL provides low-level primitives for sensor fusion and temporal logic, the INTENT (Intelligent Network for Temporal & Embodied Neuro-symbolic Tasks) layer provides domain-specific abstractions that package common spatiotemporal patterns into reusable, composable, and agent-ready components.

INTENT patterns sit between raw sensors and agents, acting as a semantic middleware that translates sensor streams into high-level behavioral events. This section describes the architecture, interface contracts, and extensibility mechanisms of the INTENT pattern library.

Pattern interface and lifecycle

All INTENT patterns implement a common Pattern base interface:

from abc import ABC, abstractmethod
from typing import List, Dict, Any
from spaxiom import Sensor, Condition

class Pattern(ABC):
    """Base interface for all INTENT patterns."""

    def __init__(self, name: str):
        self.name = name
        self._sensors: List[Sensor] = []
        self._conditions: Dict[str, Condition] = {}
        self._state: Dict[str, Any] = {}

    @abstractmethod
    def attach(self, sensors: List[Sensor]) -> None:
        """Attach pattern to sensor sources."""
        pass

    @abstractmethod
    def update(self, dt: float) -> None:
        """Update pattern state based on elapsed time dt."""
        pass

    @abstractmethod
    def emit_events(self) -> List[Dict[str, Any]]:
        """Emit structured events based on current pattern state."""
        pass

    def conditions(self) -> Dict[str, Condition]:
        """Return dictionary of named conditions for this pattern."""
        return self._conditions

    def state_dict(self) -> Dict[str, Any]:
        """Return serializable state for persistence/debugging."""
        return self._state

Key lifecycle methods:

• attach(sensors): called once during initialization to bind the pattern to specific sensor instances. The pattern registers callbacks and builds internal data structures.
• update(dt): called by the runtime on each tick with elapsed time dt (seconds). The pattern updates its internal state based on sensor readings.
• emit_events(): called after update() to generate zero or more structured event dictionaries. Events are JSON-serializable and follow a consistent schema.
• conditions(): returns a dictionary of named Condition objects that other patterns or agents can subscribe to. This enables pattern composition.

Built-in patterns

The INTENT library ships with several production-ready patterns:

OccupancyField

Purpose: spatial occupancy and crowding analysis over floor grids.

Sensors: floor pressure grid, depth cameras, or occupancy sensors.

Conditions:

• percent_above(threshold): returns Condition true when ≥ threshold% of tiles are active.
• hotspot_in(zone): crowding hotspot detected in specific zone.
• density_exceeds(people_per_sqm): density above safety threshold.

Events emitted: CrowdFormation, HotspotDetected, DensityExceeded.

State: 2D occupancy heatmap, hotspot locations, historical density.

from spaxiom.intent import OccupancyField

field = OccupancyField(floor_sensor, name="lobby", resolution=0.5)
field.attach([floor_sensor])

# Define condition: crowded for 3 minutes
crowded = within(180.0, field.percent_above(10.0))

@on(crowded)
def handle_crowding():
    events = field.emit_events()
    for event in events:
        if event["type"] == "HotspotDetected":
            print(f"Hotspot at {event['zone']}: {event['density']:.1f} ppl/m²")

QueueFlow

Purpose: queue length estimation, arrival/service rate tracking, wait time prediction.

Sensors: occupancy grid at queue entrance/exit, depth cameras, or entry/exit beam sensors.

Conditions:

• queue_length_exceeds(n): more than n people waiting.
• wait_time_exceeds(seconds): estimated wait time above threshold.
• service_stalled(): no service events for prolonged period.

Events emitted: QueueLengthChanged, WaitTimeExceeded, ServiceStalled.

State: queue length L(t), arrival rate λ(t), service rate μ(t), estimated wait time W(t) ≈ L / μ.

from spaxiom.intent import QueueFlow

queue = QueueFlow(
    entry_zone=checkout_entry,
    exit_zone=checkout_exit,
    name="checkout_queue"
)

long_wait = queue.wait_time_exceeds(300)  # > 5 minutes

@on(long_wait)
async def alert_manager():
    state = queue.state_dict()
    print(f"Queue length: {state['length']}, Wait time: {state['wait_time_s']:.0f}s")

ADLTracker

Purpose: activities of daily living (ADL) tracking for elder care, rehabilitation, or hospital monitoring.

Sensors: multi-zone occupancy, bed pressure, bathroom door, kitchen sensors, wearable accelerometers.

Conditions:

• got_up(): person transitioned from bed to standing.
• in_bathroom(duration): bathroom occupancy for duration seconds.
• meal_activity(): kitchen activity pattern consistent with meal preparation.
• no_activity(hours): no detected activity for hours (potential fall or medical event).

Events emitted: WokeUp, Meal, BathroomVisit, NoActivityAlert.

FmSteward

Purpose: facilities management "needs service" aggregator for restrooms, conference rooms, or public spaces.

Sensors: occupancy, usage counters, air quality, supply level sensors (soap, paper towels).

Conditions:

• needs_service(): composite condition based on usage thresholds, time since last clean, or air quality.
• supplies_low(): consumables below threshold.

Events emitted: ServiceNeeded, SuppliesLow, ServiceCompleted.

Pattern composition and dependencies

Patterns can depend on other patterns, enabling hierarchical abstraction. Example: a SmartBuildingAgent pattern might aggregate OccupancyField, QueueFlow, and energy sensor data:

from spaxiom.intent import Pattern, OccupancyField, QueueFlow
from spaxiom import Sensor, Condition

class SmartBuildingAgent(Pattern):
    def __init__(self, name: str):
        super().__init__(name)
        self.occupancy = OccupancyField(name=f"{name}_occupancy")
        self.queue = QueueFlow(name=f"{name}_queue")
        self.energy_sensor = None

    def attach(self, sensors: List[Sensor]) -> None:
        floor_sensors = [s for s in sensors if s.type == "floor"]
        self.occupancy.attach(floor_sensors)

        # Queue uses subset of occupancy zones
        self.queue.attach(self.occupancy.zones["entrance"])

        self.energy_sensor = next(s for s in sensors if s.name == "building_power")

    def update(self, dt: float) -> None:
        self.occupancy.update(dt)
        self.queue.update(dt)

        # Composite logic: adjust HVAC based on occupancy and queue
        occupancy_pct = self.occupancy.percent()
        queue_length = self.queue.state_dict()["length"]
        power_kw = self.energy_sensor.read().value

        self._state["comfort_score"] = self._compute_comfort(occupancy_pct, queue_length)
        self._state["efficiency_score"] = self._compute_efficiency(power_kw, occupancy_pct)

    def emit_events(self) -> List[Dict[str, Any]]:
        events = []
        if self._state["comfort_score"] < 0.5:
            events.append({
                "type": "ComfortDegradation",
                "zone": self.name,
                "score": self._state["comfort_score"],
                "timestamp": time.time()
            })
        return events

This composition enables agents to reason at multiple levels of abstraction without re-implementing low-level sensor fusion.

Creating custom patterns

Users can define domain-specific patterns by subclassing Pattern. Example: a custom ConferenceRoomUtilization pattern:

from spaxiom.intent import Pattern
from spaxiom import Sensor, Condition, within

class ConferenceRoomUtilization(Pattern):
    def __init__(self, room_name: str, capacity: int):
        super().__init__(name=room_name)
        self.capacity = capacity
        self._occupancy_sensor = None
        self._door_sensor = None
        self._meeting_start_time = None

    def attach(self, sensors: List[Sensor]) -> None:
        self._occupancy_sensor = next(s for s in sensors if s.type == "occupancy")
        self._door_sensor = next(s for s in sensors if s.type == "door")

        # Define conditions
        occupied = Condition(lambda: self._occupancy_sensor.read() > 0)
        self._conditions["meeting_in_progress"] = within(60.0, occupied)
        self._conditions["over_capacity"] = Condition(
            lambda: self._occupancy_sensor.read() > self.capacity
        )

    def update(self, dt: float) -> None:
        occupancy = self._occupancy_sensor.read()
        door_open = self._door_sensor.read() > 0.5

        # Track meeting start/end
        if self._conditions["meeting_in_progress"].holds():
            if self._meeting_start_time is None:
                self._meeting_start_time = time.time()
        else:
            self._meeting_start_time = None

        # Update utilization stats
        if self._meeting_start_time:
            duration = time.time() - self._meeting_start_time
            self._state["current_meeting_duration"] = duration
            self._state["utilization_ratio"] = occupancy / self.capacity

    def emit_events(self) -> List[Dict[str, Any]]:
        events = []
        if self._conditions["over_capacity"].holds():
            events.append({
                "type": "RoomOverCapacity",
                "room": self.name,
                "occupancy": self._occupancy_sensor.read(),
                "capacity": self.capacity,
                "timestamp": time.time()
            })
        return events

Custom patterns integrate seamlessly with the runtime and can be composed with built-in patterns.

Pattern state management and persistence

Patterns maintain internal state that may need to persist across restarts or be checkpointed for debugging. The state_dict() method returns a JSON-serializable snapshot:

# Checkpoint pattern state
state = occupancy_field.state_dict()
with open("occupancy_checkpoint.json", "w") as f:
    json.dump(state, f)

# Restore pattern state
with open("occupancy_checkpoint.json", "r") as f:
    state = json.load(f)
occupancy_field.load_state(state)

For production deployments, state can be persisted to Redis, PostgreSQL, or object storage, enabling fault tolerance and multi-instance coordination.

Performance and scalability

Pattern performance characteristics:

• Memory footprint: patterns with spatial state (e.g., OccupancyField) scale O(N) with number of grid cells. Typical floor grid: 50×50 = 2500 cells ≈ 10 KB state.
• Computation: most patterns are O(1) or O(N) per update tick. QueueFlow uses exponential moving averages for arrival/service rates, avoiding expensive window calculations.
• Event throughput: patterns emit events only on state transitions, not on every tick. Typical event rate: 0.1–10 events/second depending on domain.
• Sensor fan-in: patterns can efficiently aggregate 10s–100s of sensors through internal spatial indexing (quad-trees for 2D grids, octrees for 3D).

Benchmarks on Raspberry Pi 4 (4 GB RAM, ARMv8):

• OccupancyField with 2500 sensors: 5 ms update latency at 10 Hz
• QueueFlow with 20 entry/exit sensors: 0.2 ms update latency
• ADLTracker with 15 zone sensors: 0.5 ms update latency

These latencies are well within real-time requirements for embodied agents and building automation systems.

Pattern library extensibility

The INTENT library is designed for community contributions. Future patterns under development include:

• TrafficFlow: vehicle or pedestrian flow analysis with direction, speed, and congestion detection
• AnomalyDetector: unsupervised anomaly detection using autoencoders over sensor embeddings
• EnergyOptimizer: multi-objective optimization pattern for HVAC, lighting, and load balancing
• SafetyEnvelope: runtime safety monitoring for human-robot collaboration (see Section 6)

Developers can publish patterns to a registry (similar to PyPI) and import them with:

from spaxiom.intent.registry import install_pattern

install_pattern("acme-corp/warehouse-traffic")
from spaxiom.intent.warehouse_traffic import ForkLiftSafety

This extensibility model enables Spaxiom to grow with the community while maintaining a stable core API.