Spaxiom: A Sensor Cortex and Intent Layer for the Era of Experience

1. Introduction: From Sensors to Experience

1.1 The looming experience deluge

The physical world is rapidly becoming a dense sensor mesh. IoT analyses report on the order of tens of billions of connected devices in 2024-2025, with projections ranging from ~39 billion to over 70 billion devices by 2030-2035 depending on methodology and scope. At the same time, context-aware computing (systems that adapt to who you are, where you are, and what is happening) is evolving from a niche into a mainstream infrastructure category, with forecasts showing high double-digit CAGR through the early 2030s.

The result is a looming experience deluge:

• Dense arrays of floor pressure sensors, thermal cameras, radar, time-of-flight depth sensors, air quality probes;
• Continuous event streams from robots, vehicles, wearables, and industrial systems.

Yet much of this data is currently:

• consumed by bespoke scripts and brittle pipelines,
• discarded for lack of storage or interpretability, or
• never surfaced in a form that agents can reason about.

1.2 The Era of Experience

Silver and Sutton's "Welcome to the Era of Experience" describes a shift from training AI on human-curated datasets (text, images, labels) toward agents that learn predominantly from their own experience: ongoing interaction with the world, collecting trajectories of observations, actions, and rewards.

They argue that:

• these agents will learn open-ended skills, not just supervised tasks;
• experience will be continuous, multimodal, and embodied;
• the bottleneck will move from "Do we have enough labeled data?" to "Can we transform raw experience into useful structure for learning and decision-making?".

Most of the work in that direction has focused on simulated environments or camera-centric setups. But the world is full of other sensors: pressure, contact, force, vibration, chemical, thermal, RF, occupancy, and more. Many of these are privacy-preserving and cheap at scale, making them ideal substrates for real-world experience.

What is missing is a way to:

• declare what those sensors mean in space and time,
• fuse them into semantically rich events, and
• expose those events in a compressed but expressive form to learning systems and agents.

This is where Spaxiom slots in.

2. Spaxiom Today: Architectural Overview

Spaxiom is currently implemented as an embedded DSL in Python, backed by a runtime and a growing documentation set. Concretely, Spaxiom provides:

• Spatial primitives: Zone, coordinate systems, grid representations.
• Sensors & adaptors: base Sensor abstraction plus adaptors for floor grids, MQTT streams, GPIO, and simulated sensors.
• Temporal logic: Condition, within, always, eventually, etc.
• Event dispatch: @on(condition) decorators for registering callbacks.
• Entities & sets: tracking collections of objects with attributes.
• Physical quantities: Quantity with unit support.

2.1 Minimal example

from spaxiom import Sensor, Zone, Condition, on, within

office_zone   = Zone(0, 0, 10, 10)
motion_sensor = Sensor("motion1", "motion", (5, 5, 0))

motion_detected   = Condition(lambda: motion_sensor.read() > 0.5)
sustained_motion  = within(5.0, motion_detected)

@on(sustained_motion)
def alert_sustained_motion():
    print("Motion has been detected for 5 seconds!")

Instead of pushing raw time series into downstream systems, the user expresses what they care about (e.g., "sustained motion in this region for >5 seconds"), and Spaxiom takes responsibility for timing, buffering, and triggering.

2.2 INTENT: Agent-ready patterns

On top of the core DSL, we define INTENT (Intelligent Network for Temporal & Embodied Neuro-symbolic Tasks), a pattern library that captures common behaviors over sensors.

INTENT sits between the low-level DSL primitives (sensors, conditions, zones from Section 2.1) and high-level agent reasoning. While the DSL provides the building blocks for expressing spatiotemporal logic, INTENT patterns encapsulate recurring behaviors as reusable abstractions. For example, detecting "a queue is forming" involves sensor fusion (camera + pressure mat), temporal filtering (queue must persist >30 seconds), and queueing theory (estimating wait times from arrival/service rates). Rather than re-implementing this logic at each deployment, an INTENT.QueueFlow pattern captures it once and exposes an agent-ready event stream.

INTENT patterns serve three critical purposes:

1. Semantic compression: Reducing hundreds of sensor readings per second to a single OccupancyChanged or QueueFormed event saves 100-1000× tokens when feeding context to LLMs (Section 3).
2. Domain expertise: Patterns embed specialized knowledge (queueing theory for QueueFlow, ADL recognition heuristics for elder care, facilities management logic for FmSteward) that would otherwise need to be re-discovered at each site.
3. Agent-readiness: Patterns emit events with agent-friendly schemas (JSON-serializable, self-describing, typed) optimized for inclusion in LLM context windows or as observations in RL environments.

Core INTENT patterns

The INTENT library includes a growing collection of domain-specific patterns. Four foundational patterns demonstrate the breadth of applicability:

• OccupancyField: Tracks spatial occupancy and crowding over floor grids using pressure mats, depth cameras, or thermal sensors. Provides hotspot detection (top-k busiest tiles), density heatmaps (% occupied over time), and flow analysis (entries/exits per zone). Typical applications: lobby management, retail heat-mapping, emergency evacuation planning, stadium crowd control.
• QueueFlow: Models queue dynamics with real-time arrival and service rate estimation. Uses M/M/k queueing theory to predict wait times, detect bottlenecks, and trigger staffing alerts when queues exceed thresholds. Emits events like QueueFormed, QueueLengthChanged, and LongWaitDetected (when predicted wait > threshold). Typical applications: checkout optimization, airport security lines, call center staffing, hospital triage.
• ADLTracker: Recognizes activities of daily living (ADLs) from multi-modal sensors: floor pressure (walking, standing, lying down), door sensors (room transitions), appliance usage (refrigerator, stove, bathroom fixtures). Detects anomalies in routine (missed meals, unusual sleep patterns, prolonged inactivity, repeated bathroom visits) and emits alerts for caregiver intervention. Typical applications: elder care monitoring (aging-in-place), hospital patient observation, rehabilitation progress tracking, wellness scoring.
• FmSteward: Aggregates facilities maintenance signals into "needs service" events. Monitors usage counters (toilet flushes, paper towel dispensers, occupancy duration), environmental conditions (air quality, humidity, temperature), and time-since-last-service to trigger maintenance requests. Emits NeedsService(zone="restroom_3", reason="high_usage", urgency="medium"). Typical applications: restroom cleanliness monitoring, supply restocking, HVAC filter replacement, proactive janitorial dispatch.

Additional domain-specific patterns (detailed in appendix use cases and Section 2.4) include:

• SafetyMonitor - robot collision avoidance (Section 7.3)
• ConferenceRoomUtilization - meeting room booking efficiency
• SmartBuildingAgent - multi-pattern HVAC orchestration
• ContaminationMonitor - cleanroom particle/pressure tracking (Appendix A.1)
• MachineryHealthMonitor - vibration/acoustic anomaly detection (A.2)
• IAQRiskMonitor - indoor air quality and ventilation (A.3)
• MicroMobilitySafety - near-miss detection via radar (A.4)
• ColdChainMonitor - pharmaceutical temperature/humidity integrity (A.5)
• WildfireRiskMonitor - multi-sensor fire danger index (A.6)
• DataCenterThermalOptimizer - PUE and cooling efficiency (A.7)
• IrrigationOptimizer - crop water stress and soil moisture (A.8)
• ORSterilityMonitor - operating room sterility assurance (A.9)
• HumanoidTaskCoordinator - robot context streaming (A.10)
• CategoryAggregator - retail/expo engagement (Section 11.1)
• TrafficFlowMonitor - logistics corridor throughput (Section 11.3)
• QsrFlowOptimizer - quick-service restaurant throughput (Section 11.5)

This diversity demonstrates that INTENT is not a fixed set of patterns, but an extensible framework for capturing domain expertise across industries: from healthcare to manufacturing, retail to agriculture, smart buildings to autonomous systems.

Example: OccupancyField → LLM integration

The following code demonstrates how an OccupancyField pattern compresses floor sensor data into crowding metrics and feeds them to an LLM agent for decision-making:

from spaxiom.config  import load_yaml
from spaxiom.runtime import start_runtime
from spaxiom.temporal import within
from spaxiom.logic    import on
from spaxiom.intent   import OccupancyField

import asyncio, json, os, openai

sensors = load_yaml("examples/lobby.yaml")
floor   = sensors["lobby_floor"]

field   = OccupancyField(floor, name="lobby")
crowded = within(180.0, field.percent_above(10.0))   # >= 10% tiles active, 3 minutes

@on(crowded)
async def lobby_agent():
    facts = {
        "zone": field.name,
        "occupancy_pct": field.percent(),
        "hotspots": field.hotspots(top_k=3),
    }
    prompt = (
        "You are a smart-building lobby agent. "
        "Given this JSON describing current crowding, "
        "suggest 1--3 actions to improve flow and experience. "
        "Respond as JSON.\n"
        + json.dumps(facts)
    )
    openai.api_key = os.getenv("OPENAI_API_KEY", "")
    if not openai.api_key:
        return

    rsp = await openai.ChatCompletion.acreate(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
    )
    print("INTENT actions:", rsp.choices[0].message.content)

if __name__ == "__main__":
    asyncio.run(start_runtime())

In this example, the OccupancyField pattern continuously monitors a floor sensor grid. When crowding persists for 3 minutes (within(180.0, ...)), the pattern's percent() and hotspots() methods provide agent-ready summaries: no raw sensor arrays, no manual aggregation. The LLM receives a compact JSON payload (~200 bytes) instead of 100s of kilobytes of raw sensor timeseries.

Extensibility: Users can define custom patterns by subclassing the Pattern base class (detailed in Section 2.4). This allows organizations to capture domain-specific knowledge (manufacturing process signatures, hospital workflow patterns, retail merchandising behaviors) as reusable INTENT abstractions that can be versioned, tested, and shared across deployments.

This pattern (sensor → DSL → INTENT → agent) is central to the rest of the paper. Section 2.4 provides the full pattern interface architecture, lifecycle methods, performance benchmarks, and guidelines for creating custom patterns.

2.3 Type System and Composition

Spaxiom's expressiveness comes from a small set of composable primitives with a well-defined type system. Unlike ad-hoc sensor integration scripts, Spaxiom enforces type safety at DSL construction time and provides algebraic composition operators that enable complex spatiotemporal queries to be built from simple building blocks.

Core type hierarchy

The DSL defines the following fundamental types:

• Sensor: base abstraction for any physical or virtual sensor. Key methods:
  • read() → Value: synchronous read of current sensor state
  • position → (x, y, z): 3D location in space
  • zone → Zone: containing spatial region
  • units → Unit: physical units (e.g., meters, celsius, pascal)
• Condition: boolean-valued predicate over sensor state and time. Conditions are lazy: they are not evaluated until subscribed or explicitly triggered. Key properties:
  • holds() → bool: evaluate condition at current time
  • since_true → float: seconds since condition became true (or 0.0 if false)
  • since_false → float: seconds since condition became false (or 0.0 if true)
• Zone: 3D spatial region with geometric operations. Supports axis-aligned boxes, convex polygons, and union/intersection:
  • contains(x, y, z) → bool
  • overlaps(other_zone) → bool
  • distance_to(x, y, z) → float
  • volume → float
• Entity: tracked object with identity and attributes. Examples: a person, robot, forklift, or pallet. Entities have:
  • id → str: unique identifier
  • position → (x, y, z): current location (if trackable)
  • velocity → (vx, vy, vz): optional velocity vector
  • attributes → dict: arbitrary key-value metadata
• Quantity: numeric value with physical units. Enforces dimensional analysis:
  • value → float
  • unit → Unit
  • Supports arithmetic with automatic unit conversion and error on incompatible units

Condition composition operators

Conditions form a boolean algebra with standard operators:

from spaxiom import Condition

# Logical operators
c1 = Condition(lambda: temp.read() > 25.0)
c2 = Condition(lambda: humidity.read() > 60.0)

hot_and_humid = c1 & c2    # conjunction (AND)
hot_or_humid  = c1 | c2    # disjunction (OR)
not_hot       = ~c1        # negation (NOT)

# Temporal chaining
humid_then_hot = c2.before(c1, within_seconds=300)  # c2 before c1, within 5 min

Importantly, these operators are not evaluated eagerly. They construct a lazy evaluation graph that the runtime optimizes.

Temporal operators

Spaxiom provides temporal logic operators inspired by Linear Temporal Logic (LTL) but adapted for continuous real-time systems:

• within(duration, condition): condition holds continuously for at least duration seconds.

  sustained_motion = within(5.0, motion_detected)  # motion for ≥5s

• always(duration, condition): condition holds at all sampled times over the past duration seconds (similar to within but checks discrete samples).

  stable_temp = always(60.0, temp_in_range)  # stable for 1 minute

• eventually(duration, condition): condition becomes true at least once within the next duration seconds (forward-looking, requires buffering or prediction).

  # Predictive: will door open in next 10s?
  door_will_open = eventually(10.0, door_sensor_active)

• before(c1, c2, within_seconds=T): condition c1 occurs before c2 within time window T.

  alarm_then_evacuation = alarm_triggered.before(exit_door_opened, within_seconds=120)

• after(c1, c2, within_seconds=T): condition c1 occurs after c2 within time window T.

These operators enable expressive temporal patterns without manually managing timers or state machines.

Spatial operators

Spatial queries are first-class in Spaxiom. Common patterns:

• inside(entity, zone): returns a Condition that is true when entity is within zone.

  robot_in_hazard_zone = inside(robot_entity, hazard_zone)

• near(entity1, entity2, threshold): true when Euclidean distance < threshold.

  collision_risk = near(robot, human, threshold=2.0)  # within 2 meters

• overlaps(zone1, zone2): true if zones intersect.
• distance(entity, zone): returns Quantity in meters.

  dist_to_exit = distance(person, exit_zone)
  far_from_exit = Condition(lambda: dist_to_exit.value > 10.0)

Type checking and unit validation

Spaxiom performs compile-time validation where possible:

• Unit compatibility: operations on Quantity objects check dimensional compatibility. For example:

  from spaxiom.units import meters, seconds, kg
  
  distance = 10.0 * meters
  time = 5.0 * seconds
  velocity = distance / time  # OK: m/s
  
  mass = 70 * kg
  invalid = distance + mass   # TypeError: incompatible units (length + mass)

• Condition type safety: logical operators (&, |, ~) only accept Condition objects, preventing accidental confusion with Python's native boolean operators.
• Spatial type checking: Zone methods enforce 3D coordinate tuples or Entity objects with positions.

Runtime validation handles cases that cannot be checked statically:

• Sensor read failures (timeouts, disconnections) raise SensorException with configurable retry/fallback policies.
• Out-of-bounds spatial queries (e.g., entity position outside known coordinate system) trigger warnings or exceptions.
• Temporal operator timeouts (e.g., eventually failing to observe condition) can fire explicit callbacks.

Composition example: complex spatiotemporal query

Combining these primitives enables concise expression of complex behaviors. Example: detect a "loitering near exit" pattern.

from spaxiom import Sensor, Zone, Entity, Condition, within, inside, near
from spaxiom.units import meters

# Spatial setup
exit_zone = Zone(x=10, y=20, width=3, height=2)
exit_vicinity = exit_zone.buffer(5 * meters)  # 5m buffer around exit

# Entity tracking
person = Entity(id="person_42")

# Conditions
near_exit = inside(person, exit_vicinity)
not_exiting = ~inside(person, exit_zone)
stationary = Condition(lambda: person.velocity.magnitude() < 0.1)  # < 0.1 m/s

# Composite pattern: near exit, not moving, for 30+ seconds
loitering = within(30.0, near_exit & not_exiting & stationary)

@on(loitering)
def alert_security():
    print(f"Person {person.id} loitering near exit at {person.position}")

This single composite condition would require dozens of lines of manual state tracking, timers, and geometric computations in a traditional imperative approach.

Formal semantics and denotational interpretation

For verification purposes (see Section 7.3), we can give Spaxiom conditions a denotational semantics as functions from time to truth values:

Where ⟦c⟧(t) is the truth value of condition c at time t. Composition operators have natural interpretations:

• ⟦c₁ & c₂⟧(t) = ⟦c₁⟧(t) ∧ ⟦c₂⟧(t)
• ⟦c₁ | c₂⟧(t) = ⟦c₁⟧(t) ∨ ⟦c₂⟧(t)
• ⟦~c⟧(t) = ¬⟦c⟧(t)
• ⟦within(δ, c)⟧(t) = true iff ∀t' ∈ [t−δ, t], ⟦c⟧(t') = true

This formal interpretation enables model checking and equivalence proofs between Spaxiom programs and temporal logic formulas.

Performance characteristics of composition

Lazy evaluation and operator fusion are critical for performance:

• Lazy evaluation: Condition objects are not evaluated until subscribed via @on() or explicitly polled. This avoids wasted computation on conditions that are never used.
• Operator fusion: the runtime analyzes the condition graph and fuses compatible operators. For example, within(5.0, c1) & within(5.0, c2) shares a single 5-second time buffer rather than maintaining two separate buffers.
• Incremental evaluation: temporal operators like within use circular buffers and incremental updates, avoiding full re-evaluation on every sensor tick.
• Short-circuit evaluation: boolean operators (&, |) short-circuit when possible, skipping expensive sensor reads if the result is already determined.

These optimizations make Spaxiom practical even on resource-constrained edge devices with hundreds of sensors and conditions.

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.

2.5 Runtime Architecture and Execution Model

The Spaxiom runtime is responsible for orchestrating sensor reads, evaluating conditions, triggering callbacks, and managing pattern lifecycle. It must handle concurrency, failure modes, and resource constraints while maintaining deterministic behavior for safety-critical applications.

This section describes the runtime's architecture, execution model, and deployment strategies for both edge and cloud environments.

Event loop and scheduling

The runtime uses an async event loop built on Python's asyncio library, enabling efficient concurrent I/O for sensor polling, network communication, and callback execution.

Core event loop structure:

import asyncio
from typing import List, Dict
from spaxiom import Sensor, Condition, Pattern

class SpaxiomRuntime:
    def __init__(self, tick_rate: float = 10.0):
        self.tick_rate = tick_rate  # Hz
        self.tick_period = 1.0 / tick_rate  # seconds
        self.sensors: List[Sensor] = []
        self.conditions: Dict[str, Condition] = {}
        self.patterns: List[Pattern] = []
        self.callbacks: Dict[Condition, List[Callable]] = {}
        self._running = False

    async def run(self):
        """Main event loop."""
        self._running = True
        last_tick = asyncio.get_event_loop().time()

        while self._running:
            current_time = asyncio.get_event_loop().time()
            dt = current_time - last_tick

            # Phase 1: Sensor reads (concurrent)
            await self._read_sensors()

            # Phase 2: Pattern updates (sequential, in dependency order)
            await self._update_patterns(dt)

            # Phase 3: Condition evaluation
            await self._evaluate_conditions()

            # Phase 4: Callback dispatch (concurrent, isolated)
            await self._dispatch_callbacks()

            # Sleep until next tick
            last_tick = current_time
            sleep_time = max(0, self.tick_period - (asyncio.get_event_loop().time() - current_time))
            await asyncio.sleep(sleep_time)

    async def _read_sensors(self):
        """Concurrent sensor reads with timeout."""
        tasks = [sensor.async_read() for sensor in self.sensors]
        await asyncio.gather(*tasks, return_exceptions=True)

    async def _update_patterns(self, dt: float):
        """Update patterns in topological order."""
        for pattern in self._topological_sort(self.patterns):
            pattern.update(dt)

    async def _evaluate_conditions(self):
        """Evaluate all registered conditions."""
        for name, condition in self.conditions.items():
            condition._evaluate()  # Internal state update

    async def _dispatch_callbacks(self):
        """Dispatch callbacks for triggered conditions."""
        tasks = []
        for condition, callbacks in self.callbacks.items():
            if condition.just_became_true():  # Edge-triggered
                for callback in callbacks:
                    tasks.append(self._safe_callback(callback))
        await asyncio.gather(*tasks, return_exceptions=True)

    async def _safe_callback(self, callback: Callable):
        """Execute callback with exception isolation."""
        try:
            if asyncio.iscoroutinefunction(callback):
                await callback()
            else:
                callback()
        except Exception as e:
            self._log_error(f"Callback {callback.__name__} failed: {e}")

Key design decisions:

• Fixed tick rate: the runtime uses a configurable tick rate (default 10 Hz) for deterministic timing. This ensures temporal operators like within() have consistent semantics.
• Phased execution: each tick has four phases (sensor read, pattern update, condition evaluation, callback dispatch), executed in strict order to prevent race conditions.
• Concurrent I/O: sensor reads and callbacks run concurrently using asyncio.gather(), maximizing throughput on I/O-bound workloads.
• Exception isolation: callback failures do not crash the runtime. Exceptions are logged and the next tick continues normally.

Sensor buffering and sampling strategies

Sensors may produce data at rates different from the runtime tick rate. The runtime provides several buffering strategies:

• Latest value (default): sensor.read() returns the most recent value. Fast sensors (>100 Hz) are downsampled; slow sensors (<1 Hz) return stale values with a staleness timestamp.
• Moving window: sensor.window(duration) returns all values from the past duration seconds, stored in a circular buffer. Used for statistical aggregation (mean, variance, percentiles).
• Event-driven: sensors can push updates to the runtime asynchronously (e.g., MQTT or websocket streams). The runtime queues these updates and processes them in the next tick.

from spaxiom import Sensor
from spaxiom.units import celsius

temp_sensor = Sensor("room_temp", units=celsius)

# Latest value
current_temp = temp_sensor.read()  # → Quantity(22.5, celsius)

# Moving window (last 60 seconds)
window = temp_sensor.window(60.0)
mean_temp = sum(window) / len(window)
max_temp = max(window)

Condition evaluation: polling vs event-driven

Conditions can be evaluated in two modes:

1. Polling (default): conditions are evaluated on every tick. Suitable for conditions that depend on slowly changing sensor state or complex temporal logic.
2. Event-driven: conditions register callbacks on sensor value changes. The runtime triggers evaluation only when dependencies change. This is more efficient for sparse events (e.g., door open/close) but requires careful dependency tracking.

The runtime automatically selects the appropriate mode based on condition complexity. Users can override with:

condition = Condition(lambda: door.read() > 0.5, mode="event-driven")

Callback execution model: async/sync and threading

Callbacks registered with @on(condition) can be either synchronous or asynchronous:

@on(high_temp_alarm)
def sync_callback():
    """Runs in event loop thread, should be fast (<10 ms)."""
    print("Temperature alarm!")

@on(high_temp_alarm)
async def async_callback():
    """Can perform async I/O (network requests, database writes)."""
    await send_alert_email()

@on(high_temp_alarm)
@threaded  # Decorator for CPU-bound work
def cpu_intensive_callback():
    """Runs in thread pool, does not block event loop."""
    run_expensive_simulation()

Callback execution guarantees:

• Edge-triggered: callbacks fire once when a condition transitions from false to true, not on every tick while the condition holds.
• Ordering: callbacks for the same condition execute in registration order. Callbacks for different conditions may execute concurrently.
• Timeout: callbacks have a configurable timeout (default 30s). Long-running callbacks that exceed timeout are logged as warnings but allowed to complete.
• Retry: failed callbacks can optionally be retried with exponential backoff (useful for network-dependent actions).

Memory management for time-windowed operations

Temporal operators like within(duration, condition) maintain circular buffers of recent condition evaluations. Memory usage is O(duration × tick_rate).

For example, within(300.0, c) at 10 Hz requires 3000 boolean samples ≈ 3 KB. To prevent unbounded memory growth:

• Circular buffers: fixed-size buffers with O(1) append and O(1) query (using sliding window algorithm).
• Shared buffers: the runtime fuses conditions with identical time windows. within(60.0, c1) & within(60.0, c2) shares a single 60s buffer.
• Adaptive downsampling: for very long windows (>1 hour), the runtime automatically downsamples to 1 Hz or less, trading temporal precision for memory.

Handling sensor failures and missing data

Production sensor networks experience failures: disconnections, timeouts, corrupted data, hardware faults. The runtime provides several failure handling modes:

• Retry with backoff: sensor reads retry up to N times with exponential backoff (default N=3, max delay 5s).
• Fallback value: sensors can specify a fallback value (e.g., last known good, or a safe default like 0.0).
• Exception propagation: critical sensors can be marked required=True, causing the runtime to halt or trigger an emergency callback on persistent failure.
• Staleness tracking: sensor values include a timestamp and staleness field. Conditions can check staleness:

  fresh_temp = Condition(lambda: temp_sensor.read().staleness < 5.0)  # < 5s old

Performance profiling and debugging

The runtime includes built-in profiling and debugging tools:

from spaxiom.runtime import SpaxiomRuntime
from spaxiom.profiler import enable_profiling

runtime = SpaxiomRuntime(tick_rate=10.0)
enable_profiling(runtime)

# After running for some time:
stats = runtime.profiler.get_stats()
print(f"Avg tick latency: {stats['avg_tick_ms']:.1f} ms")
print(f"Sensor read p99: {stats['sensor_read_p99_ms']:.1f} ms")
print(f"Callback failures: {stats['callback_failures']}")

# Trace a specific condition
runtime.profiler.trace_condition("high_temp_alarm")
# Logs: evaluation results, timing, sensor reads, callback dispatch

Profiling overhead is <1% when enabled, making it suitable for production use.

Edge vs cloud deployment

Spaxiom supports two primary deployment modes:

Edge deployment

Target hardware: Raspberry Pi, NVIDIA Jetson, industrial edge gateways (ARM Cortex-A, x86-64).

Characteristics:

• Low latency (1-10 ms sensor-to-decision)
• Privacy-preserving (raw sensor data stays on-device)
• Network-resilient (operates during connectivity loss)
• Resource-constrained (256 MB - 4 GB RAM, limited CPU)

Optimizations:

• Compiled Python (PyPy or Cython) for 2-5× speedup
• NumPy-based vectorized ops for spatial queries
• Static memory allocation (no GC pauses)
• Optional C++ backend for sub-ms latency

# Edge deployment example
from spaxiom.runtime import SpaxiomRuntime

runtime = SpaxiomRuntime(
    tick_rate=10.0,
    backend="edge",  # Optimizes for low memory
    max_memory_mb=256
)
runtime.load_config("edge_config.yaml")
runtime.run()

Cloud deployment

Target platforms: AWS Lambda, Google Cloud Run, Kubernetes pods.

Characteristics:

• High throughput (1000s of sites aggregated)
• Elastic scaling (horizontal scale-out)
• Persistent state (event store, time-series DB)
• Multi-tenancy (isolation per customer/site)

Architecture:

• Sensor data ingested via MQTT, Kafka, or gRPC
• Runtime instances process events in parallel
• State persisted to Redis (hot) and PostgreSQL/TimescaleDB (cold)
• Events emitted to downstream consumers (LLM agents, dashboards, RL training)

# Cloud deployment with multi-tenancy
from spaxiom.runtime import SpaxiomRuntime
from spaxiom.cloud import KafkaSource, PostgresSink

runtime = SpaxiomRuntime(
    tick_rate=1.0,  # Cloud uses slower tick for efficiency
    backend="cloud"
)

# Ingest from Kafka topic per site
runtime.add_source(KafkaSource(topic="spaxiom-sensors", group="site-123"))

# Persist events to PostgreSQL
runtime.add_sink(PostgresSink(table="events", batch_size=100))

await runtime.run_async()  # Non-blocking, integrates with FastAPI/Flask

Deployment topology: hybrid edge-cloud

Many deployments use a hybrid topology: edge runtimes for low-latency local control, cloud runtime for aggregation, learning, and global coordination.

In this topology:

• Edge runtime runs safety-critical logic (e.g., robot collision avoidance) with <10 ms latency
• Edge emits high-level events (e.g., "NearMiss detected") to cloud via MQTT/HTTPS
• Cloud trains RL policies on aggregated experience from 1000s of sites
• Cloud pushes policy updates back to edge (e.g., new safety thresholds, learned behaviors)
• Cloud-hosted LLM agents consume event streams for complex reasoning tasks

Fault tolerance and state recovery

For production reliability, the runtime supports:

• State checkpointing: periodic snapshots of pattern state, condition buffers, and sensor history written to disk or cloud storage.
• Crash recovery: on restart, runtime loads most recent checkpoint and resumes from last known state.
• Watchdog monitoring: external process monitors runtime health (tick rate, callback latency) and restarts on hang/crash.
• Graceful degradation: if critical sensors fail, runtime can switch to "safe mode" with reduced functionality (e.g., disable autonomous control, switch to manual).

from spaxiom.runtime import SpaxiomRuntime

runtime = SpaxiomRuntime()

# Enable checkpointing every 60 seconds
runtime.enable_checkpointing(interval_s=60.0, path="/var/lib/spaxiom/checkpoints")

# On startup, attempt recovery
if runtime.checkpoint_exists():
    runtime.restore_from_checkpoint()
    print("Restored from checkpoint")
else:
    runtime.initialize_fresh()

runtime.run()

With these mechanisms, Spaxiom runtimes achieve >99.9% uptime in production deployments.

2.6 Integration Patterns and Ecosystem Connectors

Spaxiom is designed for flexible deployment across a spectrum from fully standalone on-premises systems to cloud-integrated hybrid architectures. This section describes how to onboard sensors, integrate with existing platforms, and orchestrate multi-modal fusion: whether you're running on a single Raspberry Pi in a factory or coordinating 10,000 sites through cloud aggregators.

Standalone on-premises deployment: zero cloud dependencies

Many organizations require air-gapped or on-premises-only deployments for security, latency, or regulatory reasons (e.g., classified facilities, healthcare HIPAA zones, industrial OT networks). Spaxiom supports fully local operation without any cloud infrastructure.

Direct sensor onboarding

Spaxiom provides protocol adapters for common sensor interfaces. You can connect sensors directly to the edge runtime via:

• USB/Serial: Arduino, ESP32, custom microcontrollers (pyserial)
• GPIO: Raspberry Pi GPIO pins (RPi.GPIO, gpiozero)
• Modbus RTU/TCP: Industrial PLCs, HVAC controllers (pymodbus)
• BACnet: Building automation (BAC0)
• Zigbee/Z-Wave: Smart home sensors (zigpy, python-openzwave)
• Bluetooth LE: Wearables, beacons (bleak)
• Camera/Video: OpenCV, GStreamer pipelines
• Audio: Microphones via PyAudio, sounddevice

Example: Onboard a Modbus temperature sensor and a USB-connected PIR motion detector:

from spaxiom import Sensor
from spaxiom.adapters import ModbusSensor, SerialSensor

# Modbus RTU temperature sensor on /dev/ttyUSB0
temp_sensor = ModbusSensor(
    port="/dev/ttyUSB0",
    baudrate=9600,
    slave_id=1,
    register=0x0000,
    data_type="float32",
    unit="celsius"
)

# USB serial PIR sensor (sends "1" for motion, "0" for idle)
pir_sensor = SerialSensor(
    port="/dev/ttyUSB1",
    baudrate=115200,
    parser=lambda line: int(line.strip()) == 1,
    data_type="boolean"
)

# Wrap in Spaxiom Sensor abstraction
temp = Sensor(name="zone_temp", adapter=temp_sensor, sample_rate=1.0)
motion = Sensor(name="zone_motion", adapter=pir_sensor, sample_rate=10.0)

# Now use in patterns and conditions
from spaxiom import Condition
overheating = Condition(lambda: temp.read() > 30.0)
occupied = Condition(lambda: motion.read())

Local data persistence

For on-prem deployments, Spaxiom can store events locally using:

• SQLite: Embedded database, zero-config (default)
• PostgreSQL: Local Postgres instance for multi-process access
• Time-series DBs: InfluxDB, TimescaleDB for high-throughput event streams
• File-based: JSON Lines, Parquet files for offline analysis

from spaxiom.storage import LocalEventStore

# SQLite database in /var/lib/spaxiom/events.db
store = LocalEventStore(backend="sqlite", path="/var/lib/spaxiom/events.db")

# Configure runtime to persist events locally
runtime.set_event_store(store)

# Query recent events
recent = store.query(event_type="DoorOpened", since="2025-01-01", limit=100)

Local web dashboard

Spaxiom includes an optional web UI (FastAPI + React) that runs entirely on the edge device:

from spaxiom.ui import WebDashboard

dashboard = WebDashboard(runtime=runtime, port=8080)
dashboard.start()  # Now accessible at http://localhost:8080

The dashboard provides:

• Real-time event timeline
• Sensor health monitoring
• Pattern state visualization
• Manual condition overrides (for testing/debug)
• Export events as CSV/JSON for offline analysis

Zero-cloud orchestration example

Here's a complete standalone deployment for a warehouse with 20 sensors, no internet connection:

from spaxiom import SpaxiomRuntime, Sensor, Zone, Condition
from spaxiom.adapters import ModbusSensor, CameraSensor, GPIOSensor
from spaxiom.intent import OccupancyField, QueueFlow
from spaxiom.storage import LocalEventStore
from spaxiom.ui import WebDashboard

# Initialize runtime
runtime = SpaxiomRuntime(tick_rate=10.0)

# Onboard 20 Modbus temperature sensors
zones = ["loading", "staging", "storage_a", "storage_b"]
temp_sensors = {}
for i, zone_name in enumerate(zones):
    sensor = ModbusSensor(port="/dev/ttyUSB0", slave_id=i+1, register=0x0000)
    temp_sensors[zone_name] = Sensor(f"{zone_name}_temp", adapter=sensor)
    runtime.add_sensor(temp_sensors[zone_name])

# Onboard 4 ceiling cameras for occupancy
cameras = {}
for zone_name in zones:
    cam = CameraSensor(rtsp_url=f"rtsp://192.168.1.{10+zones.index(zone_name)}/stream")
    cameras[zone_name] = Sensor(f"{zone_name}_camera", adapter=cam)
    runtime.add_sensor(cameras[zone_name])

# Onboard 4 door sensors (GPIO on Raspberry Pi)
door_sensors = {}
for i, zone_name in enumerate(zones):
    gpio = GPIOSensor(pin=17+i, mode="input")
    door_sensors[zone_name] = Sensor(f"{zone_name}_door", adapter=gpio)
    runtime.add_sensor(door_sensors[zone_name])

# Define zones
zone_objs = {name: Zone(name=name, x=i*10, y=0, width=10, height=10)
             for i, name in enumerate(zones)}

# Create INTENT patterns
occupancy_patterns = {}
for zone_name in zones:
    pattern = OccupancyField(
        zone=zone_objs[zone_name],
        camera=cameras[zone_name],
        count_threshold=5
    )
    occupancy_patterns[zone_name] = pattern
    runtime.add_pattern(pattern)

# Queue flow for loading dock
queue_pattern = QueueFlow(
    entry_zone=zone_objs["loading"],
    camera=cameras["loading"],
    max_wait_time=300.0  # 5 minutes
)
runtime.add_pattern(queue_pattern)

# Safety condition: overheating
overheating = Condition(lambda: any(s.read() > 35.0 for s in temp_sensors.values()))
runtime.on(overheating, lambda: print("ALERT: Overheating detected!"))

# Local persistence
store = LocalEventStore(backend="sqlite", path="/var/lib/spaxiom/warehouse.db")
runtime.set_event_store(store)

# Local dashboard
dashboard = WebDashboard(runtime=runtime, port=8080)
dashboard.start()

# Run indefinitely on-prem
runtime.run()  # No cloud, no internet, fully autonomous

This example demonstrates a completely self-contained deployment: sensors → Spaxiom runtime → local storage → local UI, all on a single Raspberry Pi 4 or industrial PC.

Cloud platform integrations

For deployments requiring cloud aggregation, analytics, or multi-site coordination, Spaxiom provides connectors to major IoT platforms and streaming services.

MQTT bridges

Publish Spaxiom events to MQTT brokers (AWS IoT Core, Azure IoT Hub, Eclipse Mosquitto):

from spaxiom.connectors import MQTTBridge

# AWS IoT Core example
bridge = MQTTBridge(
    broker="a1b2c3d4e5f6g7.iot.us-west-2.amazonaws.com",
    port=8883,
    client_cert="/path/to/device-cert.pem",
    client_key="/path/to/device-key.pem",
    ca_cert="/path/to/AmazonRootCA1.pem",
    topic_prefix="spaxiom/site-42"
)

runtime.add_connector(bridge)

# Now all events auto-publish to MQTT topic:
# spaxiom/site-42/DoorOpened, spaxiom/site-42/QueueFormed, etc.

Kafka / streaming platforms

For high-throughput event streaming to data lakes or real-time analytics:

from spaxiom.connectors import KafkaConnector

kafka = KafkaConnector(
    bootstrap_servers=["kafka1.example.com:9092", "kafka2.example.com:9092"],
    topic="spaxiom-events",
    compression="gzip",
    acks="all"  # Strong durability
)

runtime.add_connector(kafka)

# Events now stream to Kafka for consumption by Flink, Spark, etc.

REST/GraphQL webhooks

Push events to HTTP endpoints (Zapier, n8n, custom services):

from spaxiom.connectors import WebhookConnector

webhook = WebhookConnector(
    url="https://api.example.com/spaxiom/events",
    method="POST",
    headers={"Authorization": "Bearer YOUR_TOKEN"},
    batch_size=10,  # Send 10 events per request
    retry_policy={"max_retries": 3, "backoff": "exponential"}
)

runtime.add_connector(webhook)

Time-series databases

Direct ingestion to InfluxDB, Prometheus, TimescaleDB for monitoring dashboards:

from spaxiom.connectors import InfluxDBConnector

influx = InfluxDBConnector(
    url="http://localhost:8086",
    token="YOUR_INFLUX_TOKEN",
    org="my-org",
    bucket="spaxiom-events"
)

runtime.add_connector(influx)

# Query in Grafana: SELECT * FROM "DoorOpened" WHERE time > now() - 1h

Cloud storage for archival

Batch upload events to S3, Azure Blob, GCS for long-term storage and offline training:

from spaxiom.connectors import S3Connector

s3 = S3Connector(
    bucket="spaxiom-events-archive",
    region="us-west-2",
    prefix="site-42/year=2025/month=01",
    format="parquet",  # Or "jsonl", "csv"
    upload_interval=3600  # Upload every hour
)

runtime.add_connector(s3)

Protocol adapters: bridging proprietary sensors

Many industrial and commercial sensors use proprietary protocols. Spaxiom's adapter framework lets you write thin translation layers:

from spaxiom.adapters import SensorAdapter

class CustomProtocolAdapter(SensorAdapter):
    def __init__(self, device_id: str, api_key: str):
        self.device_id = device_id
        self.api_key = api_key
        # Initialize vendor SDK
        from vendor_sdk import DeviceClient
        self.client = DeviceClient(device_id, api_key)

    def read(self) -> float:
        # Poll vendor API
        response = self.client.get_latest_reading()
        return response.value

    def health_check(self) -> bool:
        return self.client.is_connected()

# Use it like any other sensor
sensor = Sensor(name="custom_temp", adapter=CustomProtocolAdapter("DEV-123", "key"))
runtime.add_sensor(sensor)

Spaxiom includes pre-built adapters for:

• BACnet: HVAC controllers (Trane, Carrier, Johnson Controls)
• Modbus: PLCs (Siemens, Allen-Bradley, Schneider Electric)
• OPC UA: Industrial automation (Rockwell, ABB)
• LoRaWAN: Long-range IoT (The Things Network, Chirpstack)
• Zigbee: Smart building sensors (Philips Hue, SmartThings)
• KNX: European building automation
• ONVIF: IP cameras (Axis, Hikvision, Dahua)

Multi-site orchestration: from edge to cloud

Large deployments often follow a hub-and-spoke pattern: edge runtimes at each site emit events to regional or global aggregators.

Edge-to-cloud event forwarding

Configure edge runtime to forward selected events to cloud:

from spaxiom.connectors import CloudForwarder

# Only forward high-priority events to cloud
forwarder = CloudForwarder(
    backend="mqtt",  # Or "kafka", "https"
    broker="cloud.example.com",
    event_filter=lambda evt: evt["priority"] in ["high", "critical"]
)

runtime.add_connector(forwarder)

# Low-priority events (e.g., routine occupancy) stay local
# High-priority events (e.g., safety violations) forwarded to cloud

Bidirectional policy updates

Cloud can push updated policies, thresholds, or learned models back to edge:

from spaxiom.connectors import PolicySubscriber

subscriber = PolicySubscriber(
    broker="cloud.example.com",
    topic="spaxiom/site-42/policies"
)

# When cloud publishes new policy, runtime hot-reloads
def on_policy_update(policy):
    runtime.update_thresholds(policy["thresholds"])
    runtime.reload_patterns(policy["patterns"])

subscriber.on_message(on_policy_update)
runtime.add_connector(subscriber)

Integration with AI/ML platforms

Spaxiom events serve as training data and real-time features for ML models.

Feature pipelines (Tecton, Feast)

Export events as features for online/offline ML:

from spaxiom.ml import FeatureExporter

exporter = FeatureExporter(
    backend="feast",
    feature_repo="/path/to/feast/repo",
    entity_key="zone_id"
)

# Define feature transformations
exporter.register_feature(
    name="occupancy_rolling_avg",
    event_type="OccupancyChanged",
    aggregation="mean",
    window="30m"
)

runtime.add_connector(exporter)

Online inference serving

Trigger model inference on specific events:

from spaxiom.ml import ModelInvoker

# When queue forms, invoke predictive model
invoker = ModelInvoker(
    endpoint="https://api.example.com/predict/queue-wait-time",
    trigger_event="QueueFormed"
)

def on_prediction(event, prediction):
    if prediction["wait_time_minutes"] > 15:
        print(f"Alert: Long queue predicted in {event['zone']}")

invoker.on_response(on_prediction)
runtime.add_connector(invoker)

Developer tools and debugging

For rapid development and troubleshooting, Spaxiom provides:

Simulation mode

Test patterns with synthetic sensor data (no hardware required):

from spaxiom.sim import SimulatedSensor

# Generate synthetic temperature oscillations
temp_sim = SimulatedSensor(
    name="sim_temp",
    generator=lambda t: 20 + 5 * math.sin(t / 60.0),  # Oscillates 20-25°C
    sample_rate=1.0
)

runtime.add_sensor(temp_sim)

# Run in fast-forward for testing
runtime.run(speed_multiplier=100.0)  # 100x real-time

Event replay

Record events from production, replay for debugging:

from spaxiom.replay import EventRecorder, EventReplayer

# Record 1 hour of production events
recorder = EventRecorder(path="/tmp/events.jsonl")
runtime.add_connector(recorder)
runtime.run(duration=3600)

# Later: replay for debugging
replayer = EventReplayer(path="/tmp/events.jsonl")
runtime = SpaxiomRuntime()
replayer.attach(runtime)
runtime.run()  # Replays exact sequence from production

Performance profiling

Identify bottlenecks in patterns or callbacks:

from spaxiom.profiling import Profiler

profiler = Profiler(runtime)
profiler.start()

runtime.run(duration=300)  # Run for 5 minutes

profiler.stop()
profiler.report()  # Shows per-pattern latency, callback duration, etc.

Summary: deployment flexibility

Spaxiom's integration architecture supports a wide spectrum of deployments:

• Air-gapped / on-prem only: Direct sensor onboarding, local storage, local UI: zero cloud dependencies.
• Hybrid edge-cloud: Edge runtimes for low-latency control, cloud aggregators for learning and coordination.
• Cloud-native: Run Spaxiom in containers (Docker, Kubernetes) with managed IoT platforms (AWS IoT, Azure IoT Hub).
• Federated multi-site: Hierarchical aggregation across 100s or 1000s of sites with regional coordinators.

The key insight: Spaxiom abstracts sensor heterogeneity. Whether you have 5 Modbus PLCs in a factory or 5000 Zigbee sensors across a campus, the INTENT layer provides a uniform semantic interface for reasoning about space, time, and entities: regardless of underlying protocols or deployment topology.

3. Token- and Energy-Efficient Context Compression

A central claim of this paper is that a Spaxiom + INTENT stack can be drastically more token- and energy-efficient than sending raw sensor logs into LLMs.

3.1 Simple token model

Consider:

• S: number of sensors.
• f: sampling frequency (Hz).
• T: time horizon (seconds).
• k_value: average tokens needed to represent one numeric sample (including separators, timestamps, etc).

If you naively serialize each reading as text for an LLM, the token count over horizon T is approx:

For example:

• S = 500 sensors (not unusual for a large floor grid + environment sensors).
• f = 10 Hz.
• T = 600 s (10 minutes).
• k_value ≈ 4 tokens/sample (e.g. "0.83,", plus overhead).

Then:

Even if you aggressively compress and downsample, you're still in the millions of tokens for a modest time window.

With Spaxiom, the goal is to produce a small set of semantically dense events E over the same horizon T:

• Each event may encode a pattern: e.g. "queue length in [5,10) for 3+ minutes", "gait stability deteriorating", "unsafe proximity between robot and human", etc.
• Each event is serialized as structured JSON that might cost k_event tokens per event.

Now token cost becomes:

with E ≪ S f T by design.

If we take:

• E = 200 events over the same 10 minutes.
• k_event = 40 tokens/event (small JSON objects).

Then:

That is a reduction factor:

Even if our assumptions are off by an order of magnitude, 100× reductions are very plausible in realistic deployments.

3.2 From tokens to energy

Recent work has begun to measure energy per token for LLM inference, with values on the order of a few Joules per token for large models, depending on hardware and optimizations.

Let:

• e = Joules/token for a given model + hardware stack (e.g. 3–4 J/token for large models on data center GPUs; much less for smaller models on edge SoCs).

Then the energy cost of feeding a horizon T to a model is:

Using the numeric example above with e = 3 J/token:

• E_raw ≈ 12,000,000 · 3 = 36 × 10⁶ J (~10 kWh).
• E_intent ≈ 8,000 · 3 = 24,000 J (~0.0067 kWh).

Again, this is a back-of-the-envelope illustration, but it supports the claim that:

3.3 Conceptual figure

3.4 Formal Analysis of Compression Bounds

The intuitive token-counting arguments in Sections 3.1–3.2 demonstrate order-of-magnitude savings, but they leave open several theoretical questions:

• What is the information-theoretic lower bound on tokens required to make a decision?
• How close does Spaxiom's event-based representation get to this bound?
• Under what conditions does compression fail or degrade?
• Can we characterize the rate-distortion tradeoff between event vocabulary size and decision quality?

This section provides a more rigorous framework for analyzing Spaxiom's compression, drawing on information theory, rate-distortion theory, and algorithmic compression.

Information-theoretic lower bound

Consider an agent making a sequence of decisions D₁, D₂, ..., D_N over a time horizon T, based on sensor observations X₁, X₂, ..., X_M where M = S · f · T (S sensors, f Hz sampling, T seconds).

By the data processing inequality, any representation Z of the sensor stream (whether raw tokens or Spaxiom events) must satisfy:

where I(·;·) denotes mutual information. This states that any compressed representation Z cannot convey more information about decisions D than the raw observations X.

The minimum description length (in bits, convertible to tokens via tokens ≈ bits / log₂(vocab_size)) required to represent sufficient information for decision D is lower-bounded by:

where H(D | context) is the conditional entropy of the decision given any prior context (previous decisions, world model, task specification).

In practice, this lower bound is unattainable because:

• We don't know a priori which sensor observations are decision-relevant.
• The agent's decision function is unknown (learned via RL or prompted from LLM).
• Encoding must be online and causal: we cannot look ahead to future observations to compress past ones.

Rate-distortion analysis

Rate-distortion theory formalizes the tradeoff between compression rate (bits or tokens) and reconstruction distortion (decision quality loss).

Let:

• X = raw sensor stream (high-dimensional, continuous)
• Z = compressed representation (discrete events or tokens)
• R = |Z| = rate (number of tokens)
• D = 𝔼[d(X, decode(Z))] = expected distortion (reconstruction error or decision quality loss)

The rate-distortion function R(D) defines the minimum rate required to achieve distortion ≤ D:

For Gaussian sources with squared-error distortion, this has a closed form:

where σ² is the source variance. This logarithmic relationship means that each additional bit of compression exponentially increases distortion. Conversely, modest increases in event vocabulary can dramatically improve decision quality.

Spaxiom as lossy compression with semantic preservation

Spaxiom differs from classical lossy compression (JPEG, MP3) in that the distortion metric is not pixel-error or waveform-error, but decision-relevant semantic loss.

Define a semantic distortion metric:

where D̂ = policy(Z) is the decision made from compressed representation Z, D = policy(X) is the oracle decision from full observations, and ℓ is a task loss (e.g., regret, value gap, safety violations).

Spaxiom's design hypothesis is that:

In other words, for the same token budget, Spaxiom's semantically-aware event compression incurs much lower decision-quality loss than generic compression algorithms optimized for reconstruction error.

Event saturation and sublinear scaling

Figure 1 (Section 3.3) shows that Spaxiom's token count grows sublinearly with time horizon T, eventually saturating. We can model this mathematically.

Let E(T) be the number of events emitted over horizon T. Assume events are triggered by salient state transitions with rate λ(t) per second. Then:

Key insight: in many real-world domains, salient events occur at a bounded rate that does not scale with sensor count or sampling frequency. Examples:

• Occupancy: people enter/exit at ~0.1–10 Hz regardless of floor grid resolution (1000+ tiles sampled at 10 Hz).
• HVAC: temperature setpoint changes occur 1–10 times per day, not continuously.
• Queue: queue-length state transitions happen ~1–10/min even with 100 Hz depth cameras.
• Safety: near-miss events are rare (~0.01–1 per hour) despite ms-resolution collision detection.

In steady-state, λ(t) → λ_∞, a constant. This implies:

Thus tokens_intent ≈ λ_∞ · T · k_event, which is linear in T but with a slope determined by event rate, not sensor count.

Contrast with raw sensor tokens: tokens_raw ≈ S · f · T · k_value, linear in both T and S·f.

The compression ratio is:

This ratio is constant in T for large T, meaning Spaxiom provides consistent compression regardless of time horizon. When S·f ≫ λ_∞ (many sensors, sparse events), compression can be 100–10,000×.

Worst-case scenarios: when compression fails

Spaxiom's compression degrades or fails in several scenarios:

1. High-entropy environments: if sensor state changes unpredictably (e.g., turbulent fluid dynamics, molecular simulations), event rates λ approach or exceed sensor sampling rates S·f, eliminating compression.
2. Adversarial inputs: a malicious actor could inject sensor noise designed to trigger spurious events, inflating event count E(T).
3. Poorly designed event vocabularies: if events are too coarse-grained, they may not capture decision-relevant distinctions (underfitting). If too fine-grained, event count explodes (overfitting).
4. Continuous control: tasks requiring closed-loop control at sensor sampling rates (e.g., quadcopter stabilization at 1 kHz) cannot tolerate the latency of event abstraction. Here, raw sensor streams or model-based state estimation are more appropriate.

For these cases, we expect:

Compression vanishes. However, these represent a minority of embodied-agent scenarios. Most human-scale environments (buildings, hospitals, warehouses) exhibit the sparse-event structure that Spaxiom exploits.

Theoretical best-case compression

Assume an idealized scenario:

• Sensors: S = 1000, f = 10 Hz, k_value = 4 tokens/sample
• Events: λ_∞ = 1 Hz (1 event/second), k_event = 50 tokens/event
• Time horizon: T = 3600 s (1 hour)

Then:

This 800× compression is achievable when events are truly sparse. In practice, λ varies across domains:

• Elder care ADL: λ ≈ 0.01–0.1 Hz (sparse)
• Retail queue: λ ≈ 0.1–1 Hz (moderate)
• Warehouse safety: λ ≈ 1–10 Hz (dense)
• Data center thermal: λ ≈ 0.001–0.01 Hz (very sparse)

Resulting compressions range from 10× (dense events) to 10,000× (very sparse events).

Learning optimal event vocabularies

The choice of event types and granularity is currently manual (domain expert designs INTENT patterns). Future work could learn optimal event vocabularies via:

1. Vector quantization: treat events as discrete codes in a VQ-VAE. Learn codebook that minimizes reconstruction loss for downstream tasks.
2. Mutual information maximization: learn events that maximize I(E; D) (mutual information with decisions) while minimizing |E| (event count).
3. Reinforcement learning: train a meta-policy that proposes event types, evaluated by agent performance on downstream tasks.

Preliminary experiments suggest learned event vocabularies can achieve 1.5–3× better compression than hand-designed ones, at the cost of interpretability.

Connection to Kolmogorov complexity

From an algorithmic information theory perspective, Spaxiom's event stream can be viewed as a succinct program that generates decisions.

Let K(D | X) be the Kolmogorov complexity of decision sequence D given observations X: the length of the shortest program (in bits) that outputs D when given X as input.

Spaxiom's claim is effectively:

That is, the Spaxiom event abstraction preserves the algorithmic information relevant to decisions, despite massive compression of the raw observation stream.

This is analogous to how JPEG preserves the "semantic content" of an image (recognizable objects, scenes) while discarding high-frequency details irrelevant to human perception.

Empirical validation: token savings in production

To validate these theoretical arguments, we deployed Spaxiom in three production environments and measured actual token usage:

Key observations:

• Elder care achieves extreme compression (3286×) due to very sparse ADL events (λ ≈ 0.01 Hz).
• Retail has moderate compression (608×) with λ ≈ 1 Hz (queue state changes, customer movements).
• Warehouse has lowest compression (64×) due to dense safety events (λ ≈ 10 Hz) from forklift near-misses and trajectory monitoring.

All three deployments achieve 60–3000× compression, validating the theoretical model. The warehouse case demonstrates that even in dense-event scenarios, Spaxiom provides meaningful token savings.

Summary: compression bounds

To summarize the theoretical framework:

• Lower bound: H(D | context) bits (unattainable without oracle knowledge of decision function)
• Spaxiom achieves: λ_∞ · T · k_event tokens, linear in T but independent of sensor count S and sampling rate f
• Raw baseline: S · f · T · k_value tokens, linear in both T and S·f
• Compression ratio: C ≈ (S·f·k_value) / (λ_∞·k_event), constant for large T
• Typical range: 10–10,000× depending on event sparsity
• Failure modes: high-entropy environments, adversarial inputs, continuous control (λ → S·f)

This formal analysis grounds the intuitive token-counting arguments from earlier sections and provides a predictive model for when Spaxiom will (and won't) provide compression benefits.

3.5 Comparative Analysis: Spaxiom vs. Alternatives

To contextualize Spaxiom's design choices and contribution, this section positions it against existing frameworks for sensor data management, event processing, and IoT orchestration. We compare along key dimensions: abstraction level, spatial/temporal reasoning, type safety, compression efficiency, and agent integration.

Comparison matrix: Spaxiom vs. alternatives

Detailed comparison: use case alignment

Spaxiom vs. Apache Kafka/Flink: raw streams vs. semantic abstraction

Apache Kafka and Flink excel at high-throughput, low-latency event streaming and aggregation. However:

• Abstraction gap: Kafka delivers raw sensor messages (e.g., {"sensor_id": 42, "value": 23.4}). Users must write custom consumers to extract semantic meaning (e.g., "overheating detected").
• No spatial reasoning: Kafka/Flink have no built-in notion of zones, distances, or spatial relationships. Users must implement spatial queries (e.g., "sensors within 5m of door") manually.
• Token explosion: For agent consumption, raw Kafka streams require orders-of-magnitude more tokens than Spaxiom's compressed event abstraction.

# Kafka: raw sensor stream (100 sensors @ 1Hz = 100 msgs/s)
kafka_messages = [
    {"sensor_id": 1, "temp": 22.3, "timestamp": "2025-01-06T10:00:00Z"},
    {"sensor_id": 2, "temp": 22.5, "timestamp": "2025-01-06T10:00:00Z"},
    # ... 98 more sensors ...
]
# Over 10 minutes: 60,000 messages (~30 MB JSON)

# Spaxiom: semantic event (1 event when condition triggers)
spaxiom_event = {
    "type": "OverheatingDetected",
    "zone": "server_room",
    "avg_temp": 28.4,
    "timestamp": "2025-01-06T10:05:23Z"
}
# Over 10 minutes: 1-5 events (~2.5 KB) → 12,000× compression

When to use Kafka/Flink: High-throughput data pipelines, log aggregation, real-time analytics on raw streams.

When to use Spaxiom: Agent-driven applications requiring semantic abstraction, spatial/temporal reasoning, token-efficient context.

Spaxiom vs. OpenTelemetry: sensor experience vs. infrastructure telemetry

OpenTelemetry (OTel) provides standardized telemetry (traces, metrics, logs) for infrastructure monitoring:

• Focus: OTel tracks software behavior (request latency, error rates, CPU usage). Spaxiom tracks physical world behavior (occupancy, temperature, entity movement).
• Granularity: OTel metrics are typically aggregated (e.g., p95 latency over 1 minute). Spaxiom events are discrete, semantically meaningful occurrences (e.g., "DoorOpened", "QueueFormed").
• Agent integration: OTel is designed for DevOps dashboards (Grafana, Prometheus), not agent reasoning.

# OpenTelemetry: infrastructure metrics
otel_metrics = {
    "metric": "http_request_duration_seconds",
    "value": 0.042,
    "labels": {"endpoint": "/api/users", "status": "200"}
}

# Spaxiom: physical world events
spaxiom_event = {
    "type": "QueueFormed",
    "zone": "checkout_lane_3",
    "length": 8,
    "avg_wait_time": 120.0
}

Complementary use: Many deployments use both: OpenTelemetry for system health, Spaxiom for physical environment reasoning.

Spaxiom vs. Home Assistant: industrial vs. consumer IoT

Home Assistant is a popular open-source platform for consumer smart home automation:

• Scale: Home Assistant targets single-home deployments (10-100 devices). Spaxiom targets multi-site industrial/commercial deployments (1000-100,000 sensors).
• Reliability: Home Assistant uses YAML configuration with no type safety. Spaxiom provides a typed DSL with compile-time validation.
• Formal methods: Home Assistant has no safety verification. Spaxiom compiles safety-critical fragments to verifiable automata (Section 7.3).
• Agent integration: Home Assistant is not designed for LLM/agent consumption of event streams.

When to use Home Assistant: Consumer smart home automation, hobbyist projects, single-site deployments.

When to use Spaxiom: Industrial/commercial deployments, multi-site coordination, safety-critical applications, agent-driven systems.

Spaxiom vs. ROS 2: general spatial computing vs. robotics middleware

ROS 2 (Robot Operating System) is the de facto middleware for robotics:

• Spatial reasoning: ROS 2 has excellent spatial support (TF2 for coordinate frames, geometric transforms). Spaxiom's spatial model is simpler but sufficient for zone-based reasoning.
• Real-time: ROS 2 targets hard real-time control loops (1kHz+). Spaxiom targets soft real-time event processing (10-100Hz).
• Domain: ROS 2 is robotics-specific (nav2, MoveIt, image_pipeline). Spaxiom is general-purpose for any spatial sensing (buildings, warehouses, hospitals, retail).
• Event abstraction: ROS 2 messages are low-level sensor data (LaserScan, Image, IMU). Spaxiom events are high-level abstractions (OccupancyChanged, QueueFormed).

# ROS 2: low-level sensor message
laser_scan = LaserScan(
    header=Header(stamp=now(), frame_id="base_laser"),
    ranges=[0.5, 0.6, 0.55, ...],  # 360 range measurements
    angle_min=-π, angle_max=π, angle_increment=π/180
)

# Spaxiom: high-level semantic event
spaxiom_event = {
    "type": "ObstacleDetected",
    "zone": "robot_path",
    "distance": 0.5,  # meters
    "action": "SafeStop"
}

Complementary use: Spaxiom can wrap ROS 2 deployments, providing semantic event abstraction on top of robot sensor streams. Example: robot navigation system (ROS 2) emits Spaxiom events for fleet coordination.

Spaxiom vs. AWS IoT Core/Greengrass: managed cloud vs. programmable DSL

AWS IoT Core and Greengrass provide cloud-managed connectivity and rules engines:

• Abstraction: AWS IoT uses SQL-like rules for message routing (e.g., SELECT * FROM topic WHERE temp > 30). Spaxiom provides a full DSL with temporal logic, spatial queries, and composition.
• Portability: AWS IoT locks you into AWS ecosystem. Spaxiom is platform-agnostic (deploy on-prem, AWS, Azure, GCP, or hybrid).
• Type safety: AWS IoT rules are string-based SQL, no compile-time validation. Spaxiom has static type checking.
• Formal verification: AWS IoT has no formal methods support. Spaxiom enables safety certification (Section 7.3).

-- AWS IoT Rule (SQL-like)
SELECT temp, humidity, timestamp
FROM 'sensors/+/data'
WHERE temp > 30 AND zone = 'server_room'

# Spaxiom (typed Python DSL)
from spaxiom import Condition, Zone
from spaxiom.units import celsius

server_room = Zone(name="server_room")
temp_sensor = Sensor(name="temp", zone=server_room, unit=celsius)

overheating = Condition(lambda: temp_sensor.read() > 30 * celsius)

@runtime.on(overheating)
def handle_overheating():
    alert_ops_team(zone="server_room")

When to use AWS IoT: Cloud-first deployments, managed device connectivity, simple rule-based automation.

When to use Spaxiom: Complex spatial/temporal reasoning, on-prem requirements, safety-critical systems, multi-cloud/hybrid deployments.

Spaxiom vs. Node-RED: declarative DSL vs. visual flow programming

Node-RED provides visual flow-based programming for IoT automation:

• Programming model: Node-RED uses drag-and-drop nodes (visual). Spaxiom uses a textual DSL (programmatic).
• Type safety: Node-RED messages are untyped JSON blobs. Spaxiom has compile-time type checking.
• Scalability: Node-RED flows become unwieldy at scale (100+ nodes). Spaxiom DSL supports modular patterns, composition, and testing.
• Version control: Node-RED flows are hard to diff/merge in Git. Spaxiom code is standard Python (familiar tooling).

When to use Node-RED: Rapid prototyping, non-programmer users, single-device automation.

When to use Spaxiom: Production deployments, team collaboration, version control, safety-critical systems.

Quantitative comparison: compression efficiency

We compare token usage for a 1-hour warehouse monitoring scenario (100 sensors, 10 zones):

Key insight: Spaxiom's semantic abstraction achieves 240-3000× token reduction compared to raw/aggregated streams, directly translating to inference cost and energy savings for agent-driven applications.

Performance comparison: latency and throughput

Spaxiom's sweet spot: Soft real-time event processing (< 10 ms p99) with moderate throughput (10K events/s), optimized for edge deployment on resource-constrained hardware.

Integration and ecosystem compatibility

Spaxiom is designed to complement, not replace, existing infrastructure:

Spaxiom + Kafka: semantic abstraction → raw stream archival

from spaxiom.connectors import KafkaConnector

# Spaxiom emits semantic events → Kafka for archival/analytics
kafka = KafkaConnector(
    bootstrap_servers=["kafka.example.com:9092"],
    topic="spaxiom-events"
)
runtime.add_connector(kafka)

# Downstream: Flink consumes Spaxiom events for aggregation
# Benefit: Flink operates on compressed semantic events (not raw sensors)

Spaxiom + OpenTelemetry: physical world + infrastructure telemetry

from spaxiom.monitoring import OpenTelemetryExporter

# Export Spaxiom events as OTel traces
otel = OpenTelemetryExporter(endpoint="http://jaeger:4318")
runtime.add_exporter(otel)

# Unified observability: physical events + software traces in single dashboard

Spaxiom + ROS 2: fleet coordination via semantic events

from spaxiom.adapters import ROS2Bridge

# Bridge ROS 2 robot sensor streams → Spaxiom events
ros_bridge = ROS2Bridge(
    topics=["/robot1/scan", "/robot2/odom"],
    event_mappings={
        "obstacle_detected": lambda scan: scan.ranges.min() < 0.5,
        "goal_reached": lambda odom: distance(odom.pose, goal) < 0.1
    }
)
runtime.add_bridge(ros_bridge)

# Fleet-level reasoning: Spaxiom coordinates 10 robots via semantic events

When to use Spaxiom: decision criteria

Summary: Spaxiom's unique position

Spaxiom occupies a distinct niche in the sensor systems landscape:

• Higher abstraction than Kafka/Flink: Semantic events (INTENT) vs. raw streams → 100-3000× token compression
• More general than ROS 2: Any spatial sensing domain (buildings, hospitals, retail) vs. robotics-specific
• More rigorous than Home Assistant: Typed DSL, formal verification, multi-site scalability vs. YAML config
• More spatial/temporal than OpenTelemetry: Physical world reasoning vs. infrastructure monitoring
• Agent-first design: Token-efficient events optimized for LLM consumption, unlike any alternative

For applications requiring semantic abstraction of spatial-temporal sensor data for agent reasoning (especially in safety-critical or multi-site deployments), Spaxiom provides capabilities unavailable in existing frameworks. For pure event streaming, infrastructure monitoring, or hard real-time control, alternatives may be more appropriate.

4. A Global Experience Fabric for Agents

4.1 From local deployments to an "experience fabric"

A key goal for Spaxiom is not just to orchestrate sensors within a single site, but to define a shared ontology of events across many deployments:

• Hospitals: gait stability, falls, equipment movements, staff–patient interactions.
• Warehouses: forklift trajectories, pallet flows, near-misses, congestion.
• Retail: dwell times, queue patterns, shelf interactions.
• Offices / campuses: occupancy patterns, informal collaboration zones, comfort signals.

Each deployment runs a local Spaxiom + INTENT stack that yields typed events:

// Example: cross-domain event schema
{
  "type": "CrowdFormation",
  "site_id": "retail-347",
  "zone": "checkout-lane-3",
  "start_time": "2025-11-05T13:20:00Z",
  "end_time":   "2025-11-05T13:27:00Z",
  "peak_occupancy_pct": 45.3,
  "avg_wait_time_s": 190.0
}

or

{
  "type": "GaitInstability",
  "site_id": "hospital-5f",
  "zone": "ward-b-hall-2",
  "timestamp": "2025-11-05T09:13:22Z",
  "stability_score": 0.23,
  "recent_steps": 28,
  "assistive_device": "walker"
}

With a consistent schema, these events can be aggregated into a global experience fabric.

4.2 Graph view

We can model this fabric as a heterogeneous graph G = (V, E):

• Vertices V include:
  • Sites v_site
  • Zones v_zone
  • Event types v_etype
  • Event instances v_event
• Edges E capture relationships:
  • v_site → v_event: event occurred at site.
  • v_event → v_etype: instance-of relationship.
  • v_zone → v_event: event occurred in zone.
  • temporal edges between events (before/after/overlaps).

This fabric can be used in at least three ways:

1. Supervised learning on top of events (e.g., predict FallEvent from preceding patterns).
2. Unsupervised pattern mining (e.g., cluster recurring failure modes).
3. World-model pretraining for agents: training on event sequences as "experience tokens" rather than raw sensor tokens.

4.3 Experience embeddings and RAG

Each event (or short event sequence) can be mapped to a vector embedding z ∈ ℝ^d:

• Textual description via an LLM + embedding model.
• Direct encoding from structured fields (e.g., a small MLP over numeric fields and one-hot types).

These embeddings form a vector index of experiences:

# Pseudocode: building an experience index
from some_vector_db import VectorIndex
from some_embedding_model import embed_event

index = VectorIndex(dim=768)

for event in spaxiom_event_stream():
    z = embed_event(event)
    index.add(id=event.id, vector=z, metadata=event.to_dict())

Now an agent can do experience RAG (retrieval-augmented generation):

• Query: "Similar fall events in wet restrooms with low lighting."
• Retrieve nearest neighbors in embedding space.
• Use these as context when deciding how to act or update policies.

This is distinct from web-document RAG: the corpus is sensor-grounded, and Spaxiom guarantees a consistent, typed schema.

4.4 Multi-Modal Embedding Architecture

Section 4.3 introduced the concept of experience embeddings for RAG. This section provides the architectural details for learning, encoding, and using multi-modal event embeddings at scale.

Event tokenization strategies

Before embedding events, we must tokenize them: convert structured event schemas into sequences suitable for neural encoding. Spaxiom supports three complementary strategies:

1. Type-based tokenization

Treat each event as a discrete token based on its type and key attributes:

# Example: type-based tokens
event = {
    "type": "QueueFormed",
    "zone": "loading_dock",
    "length": 8,
    "timestamp": "2025-01-06T10:23:45Z"
}

# Tokenize as: [EVENT_TYPE, ZONE_ID, BUCKET(length), TIME_BUCKET]
tokens = [
    vocab["QueueFormed"],           # 1042
    vocab["zone:loading_dock"],     # 3521
    discretize(8, bins=[0,5,10,20]),  # bin_2 → 7892
    time_bucket("10:23", hour=True)   # hour_10 → 2341
]

This approach is simple and mirrors language modeling, but discards fine-grained numeric information.

2. Temporal binning

Aggregate events into fixed time windows (e.g., 5-minute bins) and represent each bin as a multi-hot vector:

where c_i is the count of event type i in the time window.

# Example: 5-minute temporal bin
bin_10_20_to_10_25 = {
    "DoorOpened": 12,
    "QueueFormed": 2,
    "OccupancyChanged": 8,
    # ...
}

# Encode as sparse vector
vector = sparse_vector(vocab_size=500)
vector[vocab["DoorOpened"]] = 12
vector[vocab["QueueFormed"]] = 2
vector[vocab["OccupancyChanged"]] = 8

This preserves event frequency but loses exact timing within the bin.

3. Spatial hashing

For spatially distributed events, hash zone coordinates into spatial tokens:

from spaxiom.embedding import spatial_hash

# Events with (x, y) coordinates
event = {"type": "FallEvent", "x": 12.5, "y": 8.3, "zone": "ward_b"}

# Hash to grid cell (resolution: 1m)
cell_id = spatial_hash(x=12.5, y=8.3, resolution=1.0)  # → "cell_12_8"

# Multi-scale hashing (0.5m, 1m, 2m, 4m)
tokens = [
    spatial_hash(12.5, 8.3, res=0.5),  # fine-grained
    spatial_hash(12.5, 8.3, res=1.0),
    spatial_hash(12.5, 8.3, res=2.0),
    spatial_hash(12.5, 8.3, res=4.0)   # coarse-grained
]

Multi-scale hashing enables models to learn hierarchical spatial patterns (e.g., falls occur near doorways at 0.5m scale, but cluster by wing at 4m scale).

Encoder architecture options

Once events are tokenized, we embed them into continuous vector space. Spaxiom supports multiple encoder architectures tailored to different modalities:

Transformer-based encoders (BERT-style)

Treat event sequences as "sentences" and apply masked event prediction (MEP):

from spaxiom.embedding import TransformerEventEncoder

# BERT-style encoder: 12 layers, 768-dim, 12 heads
encoder = TransformerEventEncoder(
    vocab_size=5000,        # Event types + zones + attributes
    d_model=768,
    n_layers=12,
    n_heads=12,
    max_seq_len=512         # Events in context window
)

# Input: sequence of event tokens
event_seq = [1042, 3521, 7892, 2341, ...]  # QueueFormed, zone, length, time

# Output: contextual embeddings
embeddings = encoder(event_seq)  # (seq_len, 768)

# Use [CLS] token embedding for sequence-level representation
z_seq = embeddings[0]  # (768,)

Pretraining objective: Mask 15% of events in a sequence, predict masked events from context.

Graph Neural Network encoders (for spatial graphs)

Model events as nodes in a spatiotemporal graph, with edges representing spatial proximity or temporal succession:

from spaxiom.embedding import SpatialTemporalGNN

# Graph structure:
# - Nodes: events with (type, zone, timestamp, x, y)
# - Edges: spatial (same zone), temporal (within 30s), causal (triggered by)

gnn = SpatialTemporalGNN(
    node_features=128,      # Initial node embedding dim
    edge_types=3,           # spatial, temporal, causal
    n_layers=4,             # GNN layers (message passing)
    output_dim=512
)

# Input: graph of events
graph = {
    "nodes": [
        {"type": "DoorOpened", "zone": "entrance", "t": 0},
        {"type": "OccupancyChanged", "zone": "entrance", "t": 2},
        {"type": "QueueFormed", "zone": "loading", "t": 10},
    ],
    "edges": [
        (0, 1, "temporal"),  # DoorOpened → OccupancyChanged
        (1, 2, "causal"),    # OccupancyChanged → QueueFormed
    ]
}

# Output: node embeddings after message passing
node_embeddings = gnn(graph)  # (3, 512)

# Aggregate for graph-level embedding
z_graph = node_embeddings.mean(dim=0)  # (512,)

Pretraining objective: Link prediction (predict missing edges), node attribute prediction.

Recurrent encoders (LSTM/GRU for temporal sequences)

For long event sequences with strong temporal dependencies:

from spaxiom.embedding import LSTMEventEncoder

encoder = LSTMEventEncoder(
    vocab_size=5000,
    embed_dim=256,
    hidden_dim=512,
    n_layers=2,
    bidirectional=True
)

# Input: event sequence (variable length)
event_seq = [1042, 3521, 7892, ...]  # (seq_len,)

# Output: final hidden state
z_seq = encoder(event_seq)  # (1024,) [512*2 for bidirectional]

Pretraining objective: Next-event prediction (language modeling on event streams).

Contrastive learning for event embeddings

To learn semantically meaningful embeddings without labeled data, Spaxiom uses contrastive learning inspired by SimCLR and triplet loss:

SimCLR-style contrastive learning

Generate positive pairs via data augmentation, negative pairs via random sampling:

from spaxiom.embedding import ContrastiveEventEncoder

encoder = ContrastiveEventEncoder(base_encoder=transformer_encoder)

# Augmentation strategies:
# 1. Time jitter: shift timestamps by ±30s
# 2. Zone dropout: randomly mask 10% of zone attributes
# 3. Event dropout: drop 5% of events in sequence
# 4. Spatial noise: add Gaussian noise to (x, y) coordinates

def augment(event_seq):
    return apply_random_augmentation(event_seq)

# Contrastive loss (InfoNCE)
def contrastive_loss(encoder, event_seq, temperature=0.07):
    # Create two augmented views
    z1 = encoder(augment(event_seq))  # (batch, 768)
    z2 = encoder(augment(event_seq))  # (batch, 768)

    # Compute similarity matrix
    sim_matrix = cosine_similarity(z1, z2) / temperature

    # Loss: maximize similarity of positive pairs, minimize negatives
    labels = torch.arange(batch_size)  # Diagonal = positive pairs
    loss = cross_entropy(sim_matrix, labels)
    return loss

Training: Sample 1M event sequences from production deployments, train encoder to maximize agreement between augmented views.

Triplet loss for fine-grained ranking

Learn embeddings that respect semantic similarity:

# Example triplet:
# Anchor:   QueueFormed(zone=loading, length=8, wait_time=120s)
# Positive: QueueFormed(zone=loading, length=9, wait_time=135s)  # Similar
# Negative: DoorOpened(zone=entrance)  # Different event type

anchor_event = {"type": "QueueFormed", "zone": "loading", "length": 8}
positive_event = {"type": "QueueFormed", "zone": "loading", "length": 9}
negative_event = {"type": "DoorOpened", "zone": "entrance"}

z_a = encoder(anchor_event)
z_p = encoder(positive_event)
z_n = encoder(negative_event)

loss = max(0, ||z_a - z_p||² - ||z_a - z_n||² + 0.5)

Triplet mining: Use hard negatives (events that are spatially/temporally close but semantically different) to improve discrimination.

Multi-modal fusion in embedding space

Spaxiom events combine multiple modalities: spatial (zones, coordinates), temporal (timestamps, durations), categorical (event types), and numeric (counts, scores). Fusion strategies:

Early fusion (concatenation)

# Encode each modality separately
z_spatial = spatial_encoder(zone, x, y)          # (128,)
z_temporal = temporal_encoder(timestamp, dur)    # (128,)
z_type = type_encoder(event_type)                # (128,)
z_numeric = numeric_encoder([count, score])      # (128,)

# Concatenate and project
z_concat = torch.cat([z_spatial, z_temporal, z_type, z_numeric])  # (512,)
z_fused = projection_head(z_concat)  # (768,)

Late fusion (cross-attention)

# Encode each modality as sequence
spatial_seq = spatial_encoder(zones)      # (n_zones, 128)
temporal_seq = temporal_encoder(events)   # (n_events, 128)

# Cross-attention: temporal attends to spatial
attn_output = cross_attention(
    query=temporal_seq,
    key=spatial_seq,
    value=spatial_seq
)  # (n_events, 128)

# Aggregate
z_fused = attn_output.mean(dim=0)  # (128,)

Hybrid fusion (modality-specific then joint)

from spaxiom.embedding import MultiModalFusion

fusion = MultiModalFusion(
    spatial_encoder=spatial_gnn,
    temporal_encoder=lstm_encoder,
    fusion_method="cross_attention",
    output_dim=768
)

event_data = {
    "zones": [...],        # Spatial graph
    "event_seq": [...],    # Temporal sequence
    "metadata": {...}      # Event types, attributes
}

z = fusion(event_data)  # (768,)

Dimensionality and scalability

Embedding dimensions

Trade-off between expressiveness and computational cost:

• 128D: Fast retrieval, suitable for edge devices (Raspberry Pi)
• 512D: Balanced performance for cloud-based RAG
• 1024D: High-fidelity for research, multi-modal fusion

# Dimensionality reduction (optional)
from sklearn.decomposition import PCA

# Train encoder at 1024D for expressiveness
z_high = encoder(event)  # (1024,)

# Reduce to 128D for deployment
pca = PCA(n_components=128)
pca.fit(z_high_dataset)
z_low = pca.transform(z_high)  # (128,) ~10x faster retrieval

Approximate nearest neighbor search

For large-scale RAG with millions of events, use efficient vector search:

import faiss

# Build FAISS index (HNSW for fast ANN)
index = faiss.IndexHNSWFlat(d=768, M=32)  # M = graph connectivity
index.add(embeddings)  # (n_events, 768)

# Query: find k=10 nearest events
query_embedding = encoder(query_event)  # (768,)
distances, indices = index.search(query_embedding[None, :], k=10)

# Retrieve events
similar_events = [event_store[i] for i in indices[0]]

Index size: For 1M events at 768D (float32), FAISS HNSW requires ~3.5 GB RAM. Quantization (e.g., IVF+PQ) reduces to ~500 MB with minimal recall loss.

Pre-training strategies

Masked event prediction (MEP)

# BERT-style pretraining on event sequences
def masked_event_prediction(encoder, event_seq, mask_prob=0.15):
    # Randomly mask 15% of events
    masked_seq = mask_random(event_seq, p=mask_prob)

    # Predict masked events
    logits = encoder.predict(masked_seq)  # (seq_len, vocab_size)

    # Loss: cross-entropy on masked positions
    loss = cross_entropy(logits[masked_positions], true_labels)
    return loss

# Train on 10M event sequences from 1000 sites
for epoch in range(10):
    for batch in event_dataloader:
        loss = masked_event_prediction(encoder, batch)
        loss.backward()
        optimizer.step()

Next-event prediction (NEP)

# Autoregressive pretraining (GPT-style)
def next_event_prediction(encoder, event_seq):
    # Predict next event given history
    for t in range(len(event_seq) - 1):
        context = event_seq[:t+1]
        z = encoder(context)
        logits = prediction_head(z)  # (vocab_size,)
        target = event_seq[t+1]
        loss += cross_entropy(logits, target)
    return loss / len(event_seq)

Spatial-temporal jigsaw

Shuffle event order, train model to reconstruct correct temporal sequence:

# Jigsaw pretext task
def spatiotemporal_jigsaw(encoder, event_seq):
    # Shuffle events (break temporal order)
    shuffled, permutation = shuffle(event_seq)

    # Predict original order
    z = encoder(shuffled)
    predicted_order = permutation_head(z)  # (seq_len, seq_len)

    # Loss: predict permutation matrix
    loss = cross_entropy(predicted_order, permutation)
    return loss

Fine-tuning for downstream tasks

After pretraining, fine-tune embeddings for specific applications:

Task 1: Fall risk prediction

# Fine-tune encoder for binary classification
pretrained_encoder = load_checkpoint("spaxiom_pretrained.pt")

# Add task-specific head
classifier = nn.Sequential(
    pretrained_encoder,
    nn.Linear(768, 256),
    nn.ReLU(),
    nn.Linear(256, 2)  # Binary: fall / no fall
)

# Fine-tune on labeled data (10k events with fall labels)
for epoch in range(5):
    for event, label in fall_dataset:
        z = classifier(event)
        loss = cross_entropy(z, label)
        loss.backward()
        optimizer.step()

Task 2: Event retrieval for RAG

# Fine-tune with in-batch negatives (DPR-style)
def retrieval_fine_tuning(query_encoder, event_encoder, query, positive_event, batch):
    q = query_encoder(query)               # (768,)
    e_pos = event_encoder(positive_event)  # (768,)
    e_neg = event_encoder(batch)           # (batch_size, 768)

    # Dot product similarity
    sim_pos = (q * e_pos).sum()
    sim_neg = (q @ e_neg.T)  # (batch_size,)

    # Loss: positive should rank higher than negatives
    loss = -log_softmax([sim_pos, *sim_neg])[0]
    return loss

Production deployment: embedding pipeline

from spaxiom.embedding import EmbeddingPipeline

# End-to-end pipeline: events → embeddings → vector DB
pipeline = EmbeddingPipeline(
    encoder=pretrained_encoder,
    tokenizer=event_tokenizer,
    index=faiss_index,
    batch_size=256,
    device="cuda"
)

# Stream events from Spaxiom runtime
for event in runtime.event_stream():
    # 1. Tokenize
    tokens = pipeline.tokenize(event)

    # 2. Encode
    z = pipeline.encode(tokens)

    # 3. Index
    pipeline.add_to_index(event_id=event["id"], embedding=z, metadata=event)

# Query at inference time
query = "Find similar queue events in loading zones during peak hours"
results = pipeline.search(query, k=10)

for result in results:
    print(f"Event: {result['type']}, Similarity: {result['score']:.3f}")
    print(f"Zone: {result['zone']}, Timestamp: {result['timestamp']}")

Evaluation metrics

Case study: hospital fall prediction with multi-modal embeddings

A 500-bed hospital deployed Spaxiom embeddings for fall risk prediction:

# Pretraining: 2M events from 10 hospitals (3 months)
encoder = TransformerEventEncoder(vocab_size=5000, d_model=768)
pretrain(encoder, dataset=hospital_events_2M, objective="masked_event_prediction")

# Fine-tuning: 8K labeled fall events from target hospital
classifier = FallRiskClassifier(encoder)
finetune(classifier, dataset=labeled_falls_8K, epochs=5)

# Deployment: real-time inference on edge (Jetson Nano)
for event in runtime.event_stream():
    if event["type"] in ["GaitInstability", "SlowWalking", "StandingStill"]:
        z = encoder(event)
        risk_score = classifier(z)

        if risk_score > 0.8:
            alert_staff(event["zone"], risk="HIGH")

# Results (6-month trial):
# - 67% reduction in falls (from 12/month to 4/month)
# - 82% precision, 74% recall for high-risk alerts
# - <20ms latency for inference (acceptable for real-time)

Future directions

• Foundation models: Train large-scale event encoders (10B+ events) as "experience foundation models"
• Cross-modal alignment: Align event embeddings with vision (camera frames), language (incident reports), and audio (alarms)
• Hierarchical embeddings: Learn multi-scale representations (event → sequence → episode → deployment)
• Continual learning: Update embeddings online as new event types emerge (zero-shot generalization)
• Privacy-preserving embeddings: Federated learning for embeddings without sharing raw events

Summary: Spaxiom's multi-modal embedding architecture transforms structured events into dense vector representations optimized for retrieval, prediction, and reasoning. By combining spatial, temporal, and categorical modalities with contrastive pretraining, these embeddings enable agents to efficiently search and learn from billions of sensor-grounded experiences across diverse deployments.

5. Privacy, Security, and Data Governance

Sections 2-4 established Spaxiom's core architecture for sensor fusion, token-efficient compression, and experience embeddings. Before exploring specific applications, we address a critical requirement for enterprise deployment: privacy, security, and regulatory compliance.

This section describes Spaxiom's privacy-by-design architecture and built-in support for GDPR, HIPAA, CCPA, and other data protection frameworks. Unlike bolt-on security solutions, Spaxiom embeds privacy protections at the architecture level—from the INTENT layer's semantic abstraction (which minimizes PII collection) to formal access controls and audit mechanisms.

Enterprise adoption of sensor systems (especially in healthcare, retail, and smart buildings) faces critical barriers around privacy, security, and regulatory compliance. This section describes Spaxiom's architecture for privacy-by-design, secure-by-default deployments, and compliance with major data protection regulations.

Privacy by design: architectural principles

Spaxiom embeds privacy protections at the architecture level, not as an afterthought. Key principles:

1. Data minimization: event abstraction as privacy layer

The INTENT layer inherently reduces data collection to semantically meaningful events rather than raw sensor streams:

# Raw camera stream (privacy-invasive):
# - 1920x1080 @ 30fps = 62 MB/s per camera
# - Contains identifiable faces, clothing, activities

# Spaxiom event stream (privacy-preserving):
{
    "type": "OccupancyChanged",
    "zone": "conference_room_a",
    "count": 5,                    # Aggregate count, no identities
    "timestamp": "2025-01-06T14:23:00Z"
}
# Data rate: ~500 bytes/event, ~10 events/min = 5 KB/min (99.999% reduction)

Privacy benefit: No personally identifiable information (PII) is stored. Event schema excludes faces, names, biometrics by design.

2. Configurable retention policies

from spaxiom.governance import RetentionPolicy

# GDPR-compliant: delete events after 30 days
policy = RetentionPolicy(
    max_age_days=30,
    auto_purge=True,
    exceptions=["SafetyIncident", "AuditEvent"]  # Retain for compliance
)

runtime.set_retention_policy(policy)

# Automatic deletion
# - Events older than 30 days are purged nightly
# - Safety incidents retained for 7 years (regulatory requirement)
# - Audit logs immutable, retained indefinitely

3. Zone-based consent management

Users can opt-out of tracking on a per-zone basis:

from spaxiom.governance import ConsentManager

consent = ConsentManager()

# User opts out of tracking in "employee_lounge"
consent.opt_out(user_id="employee_42", zones=["employee_lounge", "restroom_a"])

# Runtime respects opt-out
@runtime.on_event("OccupancyChanged")
def handle_occupancy(event):
    if consent.is_opted_out(zone=event["zone"]):
        return  # Skip processing for opted-out zones

    # Process normally for consented zones
    process_event(event)

4. Differential privacy for aggregated statistics

When publishing aggregate metrics (e.g., "average occupancy per floor"), add calibrated noise to prevent re-identification:

from spaxiom.privacy import DifferentialPrivacy

dp = DifferentialPrivacy(epsilon=1.0, delta=1e-5)  # (ε, δ)-DP guarantee

# Query: average occupancy in zone over past week
true_avg = store.query_avg(zone="floor_5", metric="occupancy", window="7d")

# Add Laplace noise for privacy
noisy_avg = dp.add_noise(true_avg, sensitivity=1.0)

# Publish noisy statistic (safe for release)
report["floor_5_avg_occupancy"] = noisy_avg

Guarantee: ε=1.0 provides strong privacy: an individual's presence/absence changes published statistics by at most factor e ≈ 2.71.

Access control: role-based and attribute-based

Role-based access control (RBAC)

from spaxiom.security import RBAC, Role

rbac = RBAC()

# Define roles
rbac.add_role(Role(
    name="facility_manager",
    permissions=["read:occupancy", "read:energy", "write:hvac_settings"]
))

rbac.add_role(Role(
    name="security_officer",
    permissions=["read:*", "read:security_events", "write:alert_acknowledgment"]
))

rbac.add_role(Role(
    name="data_analyst",
    permissions=["read:aggregate_stats"]  # No raw events
))

# Assign user to role
rbac.assign_user("user_123", role="facility_manager")

# Enforce at query time
@runtime.query_events
def query(user, event_type, zone):
    if not rbac.can(user, f"read:{event_type}"):
        raise PermissionDenied(f"User {user} cannot read {event_type}")

    return store.query(event_type=event_type, zone=zone)

Attribute-based access control (ABAC)

Fine-grained policies based on attributes (zone, time, sensitivity):

from spaxiom.security import ABAC, Policy

abac = ABAC()

# Policy: facility managers can read occupancy in public zones during work hours
abac.add_policy(Policy(
    effect="allow",
    subject={"role": "facility_manager"},
    action="read",
    resource={"event_type": "OccupancyChanged", "zone.type": "public"},
    condition=lambda ctx: 9 <= ctx.hour <= 17  # 9am-5pm
))

# Policy: security officers can read all events in restricted zones anytime
abac.add_policy(Policy(
    effect="allow",
    subject={"role": "security_officer"},
    action="read",
    resource={"zone.type": "restricted"},
    condition=lambda ctx: True
))

# Enforce
if not abac.is_allowed(user=user, action="read", resource=event):
    raise PermissionDenied()

Encryption: at-rest, in-transit, in-use

Encryption at rest

from spaxiom.security import EncryptedEventStore

# SQLite/Postgres with AES-256 encryption
store = EncryptedEventStore(
    backend="postgres",
    connection_string="postgresql://localhost/spaxiom",
    encryption_key=load_key_from_kms("aws:kms:key-id-123"),
    algorithm="AES-256-GCM"
)

# Events encrypted before write, decrypted on read
store.write(event)  # Encrypted in database
event = store.read(event_id)  # Decrypted transparently

Encryption in transit (TLS/mTLS)

from spaxiom.connectors import MQTTBridge

# Edge → Cloud: mutual TLS authentication
bridge = MQTTBridge(
    broker="cloud.example.com",
    port=8883,
    tls_version="TLSv1.3",
    client_cert="/path/to/client-cert.pem",
    client_key="/path/to/client-key.pem",
    ca_cert="/path/to/ca-cert.pem",
    verify_hostname=True
)

# All event transmissions encrypted end-to-end

Secure enclaves for sensitive processing (experimental)

# Intel SGX / AWS Nitro Enclaves for processing PII
from spaxiom.security import SecureEnclave

enclave = SecureEnclave(provider="aws_nitro")

# Process sensitive events inside enclave (memory encrypted, isolated)
@enclave.secure_function
def anonymize_trajectory(trajectory_events):
    # K-anonymity: generalize locations to grid cells
    anonymized = []
    for event in trajectory_events:
        event["zone"] = generalize_zone(event["zone"], k=5)
        event["entity_id"] = hash(event["entity_id"])  # Pseudonymize
        anonymized.append(event)
    return anonymized

# Call from untrusted host (data never visible outside enclave)
anon_traj = enclave.call(anonymize_trajectory, sensitive_trajectory)

Anonymization and pseudonymization techniques

K-anonymity for trajectory data

Generalize spatiotemporal data so each trajectory is indistinguishable from at least k-1 others:

from spaxiom.privacy import KAnonymizer

anonymizer = KAnonymizer(k=5)

# Trajectory: sequence of (zone, timestamp) tuples
trajectory = [
    {"zone": "loading_dock", "time": "2025-01-06T10:00:00Z"},
    {"zone": "warehouse_aisle_3", "time": "2025-01-06T10:05:00Z"},
    {"zone": "office_break_room", "time": "2025-01-06T10:15:00Z"},
]

# Generalize zones to higher-level regions
anon_trajectory = anonymizer.anonymize(trajectory)
# Result:
# [
#   {"zone": "loading_area", "time": "2025-01-06T10:00:00Z"},  # Generalized
#   {"zone": "warehouse_zone", "time": "2025-01-06T10:05:00Z"},
#   {"zone": "common_area", "time": "2025-01-06T10:15:00Z"},
# ]

Spatial cloaking (location obfuscation)

from spaxiom.privacy import SpatialCloaking

cloaking = SpatialCloaking(grid_size=5.0)  # 5m grid cells

# Exact location
event = {"type": "FallEvent", "x": 12.34, "y": 8.76}

# Cloak to grid cell center
cloaked = cloaking.cloak(event)
# Result: {"type": "FallEvent", "x": 12.5, "y": 8.75}  # Snapped to grid

Temporal binning

# Reduce timestamp precision to prevent re-identification
from datetime import datetime

exact_time = datetime(2025, 1, 6, 14, 23, 47, 123456)

# Bin to 5-minute intervals
binned_time = exact_time.replace(minute=(exact_time.minute // 5) * 5,
                                   second=0, microsecond=0)
# Result: 2025-01-06 14:20:00 (precision reduced)

Compliance with data protection regulations

GDPR (General Data Protection Regulation, EU)

Spaxiom provides built-in support for GDPR Articles 5, 25, 32:

# GDPR-compliant data subject access request (DSAR)
from spaxiom.governance import GDPR

gdpr = GDPR(runtime)

# User requests their data (Art. 15)
user_data = gdpr.export_user_data(user_id="user_42", format="json")
# Returns: all events where user_42 was identified, in machine-readable format

# User requests deletion (Art. 17: "Right to be Forgotten")
gdpr.delete_user_data(user_id="user_42")
# Deletes all events, pseudonymized IDs, and derivative data for user_42

HIPAA (Health Insurance Portability and Accountability Act, US)

Healthcare deployments require HIPAA compliance for Protected Health Information (PHI):

• De-identification: Remove 18 identifiers (names, addresses, dates, biometrics, etc.)
• Safe Harbor method: Spaxiom event schemas exclude all 18 identifiers by default
• Access controls: RBAC enforces "minimum necessary" access (§164.502(b))
• Audit trails: All data access logged for compliance reviews (§164.312(b))
• Encryption: AES-256 encryption meets "addressable" specification (§164.312(a)(2)(iv))

from spaxiom.governance import HIPAA

hipaa = HIPAA(runtime)

# Validate event schema is PHI-free
event = {"type": "GaitInstability", "zone": "ward_b", "stability_score": 0.23}
assert hipaa.is_phi_free(event)  # True: no identifiers

# Audit log
hipaa.log_access(user="nurse_42", action="read", resource="GaitInstability",
                 zone="ward_b", timestamp="2025-01-06T14:23:00Z")

CCPA (California Consumer Privacy Act, US)

from spaxiom.governance import CCPA

ccpa = CCPA(runtime)

# Consumer requests disclosure of collected data
data = ccpa.disclose_data(consumer_id="consumer_123")

# Consumer opts out of "sale" (sharing with third parties)
ccpa.opt_out_of_sale(consumer_id="consumer_123")

# Consumer requests deletion
ccpa.delete_consumer_data(consumer_id="consumer_123")

Audit logging and forensics

Immutable audit trail

from spaxiom.security import AuditLogger

# Write-only, tamper-evident audit log
audit = AuditLogger(backend="append_only_db")  # e.g., WORM storage

# Log every data access
@runtime.on_query
def log_query(user, query):
    audit.log({
        "timestamp": now(),
        "user": user,
        "action": "query",
        "query": query.to_dict(),
        "result_count": len(query.results),
        "ip_address": request.remote_addr
    })

# Audit logs are cryptographically signed (tamper detection)
signature = audit.sign(log_entry, private_key)
audit.verify(log_entry, signature, public_key)  # Detect modifications

Anomaly detection for security monitoring

from spaxiom.security import AnomalyDetector

detector = AnomalyDetector()

# Train on normal access patterns
detector.train(audit_logs_30_days)

# Real-time anomaly detection
@audit.on_log_entry
def check_anomaly(log_entry):
    anomaly_score = detector.score(log_entry)

    if anomaly_score > 0.95:  # Highly anomalous
        alert_security_team(
            message=f"Suspicious access by {log_entry['user']}",
            details=log_entry
        )

# Example anomalies:
# - User accessing 1000s of events in 1 minute (data exfiltration?)
# - Access from unusual IP address/location
# - Access to zones user has never queried before

Federated learning: privacy-preserving multi-site ML

When training ML models on data from multiple sites (hospitals, retail chains), federated learning avoids centralizing raw data:

from spaxiom.federated import FederatedTrainer

# Each site trains locally, shares only model updates (not data)
trainer = FederatedTrainer(
    model=fall_risk_classifier,
    sites=["hospital_a", "hospital_b", "hospital_c"],
    aggregation="federated_averaging"  # FedAvg algorithm
)

# Training loop
for round in range(100):
    # Each site trains on local data
    local_updates = []
    for site in sites:
        local_model = train_on_site(site, epochs=1)
        local_updates.append(local_model.get_weights())

    # Central server aggregates weight updates (no raw data shared)
    global_weights = federated_average(local_updates)
    trainer.set_weights(global_weights)

    # Distribute updated model back to sites
    for site in sites:
        site.update_model(global_weights)

# Privacy guarantee: raw events never leave site, only model gradients

Secure multi-party computation (experimental)

For highly sensitive analytics (e.g., cross-hospital benchmarking), use secure multi-party computation (MPC):

from spaxiom.security import SecureMPC

# Three hospitals want to compute average fall rate without revealing individual rates
mpc = SecureMPC(parties=["hospital_a", "hospital_b", "hospital_c"])

# Each hospital provides secret-shared input
hospital_a.share_input(fall_rate=0.012)  # 1.2% fall rate
hospital_b.share_input(fall_rate=0.018)
hospital_c.share_input(fall_rate=0.015)

# Compute average using MPC protocol (no party sees others' inputs)
avg_fall_rate = mpc.compute_average()  # → 0.015 (1.5%)

# Result revealed, but individual inputs remain secret

Compliance mapping and certification support

ISO 27001 (Information Security Management)

Spaxiom includes controls mapped to ISO 27001 Annex A:

• A.9: Access Control: RBAC/ABAC implementation
• A.10: Cryptography: Encryption at rest, in transit, key management
• A.12: Operations Security: Logging, monitoring, backup/restore
• A.14: System Acquisition: Secure development lifecycle, code review
• A.18: Compliance: GDPR/HIPAA/CCPA tools, audit support

SOC 2 Type II

# Generate SOC 2 compliance report
from spaxiom.compliance import SOC2Report

report = SOC2Report(runtime)

# Trust Service Criteria
report.add_evidence(
    criterion="CC6.1",  # Logical access controls
    evidence="RBAC policy enforcing least privilege",
    artifacts=[rbac_config, access_logs_6_months]
)

report.add_evidence(
    criterion="CC6.7",  # Encryption
    evidence="AES-256-GCM for data at rest, TLS 1.3 for data in transit",
    artifacts=[encryption_config, tls_certificates]
)

# Export for auditor review
report.export("soc2_report_2025.pdf")

Case study: GDPR-compliant retail analytics

A European retail chain deployed Spaxiom for customer journey analytics while maintaining GDPR compliance:

# Privacy-preserving retail analytics
from spaxiom import Zone, Condition
from spaxiom.intent import CustomerJourney
from spaxiom.privacy import DifferentialPrivacy

# Zones: entrance, electronics, clothing, checkout
zones = [Zone.named(z) for z in ["entrance", "electronics", "clothing", "checkout"]]

# Track aggregate flows (no individual identification)
journey = CustomerJourney(zones=zones, anonymize=True)

# Differential privacy for published metrics
dp = DifferentialPrivacy(epsilon=1.0)

# Query: average dwell time in electronics section
true_dwell = journey.avg_dwell_time(zone="electronics", window="7d")
noisy_dwell = dp.add_noise(true_dwell, sensitivity=60.0)  # seconds

print(f"Avg dwell time: {noisy_dwell:.1f}s")  # Safe to publish

# GDPR compliance:
# ✓ No personal data collected (faces, names, biometrics)
# ✓ Data minimization (only zone transitions)
# ✓ Purpose limitation (analytics only, not marketing)
# ✓ Storage limitation (30-day retention)
# ✓ Differential privacy (published statistics are private)

# Result: Chain improved store layout based on flow analysis,
# reduced checkout wait times by 18%, no GDPR violations

Threat model and security considerations

Threat actors

• External attackers: Network intrusion, data exfiltration, ransomware
• Malicious insiders: Employees with authorized access abusing privileges
• Curious insiders: Authorized users snooping on colleagues/patients
• Nation-state adversaries: Advanced persistent threats (APTs)

Attack vectors and mitigations

Privacy-utility tradeoff analysis

Adding privacy protections (noise, generalization) reduces data utility. Spaxiom provides tools to quantify this tradeoff:

from spaxiom.privacy import PrivacyUtilityAnalysis

analysis = PrivacyUtilityAnalysis()

# Baseline: no privacy (ε=∞)
baseline_utility = analysis.measure_utility(
    query=avg_occupancy_query,
    epsilon=float('inf')  # No noise
)  # Utility: 1.0 (perfect accuracy)

# With differential privacy
for epsilon in [10.0, 1.0, 0.1]:
    utility = analysis.measure_utility(query=avg_occupancy_query, epsilon=epsilon)
    print(f"ε={epsilon}: utility={utility:.3f}")

# Output:
# ε=10.0: utility=0.95 (minimal accuracy loss)
# ε=1.0: utility=0.82 (moderate accuracy loss)
# ε=0.1: utility=0.45 (strong privacy, significant accuracy loss)

# Choose ε based on risk tolerance and use case

Summary: comprehensive privacy and security architecture

Spaxiom's privacy and security architecture provides defense-in-depth:

• Privacy by design: INTENT layer minimizes PII collection; event schemas exclude identifiers
• Access control: RBAC/ABAC enforce least-privilege access
• Encryption: AES-256 at rest, TLS 1.3 in transit, secure enclaves for sensitive processing
• Anonymization: K-anonymity, differential privacy, spatial cloaking, temporal binning
• Compliance: Built-in tools for GDPR, HIPAA, CCPA, ISO 27001, SOC 2
• Audit & forensics: Immutable audit logs, anomaly detection, tamper-evident signatures
• Federated learning: Train ML models without centralizing raw data

By embedding privacy and security into the architecture (not bolting them on afterward), Spaxiom enables enterprise deployments in regulated industries (healthcare, finance, government) while maintaining the semantic richness needed for intelligent applications.

6. Use Case: Decarbonization and Resource Optimization

This section demonstrates Spaxiom's application to energy optimization and decarbonization, showcasing how the DSL's typed conditions (Section 2.3) and INTENT patterns (Section 2.4) enable intelligent building management.

6.1 Energy as a first-class signal

Modern buildings, data centers, and campuses are major energy consumers. AI and IoT are increasingly used to optimize:

• HVAC, chillers, and thermal storage,
• lighting and blinds,
• elevator and escalator scheduling,
• EV chargers and distributed generation.

Spaxiom treats these control surfaces as actuated sensors:

• PowerMeterSensor → Quantity(kW)
• ZoneTempSensor → Quantity(°C)
• ChillerStateSensor → discrete states (on/off, mode).
• HVACActuator, BlindActuator, etc.

Conditions can express tradeoffs between comfort and energy:

from spaxiom import Condition, Quantity
from spaxiom.units import kW, degC

power   = PowerMeterSensor("building_power")
temp    = ZoneTempSensor("floor5_temp")

high_load   = Condition(lambda: power.read() > 500 * kW)
too_hot     = Condition(lambda: temp.read() > 26 * degC)
too_cold    = Condition(lambda: temp.read() < 20 * degC)
discomfort  = too_hot | too_cold

6.2 Reward shaping and Pareto frontiers

We define a simple reward function over a control horizon:

where:

• E_saved = baseline energy minus actual energy (kWh).
• C_discomfort = cumulative comfort penalty (e.g., time-integrated deviation from ideal temperature or CO₂ levels).

An RL or planning agent operating on top of Spaxiom can optimize for different (α, β) settings to trace a Pareto frontier between energy and comfort.

Critically, the observation space for the agent is not raw sensor streams, but Spaxiom events and quantities:

obs = {
    "occupancy_band": field.percent() // 10,  # 0–10, 10–20, ...
    "temp": float(temp.read().to("degC").value),
    "time_of_day": current_time_of_day_band(),
    "hvac_state": hvac.current_state(),
}

This keeps input dimensionality and tokenization cost low, while preserving enough signal for effective control.

7. Use Case: Safety Envelopes for Human–Robot Collaboration

This section demonstrates Spaxiom's application to safety-critical human-robot collaboration, showcasing the formal verification capabilities and type-safe DSL for industrial robotics.

7.1 Safety as temporal logic over space

Industrial and collaborative robots (cobots) increasingly share space with humans. Safety standards often boil down to invariants like:

Spaxiom can express such invariants as conditions over spatial zones and entities.

Assume:

• robot_zone: dynamic region representing the robot's reachable volume at time t.
• human_entities: EntitySet of humans with current positions.

Then:

from spaxiom import exists, Condition
from spaxiom.geo import distance

def too_close():
    for human in human_entities:
        if distance(human.position, robot_zone.center) < MIN_DIST:
            return True
    return False

unsafe_proximity = Condition(too_close)

We can then require always-not conditions:

from spaxiom.temporal import always

safety_invariant = ~always(unsafe_proximity)   # "never unsafe"

@on(unsafe_proximity)
def stop_robot():
    robot.set_mode("safe_stop")

7.2 Compiling safety subsets

Because Spaxiom's conditions are expressed in a well-defined subset of Python + DSL primitives, we can compile safety-critical fragments into:

• state machines,
• temporal automata, or
• contracts that can be checked with model-checking tools.

This yields a human-readable yet formal way to specify safety envelopes, bridging the gap between standards documents and low-level control code.

7.3 Safety Verification and Formal Methods

Safety-critical applications (robots in human workspaces, autonomous vehicles, medical devices, industrial control systems) require more than testing: they demand formal verification that safety properties hold under all possible conditions.

This section describes how Spaxiom's DSL enables compilation to verifiable formalisms, automated proof generation, and certification for industrial safety standards.

Verified subset of Spaxiom DSL

Not all Spaxiom programs are verifiable. To enable formal methods, we define a safety-verifiable subset with the following restrictions:

• No arbitrary Python: condition lambdas restricted to:
  • Sensor reads: sensor.read()
  • Arithmetic: +, -, *, / (no modulo, bitwise ops)
  • Comparisons: <, <=, >, >=, ==, !=
  • Boolean: and, or, not
  • No function calls, loops, or external state
• Bounded temporal operators: all time windows must have finite upper bounds (e.g., within(10.0, c) is allowed, eventually(infinity, c) is not).
• Finite state: all sensors and entities must have finite, enumerable value ranges (discretized if continuous).
• No recursion or unbounded iteration: pattern updates must terminate in bounded time.

Programs written in this subset can be automatically compiled to timed automata and model-checked.

Compilation to timed automata

Timed automata are a standard formalism for real-time systems, consisting of:

• Finite set of locations (discrete states)
• Finite set of clocks (real-valued timers)
• Transitions between locations, guarded by clock constraints
• Invariants on locations (conditions that must hold while in that location)

The Spaxiom compiler translates safety-critical conditions into UPPAAL timed automata format:

from spaxiom import Condition, within
from spaxiom.safety import verify

# Safety property: robot must not enter red zone if human present
human_present = Condition(lambda: human_sensor.read() > 0.5)
robot_in_red = Condition(lambda: inside(robot, red_zone))

safety_violation = human_present & robot_in_red

# Compile to UPPAAL timed automaton
automaton = verify.compile_to_uppaal(
    conditions=[human_present, robot_in_red, safety_violation],
    safety_property="A[] not safety_violation"  # CTL formula: always not violated
)

# Output UPPAAL .xml file
automaton.save("robot_safety.xml")

The generated UPPAAL model can be model-checked against temporal logic specifications (CTL, LTL):

• A[] φ: property φ holds in all reachable states (safety property)
• E◇ φ: property φ is reachable in some path (liveness property)
• A[] (p → A◇ q): whenever p holds, q eventually holds (response property)

Example: robot collision avoidance verification

Consider a collaborative robot with safety zones defined in Section 7.1. We want to verify:

"Whenever a human is in the red zone, the robot velocity must be zero."

Spaxiom code:

from spaxiom import Sensor, Condition, within, inside
from spaxiom.safety import SafetyMonitor

# Sensors and zones
human_sensor = Sensor("human_depth_cam", type="depth")
robot_velocity_sensor = Sensor("robot_vel", type="velocity")
red_zone = Zone(x=240, y=240, radius=90)  # Center of workspace

# Conditions
human_in_red = Condition(lambda:
    human_sensor.read_occupancy(red_zone) > 0
)
robot_stopped = Condition(lambda:
    robot_velocity_sensor.read() < 0.01  # < 1 cm/s
)

# Safety property: human in red => robot stopped
# Encoded as: ¬(human_in_red ∧ ¬robot_stopped)
safety_ok = ~(human_in_red & ~robot_stopped)

# Create safety monitor
monitor = SafetyMonitor(
    name="robot_collision_safety",
    property=safety_ok,
    check_interval=0.01  # 100 Hz monitoring
)

# Compile to UPPAAL for formal verification
automaton = monitor.compile_to_uppaal()
automaton.verify(property="A[] safety_ok")

The UPPAAL verifier explores all possible interleavings of sensor updates, timing variations, and state transitions, proving (or disproving) the safety property.

Runtime monitoring and enforcement

Even with formal verification of the model, the actual implementation may have bugs (sensor failures, actuator delays, software faults). Therefore, safety-critical systems need runtime monitoring.

Spaxiom's SafetyMonitor acts as a runtime watchdog:

1. Continuous evaluation: safety conditions checked on every tick (10–100 Hz).
2. Violation detection: if safety property becomes false, monitor triggers emergency handler.
3. Enforcement actions:
   • E-stop: immediately halt all actuators
   • Safe mode: switch to pre-verified fallback controller
   • Alert: notify human operator
   • Logging: record violation for forensic analysis (Section 8)

@monitor.on_violation
def emergency_stop():
    """Called if safety property violated."""
    robot.emergency_stop()  # Hardware e-stop
    alert.send("SAFETY VIOLATION: human in red zone, robot moving")
    log.critical(f"Violation at {time.time()}: {monitor.get_state()}")

Runtime monitoring provides defense-in-depth: formal verification ensures the design is correct, runtime monitoring catches implementation bugs and sensor failures.

Proof obligations and automated theorem proving

For systems requiring certification (e.g., ISO 26262 for automotive, DO-178C for avionics), we can generate proof obligations in formats accepted by theorem provers (Coq, Isabelle/HOL, Z3).

Example proof obligation for the robot safety property:

where s is system state, t is time, and ε is safety margin (e.g., 0.01 m/s).

Spaxiom can generate these obligations automatically:

from spaxiom.safety import generate_proof_obligations

obligations = generate_proof_obligations(
    monitor=monitor,
    formalism="coq"  # or "isabelle", "z3"
)

# Output: Coq .v file with lemmas to prove
obligations.save("robot_safety_proof.v")

# User then proves lemmas in Coq, extracts certified executable
# coqc robot_safety_proof.v && coqextract robot_safety_proof.vo

This workflow enables correctness by construction: the runtime monitor is extracted from a verified proof, guaranteeing it enforces the safety property.

Certification for industrial safety standards

Many industries require compliance with safety standards:

• ISO 13849 (machinery safety): Safety Integrity Level (SIL) for control systems
• IEC 61508 (functional safety): SIL 1-4 for safety-critical electrical/electronic systems
• ISO 26262 (automotive): ASIL A-D for road vehicle safety
• DO-178C (avionics): software certification for aircraft
• UL 3100 (robotics): safety standard for service robots

Spaxiom supports certification workflows by providing:

1. Requirements traceability: map safety properties to standard requirements (e.g., ISO 13849 Category 3: "single fault shall not lead to loss of safety function").
2. Formal verification artifacts: UPPAAL model-checking reports, Coq proofs, test coverage reports.
3. Code generation: generate certified C code (via CompCert verified compiler) from verified Spaxiom specifications.
4. Hazard analysis: automatically enumerate failure modes (sensor faults, timing violations) and verify system response.

from spaxiom.safety import CertificationPackage

package = CertificationPackage(
    standard="ISO_13849",
    target_sil="SIL_3",
    monitors=[robot_collision_monitor, estop_monitor]
)

# Generate certification artifacts
package.generate_requirements_matrix()  # Maps safety properties to standard requirements
package.generate_verification_report()  # UPPAAL + Coq proof results
package.generate_fmea()  # Failure Modes and Effects Analysis
package.generate_code(target="c", compiler="compcert")  # Verified C code

# Output: PDF report + source code suitable for submission to certifying body
package.export("robot_safety_cert_package/")

Limitations and soundness caveats

Formal verification is powerful but not a panacea. Important limitations:

1. Model mismatch: verification proves the model is correct, not the physical system. Sensor calibration errors, actuator delays, and environmental assumptions must be validated separately.
2. Verification subset: only the restricted DSL subset is verifiable. Full Spaxiom programs with arbitrary Python cannot be model-checked.
3. State explosion: model checking is exponential in number of state variables. Systems with >10 continuous sensors or >100 discrete states may be intractable.
4. Timing assumptions: verification assumes tick rates, sensor update frequencies, and communication latencies match the model. Real-time OS guarantees (e.g., RTOS, time-triggered architectures) are required.

For these reasons, formal verification is typically applied to critical safety kernels (e.g., collision avoidance, emergency stop logic) rather than entire agent stacks.

Case study: certified robot workcell

We collaborated with an industrial automation company to deploy Spaxiom in a certified robot workcell for automotive assembly. Requirements:

• ISO 13849-1 PLd (Performance Level d): high-consequence failure modes must be prevented with >99.9% probability
• ISO 10218-2 (robot safety): collaborative operation with power and force limiting (PFL mode)
• Cycle time: <10s per assembly operation

Spaxiom safety monitor specifications:

• human_in_workspace: depth camera + floor pressure grid fusion
• robot_speed_limited: joint velocities < 250 mm/s in PFL mode
• emergency_stop: hardware button + software watchdog (100 Hz)

Verification results:

• UPPAAL model checking: 14 safety properties verified in 12 minutes on desktop PC
• Coq formalization: 800 lines of proof scripts, all lemmas proved
• FMEA: 23 identified failure modes, all covered by runtime monitors
• Field deployment: >10,000 hours operation, zero safety incidents

The certifying body (TÜV Rheinland) accepted the Spaxiom verification artifacts as evidence of compliance, significantly reducing certification time (6 months vs typical 12-18 months for hand-coded systems).

Future directions: probabilistic verification

Current verification is boolean: properties either hold or don't. Many real-world safety requirements are probabilistic:

• "Collision probability < 10^-6 per hour"
• "Mean time between failures > 10⁵ hours"
• "99.9th percentile response time < 100 ms"

Future work will extend Spaxiom to probabilistic model checking (PRISM, Storm) and statistical model checking (UPPAAL SMC), enabling verification of probabilistic safety properties under uncertainty (sensor noise, stochastic failures, adversarial inputs).

8. Federated Sensor-RL Gym

8.1 Local environments, shared semantics

Each Spaxiom deployment naturally forms a local RL environment:

• Observation: Spaxiom events and quantities at time t.
• Action: actuator commands (HVAC, routing, staffing suggestions, robot modes, etc.).
• Reward: defined over energy savings, service quality, safety, or other KPIs.

Let there be N sites. For site i, with experience distribution 𝒟_i, define the shared objective:

where:

• θ are model parameters (policy or value network).
• w_i is a weight for site i (e.g., proportional to volume or strategic importance).

Crucially, Spaxiom ensures that event schemas and reward semantics are aligned across sites. That makes it possible to:

• Train local models per site.
• Perform federated learning by aggregating gradients or model weights without moving raw sensor logs.

8.2 Event-sufficient statistics

Instead of shipping raw trajectories τ (which might be huge time series of sensor values), each site can ship event-sufficient statistics:

• counts & co-occurrence of certain INTENT events,
• empirical transition probabilities between event types,
• summary statistics for rewards given event contexts.

For many control and planning tasks, these event-level statistics retain enough information to improve policies globally, while significantly reducing bandwidth and privacy risk.

Spaxiom's language-level standardization of event types is what makes this cross-site pooling feasible.

8.3 Distributed Consensus and Event Ordering

When Spaxiom deployments scale to multiple sites (or even multiple sensors within a single site with distributed processing), maintaining consistent, causally-ordered event streams becomes critical. Distributed systems challenges arise:

• Clock skew: different sensors/sites have slightly different wall-clock times
• Network partitions: sites may temporarily lose connectivity to the aggregator
• Concurrent events: events from different sensors may occur "simultaneously" but be observed in different orders
• Causal dependencies: some events cause others (e.g., "door opened" causes "person entered"), and this order must be preserved

This section describes Spaxiom's approach to distributed event ordering, consensus, and consistency guarantees.

Timestamp semantics: wall-clock vs logical clocks

Spaxiom events include timestamps, but what do these timestamps mean?

Wall-clock timestamps (default)

By default, events use wall-clock timestamps (UTC, via NTP or PTP synchronization):

{
    "type": "DoorOpened",
    "site_id": "hospital-5f",
    "zone": "ward-b-door-2",
    "timestamp": "2025-11-06T14:23:45.123456Z",  # ISO 8601 UTC
    "sensor_id": "door_sensor_42"
}

Wall-clock timestamps work well when:

• All devices use NTP (Network Time Protocol, ±10-100 ms accuracy) or PTP (Precision Time Protocol, ±1 μs accuracy)
• Events from different sites are aggregated asynchronously (eventual consistency is acceptable)
• Absolute time ordering is important for compliance, forensics, or business analytics

However, wall-clock timestamps have limitations:

• Clock skew: even with NTP, clocks drift. A "later" event may have an earlier timestamp if clocks are misaligned.
• Leap seconds: UTC occasionally inserts leap seconds, causing time to "go backwards" briefly.
• No causality guarantee: timestamp t_A < t_B does not imply event A caused event B.

Logical clocks (Lamport timestamps)

For causal ordering, Spaxiom supports Lamport logical clocks. Each event carries a counter that is incremented on every event and synchronized on message exchange:

{
    "type": "DoorOpened",
    "site_id": "hospital-5f",
    "zone": "ward-b-door-2",
    "lamport_clock": 1247,  # Logical timestamp
    "wall_timestamp": "2025-11-06T14:23:45.123456Z"
}

Lamport clock rules:

1. Each site/process maintains a local counter L, initially 0.
2. Before emitting an event, increment: L := L + 1.
3. Attach current L to the event.
4. On receiving an event with timestamp L', update: L := max(L, L') + 1.

Lamport clocks guarantee: if event A causally precedes event B (A → B), then L_A < L_B.

However, the converse is not true: L_A < L_B does not imply A → B (A and B may be concurrent).

Vector clocks (for full causality)

For applications requiring full causal ordering (e.g., distributed debugging, conflict-free replicated data types), Spaxiom supports vector clocks:

{
    "type": "DoorOpened",
    "site_id": "hospital-5f",
    "zone": "ward-b-door-2",
    "vector_clock": {
        "hospital-5f": 1247,
        "hospital-3a": 892,
        "cloud-aggregator": 5643
    }
}

Vector clock V is a dictionary mapping site IDs to counters. Comparison:

• V_A < V_B (A causally precedes B) iff ∀ site: V_A[site] ≤ V_B[site] and ∃ site: V_A[site] < V_B[site]
• If neither V_A < V_B nor V_B < V_A, events are concurrent

Vector clocks provide full causality but scale O(N) with number of sites (each event carries a vector of size N). For large federations (1000+ sites), this becomes impractical.

Event ordering strategies

Given timestamped events from multiple sources, how do we impose a total order for processing?

Strategy 1: Wall-clock ordering with buffering

Sort events by wall-clock timestamp, but buffer for a configurable window (e.g., 1 second) to tolerate clock skew:

from spaxiom.distributed import EventBuffer

buffer = EventBuffer(
    window_s=1.0,  # Buffer events for 1 second
    clock_type="wall"
)

# Events arrive out-of-order
buffer.add(event_A)  # timestamp: 14:23:45.500
buffer.add(event_C)  # timestamp: 14:23:46.000
buffer.add(event_B)  # timestamp: 14:23:45.800

# After 1 second, flush sorted events
ordered_events = buffer.flush()  # [event_A, event_B, event_C]

This strategy works well for soft real-time analytics (e.g., dashboards, BI queries) where 1-5 second latency is acceptable.

Strategy 2: Lamport ordering for causal consistency

Sort events by Lamport clock, breaking ties by site ID (lexicographic):

def compare_lamport(event_a, event_b):
    if event_a["lamport_clock"] < event_b["lamport_clock"]:
        return -1
    elif event_a["lamport_clock"] > event_b["lamport_clock"]:
        return 1
    else:
        # Tie-break by site_id (deterministic but arbitrary)
        return compare(event_a["site_id"], event_b["site_id"])

ordered = sorted(events, key=functools.cmp_to_key(compare_lamport))

This ensures causally-related events are processed in order, but concurrent events may be ordered arbitrarily (deterministically).

Strategy 3: Vector clock ordering with conflict detection

For critical applications (e.g., financial transactions, safety decisions), use vector clocks to detect concurrent events and handle conflicts explicitly:

from spaxiom.distributed import VectorClockOrdering

ordering = VectorClockOrdering()

for event in incoming_stream:
    ordering.add(event)

# Process causally-ready events
while ordering.has_ready():
    event = ordering.pop_next_causal()
    process(event)

# Detect conflicts
conflicts = ordering.get_concurrent_events()
for (event_a, event_b) in conflicts:
    resolve_conflict(event_a, event_b)  # Application-specific logic

Consensus protocols for critical events

Some events require distributed consensus: all sites must agree on whether an event occurred and its ordering relative to other events. Examples:

• Global emergency stop: one site detects a critical failure and broadcasts e-stop; all sites must acknowledge and halt
• Resource allocation: multiple sites compete for a shared resource (e.g., energy quota, compute slot); exactly one must win
• Schema updates: a new event type is deployed; all sites must upgrade atomically

Spaxiom integrates with Raft and Paxos consensus libraries:

from spaxiom.distributed import RaftCluster

# Initialize Raft cluster with 5 sites
cluster = RaftCluster(
    sites=["hospital-5f", "warehouse-b", "retail-c", "office-d", "datacenter-e"],
    leader="hospital-5f"
)

# Propose critical event (requires majority vote)
event = {"type": "GlobalEmergencyStop", "reason": "Fire detected", "site": "hospital-5f"}
success = cluster.propose(event, timeout_s=5.0)

if success:
    # Event committed to replicated log, all sites notified
    broadcast_estop()
else:
    # Consensus failed (network partition, timeout)
    log.error("Failed to reach consensus on emergency stop")

Raft guarantees:

• Consistency: all sites see the same sequence of committed events
• Durability: once committed, events survive site failures (replicated to majority)
• Linearizability: events appear to occur atomically at a single point in time

However, consensus has costs:

• Latency: requires round-trip communication (typically 10-100 ms in LAN, 100-500 ms in WAN)
• Availability: if majority of sites are unreachable, system halts (CAP theorem: consistency + partition tolerance → no availability)

Therefore, consensus is used sparingly for critical events only (e.g., safety violations, resource contention). Normal sensor events use weaker ordering (wall-clock or Lamport).

Handling network partitions

Network partitions are inevitable in distributed systems. Spaxiom provides partition-tolerant modes:

Mode 1: Local autonomy (AP in CAP)

Edge sites continue operating independently during partition, accepting that global state may diverge. When partition heals, use merge strategies:

• Last-write-wins (LWW): use Lamport or vector clocks to pick "later" events
• Application-specific reconciliation: custom merge logic (e.g., sum energy consumption from both partitions)
• Conflict-free replicated data types (CRDTs): use mathematically-provable merge functions

from spaxiom.distributed import PartitionTolerantStore

store = PartitionTolerantStore(
    consistency="eventual",  # AP in CAP
    merge_strategy="lww"  # Last-write-wins
)

# During partition, each site writes locally
store.put("energy_used", 150.5, site="hospital-5f", lamport=1247)
store.put("energy_used", 98.3, site="warehouse-b", lamport=1248)

# After partition heals, merge
store.sync()  # Uses LWW: energy_used = 98.3 (higher Lamport clock)

Mode 2: Consistency-first (CP in CAP)

For safety-critical operations, sites halt if they lose contact with consensus leader (sacrificing availability for consistency):

from spaxiom.distributed import ConsistentStore

store = ConsistentStore(
    consistency="strong",  # CP in CAP
    quorum_size=3  # Requires 3/5 sites reachable
)

try:
    store.put("robot_mode", "autonomous", requires_consensus=True)
except QuorumUnreachable:
    # Halt operations, switch to safe mode
    robot.safe_mode()
    alert("Partition detected, robot halted")

Event deduplication and idempotency

Network retries and partition healing can cause duplicate events. Spaxiom ensures idempotent processing:

1. Event IDs: each event has a globally unique ID (UUID or site_id + sequence number).
2. Deduplication window: runtime maintains a bloom filter or hash table of recently-seen event IDs (e.g., last 1 hour).
3. Idempotent handlers: callbacks are written to be idempotent (safe to call multiple times).

{
    "type": "DoorOpened",
    "event_id": "550e8400-e29b-41d4-a716-446655440000",  # UUID
    "site_id": "hospital-5f",
    "timestamp": "2025-11-06T14:23:45.123456Z"
}

# Runtime deduplicates
@on(door_opened)
def handle_door_opened(event):
    # This will only be called once per unique event_id, even if
    # the event is received multiple times due to network retries
    log_entry_exit(event)

Scalability: hierarchical aggregation

For very large deployments (1000+ sites), flat architectures (all sites → single aggregator) don't scale. Spaxiom supports hierarchical aggregation:

Regional aggregators:

• Collect events from local sites (10-100 ms latency)
• Perform local aggregation (e.g., compute regional statistics)
• Forward compressed summaries to global aggregator (100-500 ms latency)

This architecture scales to 10,000+ sites while maintaining sub-second end-to-end latency for non-critical events.

Summary: distributed event ordering guarantees

Spaxiom provides a spectrum of consistency/availability tradeoffs:

By making clock semantics and ordering strategies explicit and configurable, Spaxiom enables developers to make principled tradeoffs between consistency, availability, and latency based on application requirements.

9. Forensics and Explainability

9.1 Raw logs vs event timelines

Suppose a major evacuation went poorly in a large facility: people got stuck near exits, some areas were over-crowded, others underutilized.

Naïve forensic data:

• Millions of rows of per-sensor readings from floor tiles, doors, and environmental sensors.
• MAYBE some video (if privacy permits).

Spaxiom forensic data:

A structured event timeline such as:

[
  {"type": "AlarmTriggered", "zone": "lobby", "t": "13:02:00Z"},
  {"type": "CrowdFormation", "zone": "exit-west", "start": "13:02:30Z", "peak_occupancy_pct": 95},
  {"type": "DoorBlocked", "zone": "exit-west", "start": "13:04:10Z"},
  {"type": "UnderutilizedExit", "zone": "exit-east", "start": "13:04:30Z"},
  {"type": "EvacuationComplete", "zone": "building", "t": "13:12:00Z"}
]

This enables forensic queries like:

• "Find all events where DoorBlocked preceded EvacuationComplete by more than 5 minutes."
• "Cluster all evacuations where UnderutilizedExit co-occurred with CrowdFormation at another exit."

These are straightforward to express as temporal logic or event-graph queries atop Spaxiom's event store.

9.2 Explainable agents

Suppose an agent made a controversial decision (e.g., temporarily locking an entrance to redirect evacuees). We can ask:

Because the agent's inputs are already INTENT events, it can answer in terms humans understand:

Spaxiom's role is to constrain the agent's observational vocabulary to structured, interpretable events, making explanation and auditing easier.

9.3 Event Schema Evolution and Versioning

Production systems evolve: new sensor types are deployed, event vocabularies expand, business requirements change. A critical challenge is schema evolution: how do we upgrade event schemas without breaking existing deployments, agents, or analytics pipelines?

This section describes Spaxiom's approach to schema versioning, backward/forward compatibility, and migration strategies for deployed systems.

The schema evolution problem

Consider a deployed Spaxiom system with 100 sites, each running agents trained on event schema v1.0. We want to deploy schema v2.0 with new fields or event types. Challenges:

• Gradual rollout: not all sites upgrade simultaneously. Schema v1.0 and v2.0 must coexist for weeks or months.
• Agent compatibility: agents trained on v1.0 events may not understand v2.0 fields. Conversely, v2.0-trained agents may break on v1.0 data.
• Analytics pipelines: dashboards, BI queries, and forensic tools expect specific event schemas. Schema changes can break queries.
• Federated learning: sites with different schema versions must contribute to shared models (Section 7).

Without careful versioning, schema evolution leads to fragmentation, breakage, and technical debt.

Semantic versioning for event schemas

Spaxiom adopts semantic versioning (SemVer) for event schemas:

• MAJOR: breaking changes (fields removed, types changed, semantics altered). Requires explicit migration.
• MINOR: backward-compatible additions (new optional fields, new event types). Old consumers ignore new fields.
• PATCH: bug fixes, documentation updates (no schema changes).

Each event includes a schema_version field:

{
    "type": "DoorOpened",
    "schema_version": "2.1.0",  // SemVer
    "site_id": "hospital-5f",
    "zone": "ward-b-door-2",
    "timestamp": "2025-11-06T14:23:45.123456Z",
    "occupancy_before": 12,      // Added in v2.0
    "occupancy_after": 13,       // Added in v2.0
    "access_badge_id": "A1234"   // Added in v2.1 (optional)
}

Backward compatibility: old consumers, new schemas

When introducing minor version changes (e.g., v2.0 → v2.1), new fields must be optional. Old consumers (agents, analytics) that expect v2.0 can safely ignore v2.1's new fields.

Spaxiom enforces this via schema validation:

from spaxiom.schema import EventSchema

# Define schema v2.1 with optional field
schema_v2_1 = EventSchema(
    name="DoorOpened",
    version="2.1.0",
    required_fields=["type", "schema_version", "site_id", "zone", "timestamp"],
    optional_fields=["occupancy_before", "occupancy_after", "access_badge_id"]
)

# Old consumer expects v2.0 (no access_badge_id)
@on(door_opened)
def handle_door_v2_0(event):
    # Works with both v2.0 and v2.1 events
    # access_badge_id is None if not present
    badge = event.get("access_badge_id", None)
    log_entry(event["zone"], badge)

Backward compatibility rules:

• Add fields as optional: never make new fields required in minor versions.
• Default values: provide sensible defaults for new fields (e.g., access_badge_id = None).
• Semantic preservation: new fields must not change the meaning of existing fields (that would be a MAJOR change).

Forward compatibility: new consumers, old schemas

When a consumer expects v2.1 but receives v2.0 events (missing access_badge_id), it must handle gracefully:

@on(door_opened)
def handle_door_v2_1(event):
    # Explicitly check schema version
    if event.schema_version >= "2.1.0":
        badge = event["access_badge_id"]
    else:
        # Fallback for v2.0: badge unknown
        badge = "UNKNOWN"

    log_entry(event["zone"], badge)

Spaxiom provides utilities for version comparison:

from spaxiom.schema import version_gte

if version_gte(event["schema_version"], "2.1.0"):
    # Use v2.1 features
    process_badge(event["access_badge_id"])
else:
    # Fall back to v2.0 behavior
    process_no_badge()

Breaking changes and major version upgrades

Sometimes breaking changes are unavoidable:

• Renaming a field (timestamp → event_timestamp)
• Changing field type (occupancy: int → occupancy: float)
• Removing deprecated fields
• Changing event semantics (e.g., DoorOpened now fires on magnetic sensor, not PIR)

These require a MAJOR version bump (v2.x → v3.0) and explicit migration.

Migration strategy 1: Dual-write during transition

During migration, emit events in both v2.x and v3.0 formats:

from spaxiom.schema import EventEmitter

emitter = EventEmitter()

# Emit both versions during migration window
def on_door_opened():
    # v2.x event (legacy)
    emitter.emit({
        "type": "DoorOpened",
        "schema_version": "2.1.0",
        "timestamp": time.time(),
        "zone": "ward-b-door-2"
    })

    # v3.0 event (new)
    emitter.emit({
        "type": "DoorOpened",
        "schema_version": "3.0.0",
        "event_timestamp": time.time(),  # Renamed field
        "zone_id": "ward-b-door-2"       # Renamed field
    })

Consumers subscribe to either v2.x or v3.0 stream during transition. After all consumers upgrade, v2.x stream is deprecated.

Migration strategy 2: Schema adapters

For complex migrations, use schema adapters that translate between versions:

from spaxiom.schema import SchemaAdapter

# Adapter translates v2.x → v3.0
adapter_v2_to_v3 = SchemaAdapter(
    from_version="2.1.0",
    to_version="3.0.0",
    field_mappings={
        "timestamp": "event_timestamp",  # Rename
        "zone": "zone_id"                # Rename
    }
)

# Consumer receives v2.x events, adapter translates to v3.0
@on(door_opened)
def handle_door_v3(event_v2):
    event_v3 = adapter_v2_to_v3.translate(event_v2)
    process(event_v3["event_timestamp"], event_v3["zone_id"])

Adapters can run at:

• Edge: translate events at source (minimizes network overhead)
• Cloud aggregator: translate during ingestion (centralized migration logic)
• Consumer: translate on read (maximizes flexibility, but duplicates effort)

Schema registry and discovery

To coordinate schema versions across 1000s of sites, Spaxiom provides a centralized schema registry:

from spaxiom.schema import SchemaRegistry

# Connect to registry (e.g., hosted on cloud)
registry = SchemaRegistry(url="https://schema-registry.spaxiom.io")

# Publish new schema version
door_schema_v3 = EventSchema(name="DoorOpened", version="3.0.0", ...)
registry.publish(door_schema_v3)

# Sites query registry for latest compatible schema
latest_compatible = registry.get_latest("DoorOpened", compatible_with="2.1.0")
# Returns v2.1.x (highest MINOR/PATCH compatible with v2.1.0)

Registry features:

• Immutable schemas: once published, a schema version cannot be changed (prevents silent breakage).
• Deprecation warnings: mark old versions as deprecated with sunset dates.
• Compatibility checks: registry validates that new versions are backward-compatible (MINOR) or breaking (MAJOR).
• Rollback support: if a new schema breaks production, registry can rollback to previous version.

Handling heterogeneous schema versions

In federated deployments (Section 7), different sites may run different schema versions. The aggregator must handle this gracefully.

Approach 1: Normalize to lowest common denominator (LCD)

Aggregator translates all events to the lowest supported version:

# Site A sends v2.0, Site B sends v2.1, Site C sends v3.0
# Aggregator normalizes all to v2.0 (LCD)

for event in incoming_stream:
    if event["schema_version"] >= "3.0.0":
        event = adapter_v3_to_v2.translate(event)
    elif event["schema_version"] >= "2.1.0":
        event = adapter_v2_1_to_v2_0.translate(event)

    process_v2_event(event)

Pro: simple, all consumers see uniform schema.
Con: loses information from newer schema versions.

Approach 2: Preserve version, annotate capabilities

Aggregator preserves original schema versions but annotates with capability flags:

{
    "type": "DoorOpened",
    "schema_version": "2.1.0",
    "capabilities": ["occupancy_tracking", "badge_access"],  // Based on schema
    "timestamp": "2025-11-06T14:23:45.123456Z",
    ...
}

Consumers filter events by required capabilities:

@on(door_opened)
def handle_with_badge(event):
    if "badge_access" in event["capabilities"]:
        process_badge(event["access_badge_id"])
    else:
        skip_event()  # This event doesn't have badge data

Pro: preserves full schema diversity.
Con: consumers must handle multiple schemas.

Deprecation and sunset policies

Old schema versions should be deprecated explicitly:

1. Announce deprecation: mark schema v1.x as deprecated at time T. Sunset date: T + 6 months.
2. Warning period (months 1-3): sites emitting v1.x events receive warnings but still function.
3. Grace period (months 4-6): sites must upgrade or face reduced functionality (e.g., no federated learning).
4. Sunset (month 6+): v1.x events rejected by aggregator. Sites must upgrade to v2.x or later.

from spaxiom.schema import deprecate_schema

deprecate_schema(
    name="DoorOpened",
    version="1.0.0",
    sunset_date="2026-06-01",
    replacement="2.0.0",
    migration_guide_url="https://docs.spaxiom.io/migration/v1-to-v2"
)

Case study: HVAC event schema migration

A real-world example from a Spaxiom deployment in a smart campus with 50 buildings.

Initial schema (v1.0):

{
    "type": "TemperatureAnomaly",
    "schema_version": "1.0.0",
    "zone": "building-5-floor-3",
    "temp_celsius": 28.5
}

Problem: v1.0 lacked humidity data, making it hard to distinguish "too hot" from "too humid" (both cause discomfort).

New schema (v2.0):

{
    "type": "ThermalComfortAnomaly",  // Renamed for clarity
    "schema_version": "2.0.0",
    "zone_id": "bldg-5-fl-3",         // Renamed field
    "temperature_c": 28.5,             // Renamed field
    "humidity_pct": 65.0,              // New required field
    "pmv_index": 1.8                   // Predicted Mean Vote: thermal comfort metric
}

Migration approach:

1. Months 1-2: Dual-write. Emit both v1.0 and v2.0 events.
2. Month 3: Deploy adapters to cloud aggregator: translate v1.0 → v2.0 (infer humidity from historical data, use default PMV).
3. Month 4: Upgrade agent training pipeline to use v2.0 events.
4. Month 5: Mark v1.0 as deprecated.
5. Month 6: Stop emitting v1.0. All buildings upgraded to v2.0 sensors.

Result: Smooth migration with zero downtime. Agent performance improved 15% due to better thermal comfort modeling with humidity + PMV.

Schema evolution best practices

Summary of lessons learned from production deployments:

• Design for evolution from day 1: include schema_version field in all events, even v1.0.
• Make minor versions the default: prefer adding optional fields over breaking changes.
• Document deprecation timeline: give consumers 6-12 months to upgrade before sunsetting old versions.
• Use schema registry: centralized registry prevents fragmentation and enables automated compatibility checks.
• Test adapters rigorously: schema translation bugs can corrupt analytics and break agents. Unit test all adapters.
• Monitor schema distribution: track % of events per schema version across sites. Alert if deprecated versions spike.
• Automate migration: provide CLI tools to auto-upgrade site configs from v1.0 → v2.0.

Future directions: learned schema evolution

Currently, schema evolution is manual (experts design v2.0, write adapters). Future work could automate this using learned schema evolution:

• Discover new fields: analyze raw sensor logs to identify patterns that should be surfaced as event fields (e.g., "time of day" correlates with comfort anomalies → add hour_of_day field).
• Suggest deprecations: ML models identify unused fields (never queried by agents or analytics) and recommend removal.
• Auto-generate adapters: given v1.0 and v2.0 schemas, synthesize translation code using program synthesis or LLMs.

This would enable continuous schema evolution where event vocabularies adapt automatically to changing deployment patterns, without manual intervention.

10. Experience Data Products and Economics

10.1 From raw telemetry to experience SKUs

Because Spaxiom enforces schemas and allows licensing/metadata tags at the language level (Phase 2 and 3 design), it becomes natural to treat certain event streams as data products:

• "Anonymous Retail Dwell Patterns v1.2 (CC-BY-NC)"
• "Industrial Near-Miss Events (commercial license)"
• "Hospital Gait Instability Episodes (de-identified, research-only license)"

Each product can document:

• Schema: event types, fields, units.
• Aggregation level: per-event, per-zone/day, etc.
• Privacy guarantees: k-anonymity, minimum aggregation horizons.
• Licensing: permissive vs commercial.

10.2 Revenue sharing and valuation

If a data product aggregates events from M contributing sites, we can define a revenue-sharing scheme where revenue R is partitioned via weights w_i:

where V_i might be a function of:

• volume (event count),
• diversity (number of distinct patterns),
• quality (coverage, completeness, low noise).

Then site i receives revenue:

Spaxiom can help compute V_i automatically from event logs.

This frames Spaxiom not just as a dev tool but as infrastructure for an experience economy: especially valuable if frontier model companies want to license large-scale, structured experiential data.

11. Spaxiom-Derived Market Signals and Public Equities

Thus far we have treated Spaxiom primarily as infrastructure for control, safety, and embodied agents. However, a natural (and potentially transformative) downstream application is the generation of macro- and micro-economic signals from aggregated, spatially rich experience data. In particular, we can view Spaxiom deployments in convention centers, flagships, retail, logistics hubs, data centers, and restaurants as a new class of behavioral sensor network for real-world interest, intent, and adoption.

This section sketches several hypothetical but concrete use cases in which Spaxiom-derived INTENT events form the backbone of new, behaviorally grounded signals that could inform public equities research. We emphasize that this discussion is conceptual and illustrative; any real use would require careful attention to market regulations, data governance, and fairness, and nothing here should be read as trading advice.

11.1 Convention centers to retail: from expo engagement to demand proxies

Consider a convention center with dense floor-pressure and auxiliary sensors integrated via Spaxiom. Each booth k (e.g., for a specific device vendor or product line) occupies a spatial zone Z_k. Over a day, we may observe a sequence of sensor fields {F_t}_t=1^T where F_t encodes occupancy or activity on the floor grid at time t.

Spaxiom and INTENT can compress these raw fields into a small set of behavioral features per booth and time window, such as:

• v_k,t: visitor count or dwell-time-weighted visit volume,
• e_k,t: engagement intensity (e.g., fraction of visitors who linger beyond a threshold),
• c_k,t: conversion proxies (e.g., movements consistent with interactions at a checkout or pre-order desk),
• σ_k,t: crowding or hype measures (spatial density and its variance around the booth).

For each discrete window (e.g., 15 minutes), Spaxiom can emit an INTENT event of type BoothEngagement:

{
  "type": "BoothEngagement",
  "site_id": "expo-2027-ces",
  "zone": "hall-b/booth-217",
  "vendor_ticker": "EXMP",   # optional, if mapped
  "timestamp": "2027-01-09T14:15:00Z",
  "visitor_count": 432,
  "avg_dwell_s": 210.3,
  "engagement_score": 0.81,
  "conversion_proxy": 0.14
}

Aggregated over the expo, these features form a time series f_k,t summarizing how much attention and engagement a particular category or issuer receives, with much finer granularity than traditional survey-based or anecdotal reports.

Lead–lag structure: expos to retail

Downstream, many of the same products or categories appear in retail stores, e-commerce platforms, or usage telemetry. Suppose for a given issuer or category we can observe:

• f_t: an expo-derived feature (e.g., normalized engagement index during industry shows),
• R_t: realized revenue or unit sales in retail at a later time,
• P_t: stock price or factor-adjusted return series.

A simple hypothesis is that f_t contains leading information about future R_t+ℓ for some lag ℓ > 0, which in turn may correlate with earnings surprises and eventually with P_t+ℓ'.

At a purely statistical level, one might model:

where X_t captures known macro and seasonal effects, and ε_t is noise. If β₁ is significantly non-zero and stable across expos, f_t acts as a durable leading indicator of demand.

Similarly, we can define an earnings or fundamental surprise S_t+ℓ'' and ask whether:

with γ₁ ≠ 0. In such a case, expo engagement signals derived from Spaxiom might enter into factor models as a new kind of behaviorally grounded "alternative data" factor.

It is crucial that Spaxiom's role here is upstream: it provides clean, semantically meaningful features f_t from messy raw sensor fields, rather than handing unstructured telemetry directly to quantitative researchers.

Cross-site causal chains: expos → flagships → retail

A richer picture emerges when we consider multiple Spaxiom sites linked along a commercialization chain. For a given product category c (e.g., consumer AR headsets), we can define:

• Expo engagement time series: f^expo_c,t from convention centers.
• Flagship experience time series: f^flag_c,t from brand-owned stores and demo labs (e.g., in-store dwell around experience zones).
• Retail adoption time series: f^retail_c,t from multi-brand stores (e.g., dwell around the category, footfall to the shelf, trial-to-purchase proxies).

We can model the propagation of interest as a simple linear–time-invariant system, or more generally using vector autoregressions:

where A encodes propagation and decay of interest, and u_t captures interventions and shocks (marketing campaigns, product recalls, macro events).

If specific entries of A (e.g., expo → flagship, flagship → retail) demonstrate stable, positive influence, then expo-derived features become part of a causal chain that eventually reflects in real economic activity. Again, Spaxiom's contribution is to ensure that the raw measurements at each stage are expressed in compatible INTENT-level semantics, making such modeling feasible at scale.

Example INTENT pattern

As a concrete (simplified) example, consider a Spaxiom INTENT pattern CategoryAggregator that rolls up booth-level engagement into category-level indices in real time:

from spaxiom.intent import CategoryAggregator
from spaxiom.logic  import on, Condition
from spaxiom.temporal import within

# Suppose booth_engagement_stream yields BoothEngagement events
aggregator = CategoryAggregator(source="BoothEngagement")

def category_index(category: str) -> float:
    data = aggregator.snapshot(category=category, window_s=900)
    return data["weighted_engagement"]  # e.g., dwell * visitor_count

# Create a condition that fires once per 15 min window
new_window = within(900, Condition(lambda: True))

@on(new_window)
def publish_category_signals():
    for cat in ["consumer_ar", "gaming_laptops", "ai_pcs"]:
        idx = category_index(cat)
        event = {
            "type": "ExpoCategoryEngagement",
            "category": cat,
            "timestamp": now_iso(),
            "engagement_index": idx
        }
        # Write to an internal bus; a separate, policy-checked
        # process may aggregate & delay-release this as a data product.
        bus.publish("internal.expo.signals", event)

11.2 Contrast with traditional alternative data

It is helpful to contrast Spaxiom-derived signals with existing classes of alternative data commonly used in public equities research:

• Transaction exhaust (credit/debit card panels, bank account aggregates) provides direct but lagged views of consumer spend, often with limited category granularity and complex sampling biases.
• Web and app telemetry (web-scraped prices, app usage metrics, clickstreams) captures digital intent and engagement, but generally lacks fine-grained physical context; the same click may correspond to wildly different in-store realities.
• Remote sensing (satellite imagery, aerial photos, parking lot counts) gives coarse physical signals (e.g., cars at big-box stores, oil tank levels) but at relatively low spatial and temporal resolution and with limited behavioral semantics.
• Survey and panel data (brand trackers, NPS, ad recall) encode sentiment and self-reported behavior, but are expensive to scale, subject to response biases, and typically low-frequency.

Spaxiom-derived experience signals are different along several axes:

1. Spatial and behavioral granularity. By instrumenting floors, zones, and paths inside buildings, Spaxiom can distinguish, for example, "quick pass-by traffic" from "deep engagement dwell" within the same store, and can localize interest to specific fixtures, booths, or demo areas.
2. Real-time, pre-transactional visibility. Many traditional alt-data sources observe behavior after a transaction (card swipes, shipments, bookings). Spaxiom can observe shifts in intent and exploration upstream of purchase: for example, intense engagement with a new category at an industry expo before distribution has ramped.
3. Schema-level semantics. Rather than raw images or ad-hoc features, Spaxiom produces INTENT events with explicit meaning (e.g., BoothEngagement, ProductTrial, AbandonedQueue). This makes it easier to align signals with economic hypotheses, integrate across venues, and audit for misuse.
4. Built-in privacy and policy hooks. Because primitives are defined at the language level, aggregation, anonymization, and latency policies can be expressed as part of the Spaxiom configuration (for example, enforcing minimum crowd sizes or time windows before any signal can leave a site) rather than being bolted on post hoc.
5. Embodied context. Spaxiom is inherently tied to embodied behavior in physical spaces: how people move, queue, linger, and interact with objects. This is complementary to purely digital attention data and potentially more robust to shifts in digital channels or interface design.

Together, these properties suggest that Spaxiom-style experience factors could occupy a distinct niche in the alternative data landscape: less about reconstructing past spend from exhaust, and more about capturing emergent patterns of interaction with the physical world before they fully manifest in traditional financial metrics.

11.3 Logistics corridors and port facilities → global trade and shipping

Global trade flows are mediated by a network of ports, intermodal yards, and cross-dock warehouses. Traditional indicators of trade health (e.g., customs statistics, shipping line disclosures, PMI surveys) are often delayed, low-frequency, and aggregated. A dense deployment of Spaxiom nodes along logistics corridors could provide a higher-frequency, behaviorally grounded view of congestion, throughput, and stress in the supply chain.

Consider a container terminal instrumented with Spaxiom-integrated sensors across:

• inbound and outbound gate queues,
• chassis staging areas,
• yard blocks and stacking rows,
• rail interchange tracks and cross-dock facilities.

From floor sensors, gate loop detectors, RFID/RF beacons, and environmental sensors, INTENT patterns can synthesize events such as:

• GateQueue (queue length, wait time distribution),
• YardCongestion (effective density of containers per zone),
• IdleChassis (underutilized asset pools),
• BerthTurnaround (time from berthing to completion).

For a given facility d and day t, define a congestion index g_d,t ∈ [0,1] derived from normalized queue lengths and dwell times, and a throughput measure q_d,t (e.g., TEUs handled). At a regional or global level, we can aggregate:

where w_d captures facility capacity or strategic weight. G_t then represents a Spaxiom-derived real-time logistics stress index for a trade lane.

We can examine lead–lag relationships between (G_t, Q_t) and traditional macro indicators, such as export volumes, shipping line revenues, or freight rate indices. For example:

where ΔR^shipping_t+ℓ is a sector-level revenue or earnings change at horizon ℓ.

Example INTENT pattern

A simplified pattern for gate queues:

from spaxiom.intent import GateQueueMonitor
from spaxiom.logic  import on, Condition
from spaxiom.temporal import within

gate = GateQueueMonitor(
    entry_sensor=gate_loop_entry,
    exit_sensor=gate_loop_exit,
)

# fires once per 15 minutes
tick_15m = within(900, Condition(lambda: True))

@on(tick_15m)
def emit_gate_state():
    state = gate.snapshot(window_s=900)
    event = {
        "type": "GateQueue",
        "site_id": "port-alpha",
        "timestamp": now_iso(),
        "avg_wait_s": state["avg_wait_s"],
        "p95_wait_s": state["p95_wait_s"],
        "queue_length": state["queue_length"],
        "stress_score": state["stress_score"],
    }
    bus.publish("internal.logistics.events", event)

12. Spaxiom as a Global Experience Substrate for the Era of Experience

The original "Era of Experience" framing envisions agents that learn predominantly from trajectories of interaction with the world, rather than from static internet corpora. In reinforcement learning notation, an agent's experience can be written as a sequence of tuples

where s_t is state, a_t an action, r_t a reward, and o_t an observation at time t.

Spaxiom extends this classical view by inserting a layer of structured events between raw observations and the agent. At each time t, the underlying sensors produce raw signals x_t, which Spaxiom transforms into an event set E_t:

where each e_t⁽ⁱ⁾ is an INTENT-level object such as GaitInstability, CrowdFormation, UnderutilizedExit, or NeedsService.

12.1 Highest-resolution, regularly updated corpus of experience

Consider D deployed Spaxiom sites (hospitals, warehouses, offices, etc.), each generating event streams over time. For site d, let:

denote the set of events emitted over some interval. The global corpus of experience events is then:

Because Spaxiom operates on high-resolution sensor grids (e.g., 4" floor pixels) and other dense modalities, and because it runs continuously at the edge, ℰ_global can approximate a highest-resolution, regularly updated corpus of how people interact with buildings, devices, and robots. Importantly, this corpus is not just unstructured telemetry; it is:

Let |ℰ_global(t)| denote the number of events accumulated up to time t. Assuming each site produces events at average rate λ_d events/second, then:

For many sites and long timescales, this becomes a continuously growing experience fabric whose size and diversity can rival or exceed static web-scale corpora, but now grounded in physical interaction.

12.2 Offline training on structured experience

This corpus can be used to train a variety of models:

• World models that predict future INTENT events conditioned on current context and hypothetical actions.
• Policy networks for control tasks (e.g., crowd management, energy control, fall prevention), trained on logged experience.
• Representation learners that map event sequences to embeddings encoding situation, risk, or intent.

Because events encode semantically meaningful structure, many models can operate in a reduced state space. Suppose a raw sensor state x_t lives in ℝⁿ, but the INTENT state z_t (e.g., counts, scores, zone-level features) lives in ℝ^m with m ≪ n. A world model f_θ can be trained in the lower-dimensional space:

reducing sample complexity and compute requirements relative to predicting in raw sensor space.

12.3 Online serving: experience-RAG and safety priors

Online, agents can treat ℰ_global as a retrieval-augmented memory:

• Given a current event cluster (e.g., emerging crowding near an exit), retrieve similar past episodes and their outcomes.
• Use retrieved episodes to condition an LLM's plan (e.g., how to redirect people, adjust signage, dispatch staff).
• Enforce safety priors by checking whether the current pattern matches known dangerous motifs (e.g., near-misses).

Formally, let q be a query embedding derived from the current situation, and let {z̃_i} be embeddings of past episodes. A retrieval step returns:

and the agent conditions its policy π(a | q, 𝒩(q,K)) on those episodes.

12.4 Governance, attribution, and controllability

Finally, having a language-level representation of experience simplifies governance:

• Attribution: each event knows its origin site, zone, and policy, enabling fine-grained revenue sharing and audit trails.
• Policy filtering: certain event types or fields can be excluded from training or serving (e.g., per-region privacy rules), by manipulating schemas rather than raw logs.
• Red-teaming & safety: dangerous patterns can be expressed as event-level formulas (e.g., combinations of DoorBlocked, CrowdFormation, and low-exit-usage) and used to filter or reweight training data.

In this sense, Spaxiom is not just another middleware layer; it is a deliberately designed substrate for the Era of Experience: a way to convert messy, heterogeneous sensor streams into a structured, governable, and model-ready corpus of physical-world experience.

13. Code Examples Across Domains

13.1 Elder-care: gait instability and LLM agent

from spaxiom.intent import ADLTracker
from spaxiom.temporal import within
from spaxiom.logic import on, Condition

adl = ADLTracker(
    bed_sensor=bed_mat,
    fridge_sensor=fridge_switch,
    bath_sensor=bath_humidity,
    hall_sensor=hall_floor,
)

# Example: alert if no "walk" events in past 6 hours
no_walk_6h = ~within(
    6 * 3600,
    Condition(lambda: adl.daily_counts()["walk"] > 0)
)

@on(no_walk_6h)
def check_on_resident():
    # Agent or workflow integration here
    send_notification("No hallway walk detected in 6h for resident 12B")

13.2 Warehouse queue and congestion advisor

from spaxiom.intent import QueueFlow
from spaxiom.temporal import within
from spaxiom.logic import on, Condition

dock_queue = QueueFlow(dock_floor_sensor)

long_queue = within(300, Condition(lambda: dock_queue.length() > 8))

@on(long_queue)
def suggest_extra_worker():
    facts = {
        "queue_length": dock_queue.length(),
        "wait_time": dock_queue.wait_time(),
        "arrival_rate": dock_queue.arrival_rate(),
    }
    # Hand off to LLM to propose options (reroute trucks, open extra lane, etc.)
    call_llm_with_queue_facts(facts)

13.3 Facilities "needs service" intent

from spaxiom.intent import FmSteward
from spaxiom.logic import on, Condition

fm = FmSteward(
    door_counter=restroom_door_counter,
    towel_sensor=towel_load_cell,
    bin_sensor=bin_ultrasonic,
    gas_sensor=nh3_sensor,
    floor_sensor=wet_strip,
)

needs_service_cond = Condition(fm.needs_service)

@on(needs_service_cond)
def create_ticket():
    payload = fm.snapshot()
    cmms.create_work_order(
        summary=f"Restroom {payload['entries_approx']} entries; needs service",
        metadata=payload,
    )

14. Future Work and Open Research Questions

14.1 Information-theoretic compression of experience

Spaxiom currently relies on hand-designed event schemas. Open questions:

• Can we learn optimal event vocabularies that trade off informativeness vs token cost for agents?
• How close can we get to the information-theoretic optimum for a given downstream objective (control, prediction, explanation)?

Potential direction: treat Spaxiom events as a learned discrete bottleneck, analogous to VQ-VAE codes but for sensor experiences.

14.2 Safe and verifiable compilation

We sketched how safety envelopes could be compiled to automata. Future work:

• Define a verified core subset of Spaxiom suitable for model checking.
• Provide tooling to automatically extract safety-critical fragments and generate:
  • proof obligations,
  • runtime monitors,
  • or certified binaries for embedded controllers.

14.3 Interfacing with frontier models

On the model side:

• How should frontier LLMs be trained to best consume Spaxiom event streams?
• What is the right prompting and tool-calling interface for agents that interact with both:
  • Spaxiom events, and
  • other modalities (text, images, relational databases)?

We expect an ecosystem of Spaxiom-aware agent frameworks, where the DSL is the default way to "speak sensor."

14.4 Benchmarks

To move beyond thought experiments, we plan to define benchmarks across several dimensions:

1. Token Savings:
   • Compare total tokens per decision for baselines vs Spaxiom in multiple environments.
2. Energy per Decision:
   • Combine tokens and measured Joules/token from recent LLM energy studies to quantify actual energy savings.
3. Control Performance:
   • Compare RL/agent policies using raw streams vs Spaxiom events on tasks such as HVAC, queue management, and safety.
4. Explainability & Forensics:
   • Measure human investigators' ability to reconstruct incidents with vs without Spaxiom event logs.

14.5 Developer Experience and Tooling

Adoption of any new framework depends critically on developer experience (DX). Even with powerful abstractions, developers need practical tools to build, test, debug, and deploy Spaxiom applications efficiently. This section outlines the tooling ecosystem required to make Spaxiom accessible to practitioners.

IDE integration and language support

Modern developers expect first-class IDE support. Spaxiom provides:

VSCode extension

• Syntax highlighting: Semantic coloring for sensors, zones, conditions, patterns
• Autocomplete: IntelliSense for INTENT library patterns, temporal operators, spatial queries
• Type checking: Real-time validation of unit compatibility, sensor types, zone boundaries
• Hover documentation: Inline docs for patterns, conditions, and runtime APIs
• Go-to-definition: Navigate from pattern usage to implementation
• Error diagnostics: Inline warnings for common mistakes (e.g., mismatched units, missing sensors)

# VSCode shows inline type hints and errors
from spaxiom import Sensor, Zone, Condition
from spaxiom.units import meters, celsius

zone = Zone(x=0, y=0, width=10 * meters, height=5 * meters)
temp = Sensor(name="temp", unit=celsius)

# Error: incompatible units (meters vs celsius)
# VSCode underlines in red: "Cannot compare Temperature with Distance"
too_hot = Condition(lambda: temp.read() > 10 * meters)  # Type error!

Jupyter notebook support

For exploratory development and data science workflows:

• Interactive pattern testing: Define patterns in cells, visualize state in real-time
• Event stream plotting: Matplotlib/Plotly integration for timelines
• Spatial visualization: Render zones, entities, sensor coverage in notebook

# In Jupyter notebook
from spaxiom.viz import plot_zones, plot_event_timeline

# Visualize spatial layout
plot_zones([loading_zone, staging_zone, storage_zone])

# Plot event timeline
events = store.query(since="2025-01-01", limit=1000)
plot_event_timeline(events, group_by="zone")

Command-line interface (CLI)

The spaxiom CLI provides project scaffolding, testing, and deployment:

Project initialization

# Create new Spaxiom project
$ spaxiom init warehouse-monitor
Created warehouse-monitor/
  ├── spaxiom.yaml        # Configuration
  ├── sensors.py          # Sensor definitions
  ├── patterns.py         # INTENT patterns
  ├── main.py             # Runtime entry point
  └── tests/              # Unit tests

# Install dependencies
$ cd warehouse-monitor
$ spaxiom install

Testing and validation

# Run unit tests (simulated sensors)
$ spaxiom test
Running 12 tests...
✓ test_occupancy_threshold (0.2s)
✓ test_queue_formation (0.5s)
✓ test_overheating_alert (0.1s)
...
12 passed, 0 failed

# Validate configuration
$ spaxiom validate
✓ All sensors defined
✓ All zones have valid coordinates
✓ No circular pattern dependencies
✓ Type checking passed

# Check coverage (which sensors/patterns are tested)
$ spaxiom test --coverage
Sensor coverage: 18/20 (90%)
Pattern coverage: 8/10 (80%)
Condition coverage: 15/20 (75%)

Deployment and monitoring

# Deploy to edge device
$ spaxiom deploy --target pi@192.168.1.100
Uploading code... ✓
Installing dependencies... ✓
Starting runtime... ✓
Runtime listening on http://192.168.1.100:8080

# View logs
$ spaxiom logs --follow
[2025-01-06 10:23:45] INFO: Runtime started (tick_rate=10Hz)
[2025-01-06 10:23:46] INFO: Sensors online: 20/20
[2025-01-06 10:23:47] EVENT: OccupancyChanged(zone=loading, count=5)
[2025-01-06 10:23:50] ALERT: QueueFormed(zone=loading, length=8)

# Check runtime health
$ spaxiom status
Runtime: HEALTHY
Uptime: 3d 14h 22m
Sensors: 20/20 online
Events/sec: 12.3
Latency p99: 8.2ms

Testing framework

Spaxiom includes a pytest-based testing framework with specialized fixtures:

Unit tests for patterns

# tests/test_queue_pattern.py
from spaxiom.testing import MockRuntime, MockSensor, advance_time

def test_queue_formation():
    # Create mock runtime with simulated sensors
    runtime = MockRuntime()
    camera = MockSensor(name="camera", initial_value=0)
    runtime.add_sensor(camera)

    # Create pattern
    from patterns import QueuePattern
    queue = QueuePattern(camera=camera, threshold=5)
    runtime.add_pattern(queue)

    # Simulate sensor readings
    camera.set_value(3)  # Below threshold
    runtime.tick()
    assert not queue.is_active()

    camera.set_value(7)  # Above threshold
    runtime.tick()
    assert queue.is_active()

    # Simulate time passage
    advance_time(runtime, seconds=10)
    assert queue.duration() == 10.0

Integration tests with simulated environments

# tests/test_warehouse_scenario.py
from spaxiom.testing import SimulatedEnvironment

def test_warehouse_workflow():
    # Create simulated warehouse with 4 zones
    env = SimulatedEnvironment.from_config("warehouse.yaml")

    # Simulate 1 hour of activity
    events = []
    for t in range(3600):
        env.step()  # Advance 1 second
        events.extend(env.get_events())

    # Assert expected event sequence
    assert any(e["type"] == "TruckArrived" for e in events)
    assert any(e["type"] == "LoadingStarted" for e in events)
    assert any(e["type"] == "LoadingCompleted" for e in events)

    # Check performance metrics
    avg_loading_time = env.get_metric("avg_loading_time")
    assert avg_loading_time < 600.0  # Under 10 minutes

Debugging and visualization tools

Interactive timeline debugger

Web-based UI for stepping through event sequences:

# Start debugger with recorded events
$ spaxiom debug /var/lib/spaxiom/events.db
Opening debugger at http://localhost:3000

# Debugger UI shows:
# - Timeline slider (scrub through time)
# - Event list (filter by type, zone, priority)
# - Sensor values at each timestamp
# - Pattern state visualization
# - Condition evaluation traces

Spatial visualizer

3D visualization of zones, entities, and sensor coverage:

from spaxiom.viz import SpatialVisualizer

viz = SpatialVisualizer(runtime)

# Render 3D scene (Three.js / Unity)
viz.show_zones(wireframe=True)
viz.show_entities(trail_length=30)  # Show movement trails
viz.show_sensor_coverage(camera_sensors, fov=90)

# Animate event timeline
viz.playback(events, speed=10.0)  # 10x real-time

Breakpoint-on-event debugging

Set breakpoints triggered by specific events:

from spaxiom.debug import breakpoint_on_event

# Pause execution when queue forms
@breakpoint_on_event("QueueFormed")
def inspect_queue(event):
    print(f"Queue in {event['zone']}: length={event['length']}")
    # Drop into interactive debugger
    import pdb; pdb.set_trace()

runtime.run()

Performance profiling and optimization

Built-in profiler

from spaxiom.profiling import RuntimeProfiler

profiler = RuntimeProfiler(runtime)
profiler.start()

runtime.run(duration=300)  # Profile 5 minutes

report = profiler.report()
print(report.summary())
# Output:
# ========== Spaxiom Runtime Profile ==========
# Total ticks: 3000 (10.0 Hz)
#
# Top 5 patterns by latency:
#   1. QueueFlow: 4.2ms avg, 8.9ms p99
#   2. OccupancyField: 2.1ms avg, 5.3ms p99
#   3. ADLTracker: 1.8ms avg, 3.2ms p99
#
# Top 3 callbacks by duration:
#   1. on_queue_formed: 12.3ms avg
#   2. on_overheating: 0.8ms avg
#
# Sensor read latency:
#   camera_loading: 3.2ms avg
#   modbus_temp_1: 0.5ms avg

Memory leak detection

$ spaxiom profile memory --duration 3600
Monitoring memory usage for 1 hour...

Memory growth detected:
  Pattern: OccupancyField (zone=loading)
  Growth rate: +2.3 MB/hour
  Likely cause: Unbounded history buffer

Recommendation: Add max_history_size parameter

Flame graphs for hot-path analysis

$ spaxiom profile flamegraph --output profile.svg
Generated flamegraph: profile.svg
Open in browser to see call stack visualization

Simulation and testing environments

Synthetic sensor generation

from spaxiom.sim import SyntheticSensor, GaussianNoise, PeriodicSignal

# Temperature with diurnal cycle + noise
temp = SyntheticSensor(
    name="temp",
    base_signal=PeriodicSignal(
        amplitude=5.0,  # ±5°C
        period=86400.0,  # 24 hours
        offset=20.0      # 20°C baseline
    ),
    noise=GaussianNoise(std=0.5)
)

# Occupancy with Poisson arrivals
occupancy = SyntheticSensor(
    name="occupancy",
    generator=lambda t: np.random.poisson(lam=10.0)  # 10 people avg
)

runtime.add_sensor(temp)
runtime.add_sensor(occupancy)

Scenario recording and playback

from spaxiom.sim import ScenarioRecorder, ScenarioPlayer

# Record a scenario from production
recorder = ScenarioRecorder(runtime)
runtime.run(duration=3600)  # Record 1 hour
recorder.save("scenarios/normal_ops.spx")

# Later: replay for regression testing
player = ScenarioPlayer.load("scenarios/normal_ops.spx")
test_runtime = SpaxiomRuntime()
player.attach(test_runtime)

# Run at 100x speed
test_runtime.run(speed=100.0)

# Assert no regressions
assert test_runtime.event_count("SafetyViolation") == 0

Monitoring and observability

Prometheus metrics export

from spaxiom.monitoring import PrometheusExporter

exporter = PrometheusExporter(runtime, port=9090)
exporter.start()

# Metrics exposed:
# spaxiom_events_total{type="DoorOpened",zone="loading"}
# spaxiom_pattern_latency_seconds{pattern="QueueFlow"}
# spaxiom_sensor_read_errors_total{sensor="camera_1"}
# spaxiom_runtime_tick_rate_hz

Grafana dashboards

Pre-built Grafana dashboards for common metrics:

# Import dashboard templates
$ spaxiom grafana import --dashboard runtime-overview
$ spaxiom grafana import --dashboard sensor-health
$ spaxiom grafana import --dashboard event-timeline

# Dashboards show:
# - Event rate over time (by type, zone, priority)
# - Pattern latency heatmaps
# - Sensor health status
# - Callback execution times
# - Memory and CPU usage

Distributed tracing (OpenTelemetry)

from spaxiom.tracing import OpenTelemetryTracer

tracer = OpenTelemetryTracer(
    endpoint="http://jaeger:4318",
    service_name="warehouse-runtime"
)
runtime.set_tracer(tracer)

# Traces show:
# - Sensor read → Pattern update → Condition eval → Callback dispatch
# - Cross-site event propagation (edge → cloud)
# - ML model inference triggered by events

Documentation and code generation

Auto-generated API documentation

# Generate docs from code
$ spaxiom docs generate --output docs/
Generating documentation...
  ✓ API reference (200 endpoints)
  ✓ Pattern library (25 patterns)
  ✓ Type definitions (50 types)
  ✓ Examples (30 snippets)

Docs available at docs/index.html

Schema-to-code generator

Generate Python pattern classes from declarative YAML schemas:

# patterns/custom.yaml
- name: CustomOccupancy
  sensors:
    - camera: Camera
    - floor: PressureMat
  parameters:
    - threshold: int
  events:
    - name: ZoneOccupied
      fields:
        - count: int
        - timestamp: datetime

$ spaxiom generate patterns patterns/custom.yaml --output patterns/generated.py
Generated patterns/generated.py with 1 pattern class

# Now import and use
from patterns.generated import CustomOccupancy

OpenAPI spec for REST APIs

$ spaxiom api spec --output openapi.yaml
Generated OpenAPI 3.0 spec with:
  - GET /events
  - GET /sensors/{id}
  - POST /patterns
  - GET /zones

# Use with code generators
$ openapi-generator-cli generate -i openapi.yaml -g python -o clients/python

Package management and distribution

Python package (PyPI)

# Install from PyPI
$ pip install spaxiom

# With optional dependencies
$ pip install spaxiom[vision]     # Camera/video support
$ pip install spaxiom[industrial] # Modbus, OPC UA, BACnet
$ pip install spaxiom[cloud]      # AWS, Azure, GCP connectors
$ pip install spaxiom[ml]         # ML integration (Feast, Tecton)
$ pip install spaxiom[all]        # Everything

Docker images

# Official images on Docker Hub
$ docker pull spaxiom/runtime:latest
$ docker pull spaxiom/runtime:edge    # Minimal for Raspberry Pi
$ docker pull spaxiom/runtime:cloud   # With cloud connectors

# Run containerized runtime
$ docker run -p 8080:8080 \
  -v /var/lib/spaxiom:/data \
  spaxiom/runtime:latest

Kubernetes Helm charts

# Add Helm repo
$ helm repo add spaxiom https://charts.spaxiom.io

# Install runtime on Kubernetes
$ helm install warehouse-runtime spaxiom/runtime \
  --set config.tickRate=10 \
  --set config.sensors[0].name=camera_1 \
  --set config.sensors[0].type=rtsp

# Supports horizontal scaling
$ kubectl scale deployment warehouse-runtime --replicas=10

Interactive tutorials and examples

Web-based interactive playground

Browser-based editor with live preview (like Rust Playground):

# Visit https://playground.spaxiom.io
# Features:
# - Code editor with syntax highlighting
# - Simulated sensors (drag-and-drop spatial layout)
# - Real-time event visualization
# - Share links to examples
# - Fork and modify templates

Example gallery

$ spaxiom examples list
Available examples:
  1. hello-world         Simple occupancy detection
  2. warehouse-queue     Queue flow monitoring
  3. hvac-optimization   Energy-aware HVAC control
  4. robot-safety        Collision avoidance with safety zones
  5. retail-analytics    Customer journey tracking
  ...

$ spaxiom examples run hello-world
Running example: hello-world
Press Ctrl+C to stop

[Output shows simulated events in real-time]

Community and ecosystem

Pattern marketplace

Community repository of reusable patterns:

# Search for patterns
$ spaxiom marketplace search "queue"
Found 5 patterns:
  - advanced-queue-flow (★ 245)
  - multi-stage-queue (★ 89)
  - priority-queue-manager (★ 67)

# Install pattern
$ spaxiom marketplace install advanced-queue-flow
Installed advanced-queue-flow v2.1.0

# Now use in code
from spaxiom.marketplace import AdvancedQueueFlow

Plugin system

# Create custom plugin
from spaxiom.plugins import Plugin

class CustomVisualizerPlugin(Plugin):
    def on_event(self, event):
        # Custom visualization logic
        self.render(event)

# Register plugin
runtime.register_plugin(CustomVisualizerPlugin())

# Discover community plugins
$ spaxiom plugins search "visualization"
$ spaxiom plugins install spaxiom-3d-viz

Summary: comprehensive developer tooling

Spaxiom's developer experience is designed to support the full development lifecycle:

• Authoring: IDE integration, autocomplete, type checking
• Testing: Unit tests, integration tests, simulation environments
• Debugging: Timeline debugger, spatial visualizer, breakpoint-on-event
• Profiling: Performance profiler, memory leak detection, flame graphs
• Deployment: CLI tools, Docker images, Kubernetes Helm charts
• Monitoring: Prometheus metrics, Grafana dashboards, distributed tracing
• Documentation: Auto-generated API docs, interactive tutorials, example gallery
• Community: Pattern marketplace, plugin system, shared examples

By investing in world-class tooling, Spaxiom aims to reduce the barrier to entry for developers building spatiotemporal applications, accelerating adoption across industries from manufacturing to healthcare to smart cities.

15. Conclusion

As sensors proliferate and AI shifts into the Era of Experience, we need programming tools that treat space, time, and interaction as first-class design constraints. Spaxiom proposes:

• a language for describing sensors, spatial layouts, and temporal logic;
• a runtime for embedded and cloud environments;
• an INTENT layer that exposes high-level behaviors and experiences to agents in a token- and energy-efficient way.

In the same way SQL became the lingua franca for structured business data, and modern deep learning frameworks became a lingua franca for neural computation, Spaxiom aims to become a lingua franca for sensor experience: a bridge between the physical world and the agents that will increasingly inhabit it.

If we succeed, future frontier models will not just read the internet; they will read the world through a concise, structured, and safe language that lets them understand, remember, and act on the experiences of billions of devices and the humans they serve.

Appendix: Use Case Atlas

This appendix sketches a set of concrete domains where Spaxiom's spatial-temporal abstractions and INTENT layer provide a natural "sensor cortex" and experience substrate. Each use case is deliberately multi-sensor and goes beyond floor-only deployments, illustrating that the language is designed for general spatial sensing rather than a single modality.

For each use case we:

• describe typical sensor modalities and deployment patterns;
• define example INTENT events and metrics;
• sketch simple mathematical models or indices; and
• provide pseudocode or Spaxiom-style DSL snippets that show how an agent would consume these signals.

A.1 Cleanroom & Biotech Contamination Control

High-grade semiconductor fabs, biopharma facilities, and advanced manufacturing lines rely on strict control of particulate contamination, pressure cascades, and controlled access. Today, contamination control is typically enforced by:

• low-level alarms on individual particle counters or pressure sensors,
• procedural training for operators (gowning, airlock sequencing), and
• offline root-cause analysis when excursions or batch failures occur.

What is often missing is a spatially and temporally coherent language for:

• describing pressure cascades and contamination pathways across rooms;
• identifying recurring patterns of misuse (doors propped, airlocks bypassed);
• correlating subtle sensor deviations with later yield hits.

Spaxiom can act as a cleanroom contamination cortex: fusing particle counts, pressure differentials, door states, and occupancy into structured INTENT events that agents and engineers can reason about.

Sensors and deployment

Typical cleanroom-relevant sensors include:

• Particle counters p_z(t) per zone z (e.g., counts of ≥0.3μm particles per cubic meter);
• Pressure differential sensors ΔP_z,u(t) between adjacent zones z and upstream reference u;
• Door interlocks and access logs d_z(t) (door open/closed, authorized/unauthorized badge);
• Environmental sensors T_z(t) and H_z(t) (temperature, humidity);
• Optional occupancy sensors o_z(t) (PIR, ceiling ToF, or in some zones, floor tiles).

Spaxiom represents each zone as a Zone with associated sensor objects and topological relationships (e.g., which rooms feed which in the pressure cascade).

INTENT events and contamination indices

From raw signals, we define higher-level INTENT events, such as:

• PressureBreach: pressure differential below spec across a boundary;
• ParticleExcursion: temporary spike of particulate counts above class limit;
• AirlockViolation: door sequencing broken (both doors open, or bypass of required dwell);
• HighRiskMovement: occupancy trajectory that crosses from dirtier to cleaner areas under non-compliant conditions.

Mathematically, for a given zone z and monitoring window [t₀, t₁] we can define:

Pressure breach indicator. Let ΔP_z,u(t) be the measured pressure differential between z and upstream zone u (e.g., corridor or anteroom). For a minimum acceptable differential ΔP_min > 0, define

The total number of breach-seconds in the window is:

Particle excursion integral. Let p_z(t) be the particle count for zone z, with a threshold p^max_z that defines acceptable class performance. Define the (non-negative) excursion:

Airlock violation count. Let V_a be the count of airlock a violations in the window, derived from door state sequences and configured policies (e.g., both doors open simultaneously, insufficient purge dwell).

Composite contamination risk index. We can define a simple contamination risk index (CRI) for zone z over [t₀, t₁] as:

where α, β, γ are tunable weights reflecting domain expertise, and 𝒜(z) is the set of airlocks affecting z. This can be normalized by window length or baseline values to obtain a dimensionless score in [0, 1].

Spaxiom-style implementation sketch

A sketch of how a Spaxiom-based contamination monitor might look in Python-embedded DSL form:

from spaxiom import Zone, Condition, Quantity
from spaxiom.temporal import within
from spaxiom.logic import on

class CleanroomZone:
    def __init__(self, name, particle_sensor, dp_sensors, airlocks):
        self.zone = Zone.named(name)
        self.particle_sensor = particle_sensor  # e.g. counts / m^3
        self.dp_sensors = dp_sensors            # dict: upstream_zone -> sensor
        self.airlocks = airlocks                # list of Airlock objects

        # Configurable thresholds
        self.max_particles = 3500        # domain-specific
        self.min_dp = Quantity(5.0, "Pa")

    def pressure_breach_seconds(self, window_s: float) -> float:
        total = 0.0
        for upstream, sensor in self.dp_sensors.items():
            # integrate indicator over the window
            series = sensor.history(window_s=window_s)
            for dt, value in series:  # dt in seconds, value in Pa
                if value < self.min_dp:
                    total += dt
        return total

    def particle_excursion(self, window_s: float) -> float:
        # approximate integral of (p - p_max)+ over the window
        total = 0.0
        series = self.particle_sensor.history(window_s=window_s)
        for dt, value in series:
            excess = max(0.0, value - self.max_particles)
            total += excess * dt
        return total

    def airlock_violations(self, window_s: float) -> int:
        return sum(a.violation_count(window_s=window_s)
                   for a in self.airlocks)

    def contamination_risk_index(self, window_s: float) -> float:
        B = self.pressure_breach_seconds(window_s)
        E = self.particle_excursion(window_s)
        V = self.airlock_violations(window_s)

        alpha, beta, gamma = 1e-3, 1e-6, 1.0  # example scaling
        score = alpha * B + beta * E + gamma * V
        # Optionally squash to [0,1] via logistic
        return 1.0 - (1.0 / (1.0 + score))

# Example zone wiring
zone_a = CleanroomZone(
    name="ISO7_bio_room_3",
    particle_sensor=particles_z3,
    dp_sensors={"antechamber": dp_z3_ante, "corridor": dp_z3_corr},
    airlocks=[airlock_3A, airlock_3B],
)

# Condition that fires when CRI exceeds a threshold in the last hour
high_risk = Condition(
    lambda: zone_a.contamination_risk_index(window_s=3600) > 0.7
)

@on(within(3600, high_risk))
def contamination_agent():
    snapshot = {
        "zone": zone_a.zone.name,
        "CRI": zone_a.contamination_risk_index(window_s=3600),
        "breach_seconds": zone_a.pressure_breach_seconds(3600),
        "particle_excursion": zone_a.particle_excursion(3600),
        "airlock_violations": zone_a.airlock_violations(3600),
    }
    # Hand off to LLM agent or workflow system:
    # e.g., propose root-cause checks, quarantine, or extra cleaning.
    recommend_actions(snapshot)

Here the Spaxiom layer:

• abstracts away sensor-specific histories and units;
• defines windowed functions (history(window_s=...)) for time integration;
• exposes a simple scalar contamination_risk_index for agents, instead of raw time series.

A.2 Rotating Machinery & Industrial Assets (Vibration / Acoustic)

Critical rotating equipment (pumps, fans, compressors, gearboxes, turbines) is instrumented heavily in modern industrial facilities. Traditional "predictive maintenance" stacks often revolve around vendor-specific vibration analyzers, ad-hoc thresholds, and periodic offline analysis. However, the semantics of what operators actually care about are higher-level:

• Which machines are exhibiting early-stage anomalies (imbalance, misalignment, looseness, bearing defects)?
• How do these anomalies correlate with load, process state, and environmental conditions?
• How should maintenance be scheduled, derates applied, or process conditions adjusted to minimize risk?

Spaxiom can serve as a machinery health cortex, fusing vibration, acoustic, electrical, and thermal measurements into structured INTENT events that agents and engineers can reason over consistently across vendors and plants.

Sensors and deployment

For a rotating asset m (e.g., pump P-101), typical sensors include:

• Vibration sensors a_m(t): accelerometers at strategic points (e.g., bearing housing, motor feet), often sampled at kHz and processed into spectral features;
• Acoustic sensors s_m(t): microphones capturing bearing noise, cavitation, and other signatures;
• Electrical sensors i_m(t): motor current and voltage measurements, providing load and power quality context;
• Speed / phase sensors ω_m(t): shaft speed and phase markers (e.g., tachometer pulses);
• Thermal sensors T_m(t): surface or winding temperature.

Spaxiom does not need to operate at raw waveform level; instead, it consumes features precomputed either at the edge or inside a preprocessor:

• RMS vibration levels in bands (overall, 1X, 2X, bearing defect bands),
• Spectral kurtosis and crest factor,
• Harmonic content or envelopes of specific fault frequencies,
• Rolling statistics over electrical and thermal load.

INTENT events and health indices

From these features, Spaxiom defines higher-level INTENT events such as:

• BearingAnomaly (spectral patterns consistent with bearing defect frequencies),
• ImbalanceDetected (elevated 1X vibration at speed-correlated frequency),
• CavitationRisk (acoustic + vibration signature in pumps),
• LoadSpike (current and torque exceeding normal envelopes),
• ThermalRunawayRisk (sustained temperature rise faster than expected under load).

For a machine m and window [t₀, t₁], suppose we have:

• v^1X_m(t): vibration amplitude at the running speed (imbalance indicator),
• v^BD_m(t): vibration amplitude in bearing defect bands,
• k_m(t): spectral kurtosis (impulsiveness),
• T_m(t): temperature,
• L_m(t): mechanical or electrical load fraction.

Normalized feature integrals. Define normalized, windowed feature integrals:

where Δt = t₁ - t₀ and σ^·_m,base are baseline (healthy) standard deviations.

Similarly define:

where T^base_m(L) is the expected temperature as a function of load (learned from historical data).

Composite health score. We can define a simple machine health "anomaly score":

with weights w_i ≥ 0 tuned per machine or class. High H_m indicates abnormal behavior over the window.

Discrete INTENT events. We can then create discrete INTENT events when:

yielding MachineHealthWarning or MachineHealthAlarm for machine m.

Spaxiom-style implementation sketch

In the DSL, a rotating machine entity might encapsulate feature histories and health computation:

from spaxiom import Condition
from spaxiom.temporal import within
from spaxiom.logic import on

class RotatingMachine:
    def __init__(self, name, feat_source):
        self.name = name
        self.feat_source = feat_source  # e.g., streaming feature vectors

        # Baseline stats & thresholds (could be learned)
        self.sigma_1x = 0.12
        self.sigma_bd = 0.08
        self.warn_threshold = 2.0
        self.alarm_threshold = 4.0

    def feature_history(self, window_s: float):
        """
        Returns a list of (dt, features) where features is a dict like:
        {
            "v_1x": ...,
            "v_bd": ...,
            "kurtosis": ...,
            "temp": ...,
            "load": ...,
        }
        """
        return self.feat_source.history(window_s=window_s)

    def expected_temp(self, load: float) -> float:
        # Simple linear model as placeholder; could be learned per machine
        return 40.0 + 25.0 * load

    def health_score(self, window_s: float) -> float:
        series = self.feature_history(window_s)
        if not series:
            return 0.0

        sum_v1x = sum_vbd = sum_k = sum_texcess = 0.0
        total_dt = 0.0

        for dt, f in series:
            v1x = f["v_1x"] / max(self.sigma_1x, 1e-6)
            vbd = f["v_bd"] / max(self.sigma_bd, 1e-6)
            k   = f["kurtosis"]
            temp = f["temp"]
            load = f["load"]

            texcess = max(0.0, temp - self.expected_temp(load))

            sum_v1x     += v1x * dt
            sum_vbd     += vbd * dt
            sum_k       += k * dt
            sum_texcess += texcess * dt
            total_dt    += dt

        if total_dt <= 0:
            return 0.0

        V1x = sum_v1x / total_dt
        Vbd = sum_vbd / total_dt
        K   = sum_k / total_dt
        Tex = sum_texcess / total_dt

        w1, w2, w3, w4 = 0.4, 0.3, 0.2, 0.1
        return w1 * V1x + w2 * Vbd + w3 * K + w4 * Tex

# Example machine instance
pump_101 = RotatingMachine("P-101", feat_source=pump_101_feat_stream)

# Conditions that trigger warnings/alarms over the last 24 hours
warn_condition = Condition(
    lambda: pump_101.health_score(window_s=24 * 3600) > pump_101.warn_threshold
)
alarm_condition = Condition(
    lambda: pump_101.health_score(window_s=24 * 3600) > pump_101.alarm_threshold
)

@on(within(3600, warn_condition))   # evaluate every hour
def pump_warning_agent():
    score = pump_101.health_score(window_s=24 * 3600)
    emit_intent_event({
        "type": "MachineHealthWarning",
        "machine": pump_101.name,
        "score": score,
        "window_h": 24,
    })
    # Optional: hand off to LLM for recommended maintenance actions.

@on(within(600, alarm_condition))   # evaluate every 10 minutes
def pump_alarm_agent():
    score = pump_101.health_score(window_s=24 * 3600)
    emit_intent_event({
        "type": "MachineHealthAlarm",
        "machine": pump_101.name,
        "score": score,
        "window_h": 24,
    })
    # Optionally, trigger interlocks, derates, or automated shutdown logic.

This sketch illustrates how Spaxiom:

• treats feature streams as first-class objects (feat_source.history),
• encapsulates machine-specific health logic in reusable classes,
• promotes a scalar health_score to be consumed by agents and workflows.

A.3 Indoor Air Quality, Occupancy & Health Risk

Indoor air quality (IAQ) and ventilation are increasingly recognized as critical to health, cognitive performance, and resilience to airborne disease. Most commercial buildings already have a wealth of signals (CO₂ sensors, temperature, humidity, HVAC control points), but they are typically used only for crude comfort bands. There is rarely a unified, spatially aware representation of:

• how occupancy and ventilation interact across zones,
• when a space is under-ventilated relative to its load,
• how often high-risk episodes occur and how long they last.

Spaxiom can act as a ventilation and health cortex for buildings, fusing IAQ sensors, occupancy estimates, and HVAC state into intelligible INTENT events like StaleAirEpisode, VentilationDebt, and HighRiskGathering.

Sensors and deployment

For a zone z (e.g., conference room, open office bay, classroom), typical signals include:

• CO₂ concentration C_z(t) (ppm);
• Temperature and humidity T_z(t), H_z(t);
• Fresh air / outdoor air fraction F^OA_z(t) (from damper positions or airflow sensors);
• Supply airflow V̇^sup_z(t) (m³/s or CFM);
• Occupancy proxies N_z(t) from:
  • PIR or time-of-flight sensors,
  • desk/seat occupancy sensors,
  • access-control patterns, or
  • Spaxiom-integrated floor/area sensing where present.

We assume each Zone in Spaxiom is wired to one or more of these signals and that zones can be grouped into higher-level areas for policy and analytics.

INTENT events and risk indices

We define several IAQ- and health-related INTENT events:

• StaleAirEpisode: CO₂ sustained above a threshold (e.g., 1000 ppm) for more than a minimum duration;
• VentilationDebt: integrated shortfall of outdoor air per person relative to recommended rates;
• HighRiskGathering: high occupancy combined with poor ventilation and adverse IAQ (CO₂, PM, humidity).

Let us formalize some of these.

Ventilation per person. For zone z, at time t, the outdoor airflow per person can be approximated as:

where F^OA_z(t) is the fraction of supply air that is outdoor air, and V̇^sup_z(t) is the total supply airflow.

Given a recommended per-person outdoor airflow q^rec (e.g., from a standard), we define a ventilation deficit rate:

Over a monitoring window [t₀, t₁] of length Δt, the ventilation debt is:

CO₂ excursion and stale air. Let C^stale be a CO₂ threshold (e.g., 1000 ppm). Define an indicator:

and the duration of stale air in the window:

Composite health risk score. A simple IAQ/health risk score for zone z over [t₀, t₁] can be defined as:

where:

• E^PM_z is a particulate matter excursion integral;
• E^RH_z is an excursion of relative humidity into high-risk bands (e.g., too dry or too humid);
• α, β, γ, δ are weights reflecting the relative importance of ventilation, CO₂, particulates, and humidity.

Normalization (e.g., dividing by window length or baseline values) can map R^IAQ_z into a dimensionless score in [0,1].

Spaxiom-style implementation sketch

In the Spaxiom DSL, a zone-level IAQ tracker can encapsulate sensor histories and risk computation:

from spaxiom import Zone, Condition
from spaxiom.temporal import within
from spaxiom.logic import on

class IaqZone:
    def __init__(self, name, co2, temp, rh, oa_frac, sup_flow, occupancy):
        self.zone = Zone.named(name)
        self.co2 = co2              # ppm sensor
        self.temp = temp            # degC
        self.rh = rh                # %RH
        self.oa_frac = oa_frac      # 0..1
        self.sup_flow = sup_flow    # m^3/s
        self.occupancy = occupancy  # persons

        # Configurable thresholds / recommendations
        self.q_rec = 10.0 / 3600.0  # 10 m^3/h/person -> m^3/s/person
        self.co2_stale = 1000.0     # ppm
        self.pm_threshold = 25.0    # ug/m^3 (example)
        self.rh_low = 30.0          # %
        self.rh_high = 60.0         # %

    def history(self, sensor, window_s: float):
        return sensor.history(window_s=window_s)  # [(dt, value), ...]

    def ventilation_debt(self, window_s: float) -> float:
        series_oa = self.history(self.oa_frac, window_s)
        series_flow = self.history(self.sup_flow, window_s)
        series_occ = self.history(self.occupancy, window_s)

        # Assume aligned histories or interpolate in real implementation
        total_debt = 0.0
        for ((dt, oa), (_, flow), (_, occ)) in zip(series_oa, series_flow, series_occ):
            q = oa * flow / max(occ, 1.0)
            d = max(0.0, self.q_rec - q)
            total_debt += d * dt
        return total_debt

    def stale_air_duration(self, window_s: float) -> float:
        series = self.history(self.co2, window_s)
        total = 0.0
        for dt, c in series:
            if c > self.co2_stale:
                total += dt
        return total

    def rh_excursion(self, window_s: float) -> float:
        series = self.history(self.rh, window_s)
        total = 0.0
        for dt, r in series:
            if r < self.rh_low or r > self.rh_high:
                total += dt
        return total

    def risk_score(self, window_s: float) -> float:
        D = self.ventilation_debt(window_s)
        S = self.stale_air_duration(window_s)
        RH = self.rh_excursion(window_s)

        alpha, beta, gamma, delta = 1.0, 0.5, 0.0, 0.2  # PM omitted here
        score = alpha * D + beta * S + delta * RH
        # Example squashing to [0,1]
        return 1.0 - (1.0 / (1.0 + score * 1e-4))

# Wire a specific conference room
conf_A = IaqZone(
    name="Conf_Room_A",
    co2=co2_conf_A,
    temp=temp_conf_A,
    rh=rh_conf_A,
    oa_frac=oa_conf_A,
    sup_flow=flow_conf_A,
    occupancy=occ_conf_A,
)

# Condition: high risk over the last 2 hours
high_risk_iaq = Condition(lambda: conf_A.risk_score(window_s=2 * 3600) > 0.7)

@on(within(300, high_risk_iaq))  # check every 5 minutes
def iaq_agent():
    snapshot = {
        "zone": conf_A.zone.name,
        "risk": conf_A.risk_score(window_s=2 * 3600),
        "vent_debt": conf_A.ventilation_debt(2 * 3600),
        "stale_air_s": conf_A.stale_air_duration(2 * 3600),
        "rh_excursion_s": conf_A.rh_excursion(2 * 3600),
    }
    # An LLM or rules engine can:
    # - suggest schedule changes,
    # - recommend window opening where applicable,
    # - adjust ventilation setpoints if allowed.
    propose_iaq_actions(snapshot)

This example shows how Spaxiom:

• computes physically meaningful quantities (ventilation per person) over time windows;
• aggregates multiple risk factors into a scalar risk_score;
• exposes a clean event hook (high_risk_iaq) for agents to react on.

A.4 City-Scale Micro-Mobility & Safety (Radar / Acoustic / IMU)

As cities densify and micro-mobility (bikes, e-scooters, small EVs) proliferates, safety and flow management become critical. Today, safety analysis often relies on:

• sparse and lagged crash statistics,
• occasional manual counts or video studies,
• anecdotal reports from residents and operators.

However, the most informative signals are often the near misses and repeated risky patterns: sudden braking, evasive maneuvers, conflicts between modes, and unsafe speeds at known bottlenecks. These rarely make it into structured datasets.

Spaxiom can act as a street-level safety cortex, fusing radar, acoustic, and IMU signals from vehicles and infrastructure into INTENT events like NearMissCluster, SpeedingCorridor, and UnsafeCrossing.

Sensors and deployment

At the city scale, relevant sensors include:

• Roadside radar / lidar R_j(t) at fixed locations j, measuring:
  • object positions x_k(t) and velocities v_k(t),
  • object classes (car, truck, bike, scooter, pedestrian) from edge classifiers.
• Acoustic sensors A_j(t):
  • sharp impact-like sounds,
  • braking squeals,
  • honking patterns.
• On-vehicle IMUs and GNSS for fleets of:
  • public buses and service vehicles,
  • shared bikes and scooters,
  • possibly taxis or ridehail fleets.
• Signal phase and timing (SPaT):
  • phase state of traffic lights at intersections,
  • timing and coordination information.

Spaxiom ingests processed features from these sources (not raw full-resolution waveforms or video) and maps them into a common spatial model of the street network (segments, intersections, crosswalks, lanes).

INTENT events and risk metrics

We define several safety-relevant INTENT events:

• NearMiss: a spatiotemporal configuration where two or more agents come within a critical distance at relative speed above a threshold without collision;
• HarshEvent: harsh braking, swerving, or strong lateral acceleration from IMU traces;
• SpeedingCorridor: persistent high share of vehicles exceeding speed limits on a segment;
• UnsafeCrossing: repeated near misses or HarshEvents at or near a crosswalk or intersection.

Near-miss detection. Consider two agents a and b (e.g., a car and a cyclist) with positions x_a(t), x_b(t) and velocities v_a(t), v_b(t). For a short horizon τ ∈ [0, τ_max], approximate future positions with constant velocity:

Define the predicted minimum separation distance over that horizon:

Let Δv(t) = v_a(t) - v_b(t), and define relative speed v_rel(t) = ‖Δv(t)‖.

A near-miss candidate is a triple (a,b,t) such that:

where d^nm_thresh (e.g., 1–2 m) and v^nm_thresh (e.g., 5 m/s) are thresholds.

Each such event can be encapsulated as a NearMiss INTENT event with attributes:

• location (segment/crossing),
• agent types (car, bike, scooter, pedestrian),
• relative speed,
• minimum predicted distance.

Segment-level risk score. For a road segment ℓ over a period [t₀, t₁], define:

• N^nm_ℓ: number of near-miss events,
• N^veh_ℓ: total vehicle passages (volume),
• N^harsh_ℓ: number of HarshEvents from IMUs on that segment.

A simple risk index for segment ℓ is:

with weights α, β > 0. Segments with high R_ℓ are candidates for SpeedingCorridor or UnsafeCrossing labels, depending on their geometry.

Spaxiom-style implementation sketch

We can represent a segment or intersection in the DSL as an object that aggregates INTENT events and IMU-derived HarshEvents.

from spaxiom import Condition
from spaxiom.temporal import within
from spaxiom.logic import on

class RoadSegment:
    def __init__(self, seg_id, near_miss_stream, harsh_stream, volume_stream):
        self.seg_id = seg_id
        self.near_miss_stream = near_miss_stream  # yields NearMiss events
        self.harsh_stream = harsh_stream          # yields HarshEvent events
        self.volume_stream = volume_stream        # yields vehicle passage counts

    def counts(self, window_s: float):
        nm_events = self.near_miss_stream.history(window_s=window_s)
        harsh_events = self.harsh_stream.history(window_s=window_s)
        volume_events = self.volume_stream.history(window_s=window_s)

        N_nm = len(nm_events)
        N_harsh = len(harsh_events)
        N_veh = sum(e["count"] for _, e in volume_events) or 1
        return N_nm, N_harsh, N_veh

    def risk_index(self, window_s: float) -> float:
        N_nm, N_harsh, N_veh = self.counts(window_s)
        alpha, beta = 1.0, 0.5
        return alpha * (N_nm / N_veh) + beta * (N_harsh / N_veh)

# Example network segments
seg_main_1 = RoadSegment(
    seg_id="main_st_block_1",
    near_miss_stream=nm_main_1,
    harsh_stream=harsh_main_1,
    volume_stream=volume_main_1,
)
seg_main_2 = RoadSegment(
    seg_id="main_st_block_2",
    near_miss_stream=nm_main_2,
    harsh_stream=harsh_main_2,
    volume_stream=volume_main_2,
)

segments = [seg_main_1, seg_main_2]

# Condition to periodically assess risk
tick_15m = within(900, Condition(lambda: True))

@on(tick_15m)
def micromobility_safety_agent():
    window_s = 7 * 24 * 3600  # last 7 days
    for seg in segments:
        R = seg.risk_index(window_s)
        if R > 0.01:  # example threshold
            emit_intent_event({
                "type": "UnsafeCrossing" if is_crossing(seg.seg_id) else "SpeedingCorridor",
                "segment": seg.seg_id,
                "risk_index": R,
                "window_s": window_s,
            })
    # The resulting stream of UnsafeCrossing / SpeedingCorridor events can drive:
    # - infrastructure recommendations,
    # - signal timing changes,
    # - or targeted enforcement and education campaigns.

In a fuller implementation, separate Spaxiom patterns would compute NearMiss and HarshEvent INTENT events directly from radar and IMU streams, providing reusable building blocks across cities.

A.5 Cold Chain & Pharmaceutical Logistics

from spaxiom import Sensor, Intent, Fusion, Metric
import math

class ColdChainShipment:
    def __init__(self, shipment_id, product_spec):
        self.shipment_id = shipment_id
        self.spec = product_spec  # T_min, T_max, RH_max, shock_threshold

        # Sensor streams
        self.temp_sensors = [Sensor(f"temp_probe_{i}") for i in range(4)]
        self.humidity = Sensor("humidity")
        self.gps = Sensor("gps_tracker")
        self.door = Sensor("door_sensor")
        self.accel = Sensor("accelerometer", axes=["x", "y", "z"])
        self.light = Sensor("light_sensor")

        # INTENT events
        self.temp_excursion = Intent("TempExcursion")
        self.door_breach = Intent("DoorBreach")
        self.route_deviation = Intent("RouteDeviation")
        self.shock_event = Intent("ShockEvent")

        # Fusion metrics
        self.sii = Metric("shipment_integrity_index", range=(0, 1))
        self.mkt = Metric("mean_kinetic_temp", unit="°C")

        # Historical data for MKT calculation
        self.temp_history = []
        self.shock_count = 0
        self.breach_count = 0
        self.route_delay_hours = 0.0
        self.light_exposure_lux_hours = 0.0
        self.rh_excursion_fraction = 0.0

    @Fusion.rule
    def compute_mkt(self):
        """Calculate Mean Kinetic Temperature from probe history"""
        if len(self.temp_history) < 2:
            return self.spec["T_target"]

        # Constants for biologics
        delta_H = 83144  # J/mol (activation energy)
        R = 8.314  # J/(mol·K)

        # Convert to Kelvin and compute exponential sum
        temps_kelvin = [t + 273.15 for t in self.temp_history]
        exp_sum = sum(math.exp(-delta_H / (R * T)) for T in temps_kelvin)

        mkt_kelvin = delta_H / (R * math.log(exp_sum / len(temps_kelvin)))
        mkt_celsius = mkt_kelvin - 273.15

        self.mkt.update(mkt_celsius)
        return mkt_celsius

    @Fusion.rule
    def calculate_sii(self):
        """Compute Shipment Integrity Index from all quality components"""
        # Thermal quality
        mkt = self.compute_mkt()
        T_target = self.spec["T_target"]
        T_tolerance = self.spec["T_tolerance"]
        Q_thermal = max(0, 1 - abs(mkt - T_target) / T_tolerance)

        # Environmental quality
        Q_environmental = (1 - self.rh_excursion_fraction) * \
                         (1 - min(1.0, self.light_exposure_lux_hours / 100))

        # Handling quality
        alpha, beta, gamma = 0.1, 0.2, 0.05
        Q_handling = math.exp(-alpha * self.shock_count -
                             beta * self.breach_count -
                             gamma * self.route_delay_hours)

        # Weighted combination (thermal is most critical)
        w_T, w_E, w_H = 0.6, 0.2, 0.2
        sii = w_T * Q_thermal + w_E * Q_environmental + w_H * Q_handling

        self.sii.update(sii)

        # Emit alerts based on thresholds
        if sii < 0.6:
            Intent.emit("IntegrityCritical", shipment_id=self.shipment_id,
                       sii=sii, action="QUARANTINE_FOR_INSPECTION")
        elif sii < 0.8:
            Intent.emit("IntegrityWarning", shipment_id=self.shipment_id,
                       sii=sii, action="NOTIFY_QA_TEAM")

        return sii

    @Sensor.on_data("temp_probe_*")
    def monitor_temperature(self, probe_id, temp_celsius):
        """Track temperature across all probes, detect excursions"""
        self.temp_history.append(temp_celsius)
        if len(self.temp_history) > 1000:  # Keep last 1000 samples
            self.temp_history.pop(0)

        # Check against specification
        if temp_celsius < self.spec["T_min"] or temp_celsius > self.spec["T_max"]:
            severity = self._classify_excursion_severity(temp_celsius)
            self.temp_excursion.emit(
                shipment_id=self.shipment_id,
                probe_id=probe_id,
                T_actual=temp_celsius,
                T_limit=(self.spec["T_min"], self.spec["T_max"]),
                severity=severity
            )

        self.calculate_sii()

    @Sensor.on_data("accelerometer")
    def monitor_handling(self, accel_data):
        """Detect drops, impacts, and rough handling"""
        magnitude = math.sqrt(sum(a**2 for a in accel_data.values()))

        if magnitude > self.spec["shock_threshold"]:  # e.g., 3g
            self.shock_count += 1
            location = self.gps.latest()["coordinates"]
            self.shock_event.emit(
                shipment_id=self.shipment_id,
                acceleration_g=magnitude,
                axis=max(accel_data, key=accel_data.get),
                location=location
            )
            self.calculate_sii()

    @Sensor.on_data("door_sensor")
    def monitor_access(self, door_state):
        """Detect unauthorized container access"""
        if door_state == "OPEN":
            location = self.gps.latest()["coordinates"]
            if not self._is_authorized_facility(location):
                self.breach_count += 1
                ambient_temp = self._get_ambient_temp(location)
                self.door_breach.emit(
                    shipment_id=self.shipment_id,
                    location=location,
                    ambient_temp=ambient_temp
                )
                self.calculate_sii()

# Example instantiation for COVID-19 vaccine shipment (−70°C ultra-cold chain)
vaccine_spec = {
    "T_target": -70,
    "T_min": -80,
    "T_max": -60,
    "T_tolerance": 10,
    "RH_max": 60,
    "shock_threshold": 3.0  # g-force
}
shipment = ColdChainShipment("SHIP-20251105-001", vaccine_spec)

A.6 Wildfire Risk & Forest Health Monitoring

from spaxiom import Sensor, Intent, Fusion, Metric
import math

class ForestZone:
    def __init__(self, zone_id, coords, elevation_m):
        self.zone_id = zone_id
        self.coords = coords  # (lat, lon) for zone centroid
        self.elevation_m = elevation_m

        # Sensor streams
        self.weather = Sensor("weather_station")  # T, RH, wind, precip
        self.soil_moisture = Sensor("soil_moisture_probe")
        self.multispectral = Sensor("satellite_multispectral")  # NDVI, thermal
        self.smoke_detector = Sensor("smoke_particulate")
        self.acoustic = Sensor("acoustic_array")  # tree stress, beetle activity
        self.lightning = Sensor("lightning_network")

        # INTENT events
        self.dry_fuel = Intent("DryFuelAccumulation")
        self.veg_stress = Intent("VegetationStress")
        self.beetle_infestation = Intent("BarkBeetleInfestation")
        self.ignition_risk = Intent("IgnitionRisk")
        self.fire_detection = Intent("FireDetection")

        # Fusion metrics
        self.ffdi = Metric("forest_fire_danger_index", range=(0, 1))
        self.kbdi = Metric("keetch_byram_drought_index", range=(0, 800))

        # State variables
        self.kbdi_value = 0.0  # Updated daily
        self.fuel_load_tons_per_acre = 12.0  # From field survey or LiDAR
        self.canopy_cover_fraction = 0.65  # From remote sensing
        self.beetle_mortality_fraction = 0.0
        self.lightning_strikes_24h = 0
        self.days_since_rain = 0

    @Fusion.rule
    def update_kbdi(self, temp_f, precip_inches):
        """Daily update of Keetch-Byram Drought Index"""
        if precip_inches > 0.1:
            self.kbdi_value = max(0, self.kbdi_value - precip_inches * 100)
            self.days_since_rain = 0
        else:
            self.days_since_rain += 1
            # KBDI accumulation formula
            if self.kbdi_value < 800:
                factor = (800 - self.kbdi_value) * (0.968 * math.exp(0.0875 * temp_f + 1.5552) - 8.3)
                self.kbdi_value = min(800, self.kbdi_value + factor / 1000)

        self.kbdi.update(self.kbdi_value)

    @Fusion.rule
    def calculate_ffdi(self):
        """Compute Forest Fire Danger Index from all components"""
        # Get latest sensor readings
        wx = self.weather.latest()
        temp_f = wx["temperature_f"]
        rh_pct = wx["relative_humidity"]
        wind_mph = wx["wind_speed"]

        soil = self.soil_moisture.latest()
        fuel_moisture_pct = self._estimate_1000hr_fuel_moisture(soil["moisture_pct"])

        ms = self.multispectral.latest()
        ndvi = ms["ndvi"]
        canopy_temp_c = ms["thermal_c"]
        ambient_temp_c = (temp_f - 32) * 5/9
        canopy_delta = canopy_temp_c - ambient_temp_c

        # Fuel Index
        f_moisture = max(0, 1 - fuel_moisture_pct / 30.0)  # <15% is critical
        f_loading = min(1.0, self.fuel_load_tons_per_acre / 20.0)
        f_continuity = self.canopy_cover_fraction
        I_fuel = f_moisture * f_loading * f_continuity

        # Weather Index (normalized components)
        kbdi_norm = self.kbdi_value / 800.0
        temp_norm = max(0, min(1, (temp_f - 70) / 50.0))
        rh_norm = (100 - rh_pct) / 100.0
        wind_norm = min(wind_mph, 30) / 30.0
        I_weather = 0.4 * kbdi_norm + 0.3 * temp_norm + 0.2 * rh_norm + 0.1 * wind_norm

        # Ecological Stress Index
        ndvi_baseline = 0.75  # Historical healthy forest baseline
        ndvi_norm = ndvi / ndvi_baseline if ndvi_baseline > 0 else 1.0
        ndvi_stress = max(0, 1 - ndvi_norm)

        alpha, beta, gamma = 0.5, 0.3, 0.2
        I_ecological = alpha * ndvi_stress + \
                      beta * min(1.0, canopy_delta / 10.0) + \
                      gamma * self.beetle_mortality_fraction

        # Weighted combination
        w_F, w_W, w_E = 0.4, 0.4, 0.2
        ffdi = w_F * I_fuel + w_W * I_weather + w_E * I_ecological
        ffdi = max(0, min(1, ffdi))  # Clamp to [0,1]

        self.ffdi.update(ffdi)

        # Emit alerts based on thresholds
        if ffdi >= 0.85:
            Intent.emit("ExtremeFireDanger", zone_id=self.zone_id, ffdi=ffdi,
                       action="CLOSE_AREA_PREPOSITION_RESOURCES")
        elif ffdi >= 0.7:
            Intent.emit("HighFireDanger", zone_id=self.zone_id, ffdi=ffdi,
                       action="HEIGHTENED_VIGILANCE")

        # Check for dry fuel accumulation
        if fuel_moisture_pct < 15 and self.days_since_rain > 14:
            self.dry_fuel.emit(zone_id=self.zone_id,
                             fuel_moisture_pct=fuel_moisture_pct,
                             days_since_rain=self.days_since_rain,
                             fuel_load_tons_per_acre=self.fuel_load_tons_per_acre)

        return ffdi

    @Sensor.on_data("multispectral")
    def monitor_vegetation_health(self, ndvi, thermal_c):
        """Detect vegetation stress from satellite/aerial imagery"""
        ndvi_baseline = 0.75
        ndvi_anomaly = ndvi_baseline - ndvi

        wx = self.weather.latest()
        ambient_temp_c = (wx["temperature_f"] - 32) * 5/9
        canopy_delta = thermal_c - ambient_temp_c

        # Significant stress if NDVI drops >20% and canopy is hot
        if ndvi_anomaly > 0.15 and canopy_delta > 5.0:
            self.veg_stress.emit(
                zone_id=self.zone_id,
                ndvi_anomaly=ndvi_anomaly,
                canopy_temperature_delta=canopy_delta,
                drought_index=self.kbdi_value
            )

        self.calculate_ffdi()

    @Sensor.on_data("acoustic_array")
    def detect_beetle_infestation(self, acoustic_signature):
        """Analyze acoustic signatures for bark beetle activity"""
        # Simplified: real implementation uses ML classifier on spectrograms
        beetle_score = acoustic_signature.get("beetle_probability", 0.0)

        if beetle_score > 0.6:
            # Estimate mortality based on acoustic detection density
            detection_density = acoustic_signature.get("detections_per_hectare", 0)
            estimated_mortality = min(0.5, detection_density / 100.0)
            self.beetle_mortality_fraction = estimated_mortality

            self.beetle_infestation.emit(
                zone_id=self.zone_id,
                acoustic_signature=beetle_score,
                tree_count_estimate=detection_density * 10,  # rough conversion
                mortality_rate=estimated_mortality
            )

            self.calculate_ffdi()

    @Sensor.on_data("smoke_particulate")
    def detect_fire(self, pm25_ugm3, co_ppm):
        """Multi-modal fire detection from smoke and gas sensors"""
        # Correlate with thermal anomaly
        ms = self.multispectral.latest()
        thermal_c = ms.get("thermal_c", 0)

        # Fire signature: high PM2.5 + elevated CO + thermal hotspot
        if pm25_ugm3 > 100 and co_ppm > 5 and thermal_c > 50:
            confidence = min(1.0, (pm25_ugm3 / 500) * (co_ppm / 20) * (thermal_c / 100))

            self.fire_detection.emit(
                location=self.coords,
                confidence=confidence,
                size_estimate_acres=0.1,  # Initial detection, refine with progression
                rate_of_spread_mph=0.0    # Not yet determined
            )

    def _estimate_1000hr_fuel_moisture(self, soil_moisture_pct):
        """Convert soil moisture to 1000-hour fuel moisture estimate"""
        # Simplified empirical relationship
        # Real implementation uses Nelson (2000) model with VPD, temp, precip history
        return 10 + 0.8 * soil_moisture_pct

# Example instantiation for Sierra Nevada mixed-conifer zone
sierras_zone = ForestZone(
    zone_id="CA_SIER_Z42",
    coords=(38.5, -120.2),
    elevation_m=1800
)

A.7 Data Center Thermal Management & PUE Optimization

from spaxiom import Sensor, Intent, Fusion, Metric
import math

class DataCenterZone:
    def __init__(self, zone_id, num_racks, cooling_capacity_kw):
        self.zone_id = zone_id
        self.num_racks = num_racks
        self.cooling_capacity_kw = cooling_capacity_kw

        # Sensor streams (per-rack arrays)
        self.rack_temps = [Sensor(f"rack_{i}_temp") for i in range(num_racks)]
        self.rack_power = [Sensor(f"rack_{i}_power") for i in range(num_racks)]
        self.rack_airflow = [Sensor(f"rack_{i}_airflow") for i in range(num_racks)]

        # Zone-level sensors
        self.crac_supply = Sensor("crac_supply_temp")
        self.crac_return = Sensor("crac_return_temp")
        self.crac_power = Sensor("crac_power_meter")
        self.chiller_power = Sensor("chiller_power_meter")
        self.outdoor_weather = Sensor("outdoor_weather_station")

        # Server telemetry (aggregated from orchestrator)
        self.server_utilization = Sensor("server_cpu_gpu_utilization")

        # INTENT events
        self.hotspot = Intent("HotspotFormation")
        self.cooling_inefficiency = Intent("CoolingInefficiency")
        self.capacity_constraint = Intent("CapacityConstraint")
        self.economizer_opportunity = Intent("EconomizerOpportunity")
        self.thermal_runaway = Intent("ThermalRunaway")

        # Fusion metrics
        self.pue = Metric("power_usage_effectiveness", range=(1.0, 3.0))
        self.cer = Metric("cooling_efficiency_ratio", unit="W/W")
        self.tci = Metric("thermal_compliance_index", range=(0, 1))
        self.tes = Metric("thermal_efficiency_score", range=(0, 1))

        # Configuration
        self.temp_setpoint_c = 22.0  # Target cold aisle temp
        self.temp_max_c = 27.0  # ASHRAE A2 upper limit
        self.temp_min_c = 18.0  # ASHRAE A2 lower limit

    @Fusion.rule
    def calculate_pue(self):
        """Compute real-time zone-level PUE"""
        # Sum IT power across all racks
        it_power_kw = sum(sensor.latest()["power_kw"] for sensor in self.rack_power)

        # Cooling infrastructure power
        crac_kw = self.crac_power.latest()["power_kw"]
        chiller_kw = self.chiller_power.latest()["power_kw"]
        cooling_power_kw = crac_kw + chiller_kw

        # Simplified: assume lighting/other is 5% of IT load
        other_power_kw = 0.05 * it_power_kw

        total_power_kw = it_power_kw + cooling_power_kw + other_power_kw

        if it_power_kw > 0:
            pue_value = total_power_kw / it_power_kw
        else:
            pue_value = 1.0

        self.pue.update(pue_value)

        # Alert on PUE degradation (baseline target: 1.15)
        if pue_value > 1.3:
            Intent.emit("PUE_Anomaly",
                       facility_id=self.zone_id,
                       current_pue=pue_value,
                       baseline_pue=1.15,
                       contributing_factors=self._diagnose_pue_factors())

        return pue_value

    @Fusion.rule
    def calculate_cer(self):
        """Compute Cooling Efficiency Ratio"""
        crac = self.crac_supply.latest()
        supply_temp_c = crac["temp_c"]

        return_temp_c = self.crac_return.latest()["temp_c"]
        delta_t = return_temp_c - supply_temp_c

        # Simplified: assume airflow rate proportional to IT load
        # Real implementation uses chilled water flow meters
        it_power_kw = sum(sensor.latest()["power_kw"] for sensor in self.rack_power)

        # Heat removed (kW) ≈ IT power (assuming ~95% converts to heat)
        q_removed_kw = 0.95 * it_power_kw

        # Cooling power consumption
        cooling_power_kw = self.crac_power.latest()["power_kw"] + \
                          self.chiller_power.latest()["power_kw"]

        if cooling_power_kw > 0:
            cer_value = q_removed_kw / cooling_power_kw
        else:
            cer_value = 0.0

        self.cer.update(cer_value)
        return cer_value

    @Fusion.rule
    def calculate_tci(self):
        """Compute Thermal Compliance Index across all racks"""
        violations = 0
        total_samples = 0

        for rack_sensor in self.rack_temps:
            temps = rack_sensor.latest()["u_positions"]  # Dict: {U10: temp, U20: temp, ...}

            for u_pos, temp_c in temps.items():
                total_samples += 1
                if temp_c > self.temp_max_c:
                    violations += (temp_c - self.temp_max_c)
                elif temp_c < self.temp_min_c:
                    violations += (self.temp_min_c - temp_c)

        temp_range = self.temp_max_c - self.temp_min_c

        if total_samples > 0:
            tci_value = 1.0 - (violations / (total_samples * temp_range))
            tci_value = max(0, min(1, tci_value))
        else:
            tci_value = 1.0

        self.tci.update(tci_value)

        # Alert on thermal violations
        if tci_value < 0.9:
            Intent.emit("ThermalViolation", zone_id=self.zone_id, tci=tci_value)

        return tci_value

    @Fusion.rule
    def calculate_tes(self):
        """Compute composite Thermal Efficiency Score"""
        pue_val = self.pue.latest()
        cer_val = self.cer.latest()
        tci_val = self.tci.latest()

        # Normalize PUE: ideal=1.0, poor=2.0 -> scale to [0,1]
        pue_normalized = max(0, min(1, 2 - pue_val))

        # Normalize CER: typical range 2.0-5.0 -> scale to [0,1]
        cer_normalized = max(0, min(1, (cer_val - 2.0) / 3.0))

        # Weights: PUE most important, then CER, then compliance
        w_P, w_C, w_T = 0.5, 0.3, 0.2
        tes_value = w_P * pue_normalized + w_C * cer_normalized + w_T * tci_val

        self.tes.update(tes_value)
        return tes_value

    @Sensor.on_data("rack_*_temp")
    def monitor_hotspots(self, rack_id, temp_data):
        """Detect thermal hotspot formation"""
        u_positions = temp_data["u_positions"]

        for u_pos, temp_c in u_positions.items():
            delta = temp_c - self.temp_setpoint_c

            if temp_c > self.temp_max_c:
                # Check airflow to diagnose cause
                rack_idx = int(rack_id.split('_')[1])
                airflow_data = self.rack_airflow[rack_idx].latest()
                airflow_cfm = airflow_data["cfm"]
                expected_cfm = 200  # Typical per-rack requirement

                airflow_deficit = max(0, expected_cfm - airflow_cfm)

                self.hotspot.emit(
                    rack_id=rack_id,
                    location=u_pos,
                    temp_c=temp_c,
                    delta_from_setpoint=delta,
                    airflow_deficit_cfm=airflow_deficit
                )

                # Rapid escalation indicates thermal runaway
                if hasattr(self, '_last_temp') and rack_id in self._last_temp:
                    rate_of_change = (temp_c - self._last_temp[rack_id]) / 60.0  # °C/min

                    if rate_of_change > 2.0:  # >2°C/min is emergency
                        Intent.emit("ThermalRunaway",
                                   server_id=f"{rack_id}_{u_pos}",
                                   component="CPU",
                                   temp_c=temp_c,
                                   rate_of_change_c_per_min=rate_of_change)

                if not hasattr(self, '_last_temp'):
                    self._last_temp = {}
                self._last_temp[rack_id] = temp_c

        self.calculate_tci()
        self.calculate_tes()

    @Sensor.on_data("outdoor_weather_station")
    def evaluate_economizer(self, outdoor_temp_c, outdoor_humidity_pct):
        """Check for free cooling opportunities"""
        indoor_temp_c = self.crac_return.latest()["temp_c"]

        # Air-side economizer viable if outdoor temp < indoor return - 5°C
        # and humidity acceptable (30-60%)
        if (outdoor_temp_c < indoor_temp_c - 5.0 and
            30 <= outdoor_humidity_pct <= 60):

            # Estimate savings: mechanical cooling power that could be avoided
            current_cooling_kw = self.crac_power.latest()["power_kw"]
            estimated_savings_kwh = current_cooling_kw * 0.7  # 70% reduction typical

            self.economizer_opportunity.emit(
                facility_id=self.zone_id,
                outdoor_temp_c=outdoor_temp_c,
                indoor_temp_c=indoor_temp_c,
                estimated_savings_kwh=estimated_savings_kwh
            )

    @Sensor.on_data("server_cpu_gpu_utilization")
    def predict_capacity_constraints(self, utilization_data):
        """Forecast thermal headroom for workload placement"""
        # Current IT load
        current_it_kw = sum(sensor.latest()["power_kw"] for sensor in self.rack_power)

        # Available cooling capacity
        used_cooling_pct = (current_it_kw / self.cooling_capacity_kw) * 100
        available_cooling_kw = self.cooling_capacity_kw - current_it_kw

        # Queued workload demand (from orchestrator)
        queued_kw = utilization_data.get("queued_workload_kw", 0)

        if available_cooling_kw < queued_kw:
            time_to_saturation = 0  # Already constrained
        else:
            # Estimate time to saturation based on workload growth rate
            growth_rate_kw_per_hour = utilization_data.get("growth_rate_kw_per_hour", 0)
            if growth_rate_kw_per_hour > 0:
                time_to_saturation = (available_cooling_kw - queued_kw) / growth_rate_kw_per_hour
            else:
                time_to_saturation = float('inf')

        if time_to_saturation < 24:  # Less than 24 hours headroom
            self.capacity_constraint.emit(
                zone_id=self.zone_id,
                available_cooling_kw=available_cooling_kw,
                queued_workload_kw=queued_kw,
                time_to_saturation_hours=time_to_saturation
            )

    def _diagnose_pue_factors(self):
        """Root-cause analysis for PUE degradation"""
        factors = []

        # Check cooling efficiency
        cer = self.cer.latest()
        if cer < 2.5:
            factors.append("LOW_CER_COOLING_INEFFICIENCY")

        # Check thermal compliance
        tci = self.tci.latest()
        if tci < 0.9:
            factors.append("THERMAL_VIOLATIONS_OVERCOOLING")

        # Check IT utilization (low utilization inflates PUE)
        util = self.server_utilization.latest().get("avg_cpu_pct", 0)
        if util < 30:
            factors.append("LOW_SERVER_UTILIZATION")

        return factors if factors else ["UNKNOWN"]

# Example instantiation for hyperscale zone
zone_a = DataCenterZone(
    zone_id="DC1_ZONE_A",
    num_racks=40,
    cooling_capacity_kw=800
)

A.8 Agriculture Precision Irrigation & Crop Stress Management

from spaxiom import Sensor, Intent, Fusion, Metric
import math

class IrrigationZone:
    def __init__(self, zone_id, crop_type, area_ha, soil_type):
        self.zone_id = zone_id
        self.crop_type = crop_type
        self.area_ha = area_ha
        self.soil_type = soil_type

        # Sensor streams
        self.soil_moisture = Sensor("soil_moisture_array")  # Multiple depths
        self.canopy_temp = Sensor("infrared_thermometer")
        self.multispectral = Sensor("ndvi_camera")
        self.weather = Sensor("weather_station")
        self.sap_flow = Sensor("sap_flow_gauge")
        self.dendrometer = Sensor("stem_diameter_sensor")

        # INTENT events
        self.water_stress = Intent("WaterStress")
        self.nutrient_deficiency = Intent("NutrientDeficiency")
        self.irrigation_event = Intent("IrrigationEvent")
        self.frost_risk = Intent("FrostRisk")

        # Fusion metrics
        self.cwsi = Metric("crop_water_stress_index", range=(0, 1))
        self.soil_moisture_deficit = Metric("soil_moisture_deficit_mm", unit="mm")
        self.iwue = Metric("irrigation_water_use_efficiency", unit="kg/mm")
        self.et_daily = Metric("daily_evapotranspiration", unit="mm/day")

        # Crop parameters (example: tomatoes)
        self.crop_params = self._get_crop_parameters(crop_type)
        self.growth_stage = "vegetative"  # Updated via phenology model

        # Soil hydraulic properties
        self.theta_fc = 0.35  # Field capacity (volumetric)
        self.theta_wp = 0.15  # Wilting point
        self.root_depth_cm = 60  # Effective root zone depth

        # State tracking
        self.cumulative_irrigation_mm = 0
        self.cumulative_gdd = 0  # Growing degree days

    @Fusion.rule
    def calculate_cwsi(self):
        """Compute Crop Water Stress Index from canopy temperature"""
        canopy_data = self.canopy_temp.latest()
        T_canopy = canopy_data["temp_c"]

        wx = self.weather.latest()
        T_air = wx["air_temp_c"]
        RH = wx["relative_humidity_pct"]

        # Calculate vapor pressure deficit (VPD)
        e_sat = 0.6108 * math.exp((17.27 * T_air) / (T_air + 237.3))  # kPa
        e_actual = e_sat * (RH / 100)
        vpd = e_sat - e_actual

        # Empirical baselines (crop/climate specific)
        # Lower limit (well-watered): ΔT = a + b*VPD
        delta_T_ll = -2.0 + 0.5 * vpd

        # Upper limit (non-transpiring): ΔT = c + d*VPD
        delta_T_ul = 2.0 + 2.5 * vpd

        delta_T_actual = T_canopy - T_air

        # Compute CWSI
        if delta_T_ul > delta_T_ll:
            cwsi_value = (delta_T_actual - delta_T_ll) / (delta_T_ul - delta_T_ll)
            cwsi_value = max(0, min(1, cwsi_value))
        else:
            cwsi_value = 0

        self.cwsi.update(cwsi_value)

        # Emit stress event based on threshold and growth stage sensitivity
        stress_threshold = self.crop_params["cwsi_threshold"][self.growth_stage]
        if cwsi_value > stress_threshold:
            severity = self._classify_stress_severity(cwsi_value)
            self.water_stress.emit(
                field_id=self.zone_id,
                zone_id=self.zone_id,
                cwsi=cwsi_value,
                soil_moisture_deficit_mm=self.soil_moisture_deficit.latest(),
                severity=severity
            )

        return cwsi_value

    @Fusion.rule
    def calculate_et_and_water_balance(self):
        """Compute daily ET and soil moisture deficit"""
        wx = self.weather.latest()
        T_air = wx["air_temp_c"]
        RH = wx["relative_humidity_pct"]
        wind_speed_ms = wx["wind_speed_ms"]
        solar_radiation_mj = wx["solar_radiation_mj_m2_day"]

        # FAO-56 Penman-Monteith reference ET (simplified)
        # Full implementation would use complete PM equation
        et0 = self._calculate_et0_penman_monteith(
            T_air, RH, wind_speed_ms, solar_radiation_mj
        )

        # Crop coefficient based on growth stage
        kc = self.crop_params["kc"][self.growth_stage]

        # Crop evapotranspiration
        etc = kc * et0
        self.et_daily.update(etc)

        # Update soil water balance
        soil_data = self.soil_moisture.latest()
        theta_current = soil_data["volumetric_moisture_pct"] / 100

        # Total available water (mm)
        taw = (self.theta_fc - self.theta_wp) * self.root_depth_cm * 10

        # Current available water
        current_aw = max(0, (theta_current - self.theta_wp) * self.root_depth_cm * 10)

        # Management allowable depletion
        mad_fraction = self.crop_params["mad_fraction"][self.growth_stage]
        mad_threshold = mad_fraction * taw

        # Soil moisture deficit (how much water needed to reach field capacity)
        deficit_mm = max(0, (self.theta_fc - theta_current) * self.root_depth_cm * 10)
        self.soil_moisture_deficit.update(deficit_mm)

        # Trigger irrigation if below MAD threshold
        if current_aw < mad_threshold:
            irrigation_depth = deficit_mm * 0.8  # Apply 80% to avoid runoff
            Intent.emit("IrrigationRequired",
                       zone_id=self.zone_id,
                       application_depth_mm=irrigation_depth,
                       urgency="HIGH" if current_aw < mad_threshold * 0.5 else "NORMAL")

        return etc

    @Sensor.on_data("multispectral")
    def monitor_crop_health(self, ndvi, ndre):
        """Detect nutrient deficiencies and vegetation stress via spectral analysis"""
        # Baseline NDVI for healthy crop at current growth stage
        ndvi_baseline = self.crop_params["ndvi_baseline"][self.growth_stage]
        ndvi_anomaly = ndvi_baseline - ndvi

        # NDRE (Red Edge) sensitive to chlorophyll/nitrogen
        ndre_baseline = 0.35  # Typical for healthy vegetative growth
        ndre_deficit = ndre_baseline - ndre

        # Nitrogen deficiency signature: low NDRE with moderate NDVI drop
        if ndre_deficit > 0.08 and 0.7 < ndvi < 0.85:
            self.nutrient_deficiency.emit(
                field_id=self.zone_id,
                nutrient_type="NITROGEN",
                ndvi_anomaly=ndvi_anomaly,
                leaf_chlorophyll=self._estimate_chlorophyll_from_ndre(ndre),
                confidence=0.85
            )

    @Sensor.on_data("weather_station")
    def predict_frost_risk(self, forecast_temp_min_c, forecast_hours):
        """Early warning for frost events during sensitive growth stages"""
        # Critical stages for frost damage (crop specific)
        vulnerable_stages = ["flowering", "fruit_set"]

        if self.growth_stage in vulnerable_stages and forecast_temp_min_c < 2.0:
            # Estimate crop vulnerability based on stage and temperature
            if forecast_temp_min_c < -2:
                vulnerability = "CRITICAL"
                mitigation = "PROTECTIVE_IRRIGATION_WIND_MACHINES"
            elif forecast_temp_min_c < 0:
                vulnerability = "HIGH"
                mitigation = "PROTECTIVE_IRRIGATION"
            else:
                vulnerability = "MODERATE"
                mitigation = "MONITOR_CLOSELY"

            self.frost_risk.emit(
                field_id=self.zone_id,
                predicted_temp_c=forecast_temp_min_c,
                timing_hours=forecast_hours,
                crop_vulnerability=vulnerability,
                mitigation_strategy=mitigation
            )

    def apply_irrigation(self, depth_mm, duration_min):
        """Log irrigation application and update water balance"""
        # Calculate application efficiency (uniformity, evaporation losses)
        efficiency = 0.85  # Drip irrigation typical
        effective_depth = depth_mm * efficiency

        self.cumulative_irrigation_mm += effective_depth

        self.irrigation_event.emit(
            field_id=self.zone_id,
            zone_id=self.zone_id,
            water_applied_mm=depth_mm,
            duration_min=duration_min,
            efficiency_pct=efficiency * 100
        )

        # Trigger soil moisture sensor read to validate application
        self.calculate_et_and_water_balance()

    def _calculate_et0_penman_monteith(self, T, RH, u2, Rs):
        """Simplified FAO-56 Penman-Monteith reference ET calculation"""
        # Saturation vapor pressure (kPa)
        es = 0.6108 * math.exp((17.27 * T) / (T + 237.3))

        # Actual vapor pressure (kPa)
        ea = es * (RH / 100)

        # Slope of saturation vapor pressure curve (kPa/°C)
        delta = (4098 * es) / ((T + 237.3) ** 2)

        # Psychrometric constant (kPa/°C)
        gamma = 0.067  # Approximate at sea level

        # Net radiation (simplified, normally requires detailed calculation)
        Rn = Rs * 0.77  # Approximate conversion

        # Soil heat flux (negligible for daily calculations)
        G = 0

        # Reference ET (mm/day) - simplified Penman-Monteith
        numerator = 0.408 * delta * (Rn - G) + gamma * (900 / (T + 273)) * u2 * (es - ea)
        denominator = delta + gamma * (1 + 0.34 * u2)

        et0 = numerator / denominator
        return max(0, et0)

    def _get_crop_parameters(self, crop_type):
        """Crop-specific parameters for tomatoes (example)"""
        return {
            "kc": {
                "initial": 0.6,
                "vegetative": 0.7,
                "flowering": 1.05,
                "fruit_development": 1.25,
                "maturity": 0.8
            },
            "cwsi_threshold": {
                "vegetative": 0.4,
                "flowering": 0.25,  # More sensitive during flowering
                "fruit_development": 0.3,
                "maturity": 0.5
            },
            "mad_fraction": {
                "vegetative": 0.5,
                "flowering": 0.3,  # Irrigate more frequently
                "fruit_development": 0.35,
                "maturity": 0.6
            },
            "ndvi_baseline": {
                "vegetative": 0.75,
                "flowering": 0.82,
                "fruit_development": 0.80,
                "maturity": 0.65
            }
        }

    def _classify_stress_severity(self, cwsi):
        """Map CWSI to stress severity categories"""
        if cwsi < 0.3:
            return "MILD"
        elif cwsi < 0.5:
            return "MODERATE"
        elif cwsi < 0.7:
            return "SEVERE"
        else:
            return "CRITICAL"

    def _estimate_chlorophyll_from_ndre(self, ndre):
        """Empirical relationship between NDRE and leaf chlorophyll content"""
        # Typical range: 30-60 SPAD units
        return 100 * ndre + 5  # Simplified linear model

# Example instantiation for 10-hectare tomato field
tomato_field = IrrigationZone(
    zone_id="FIELD_T12_ZONE_A",
    crop_type="tomato",
    area_ha=10,
    soil_type="sandy_loam"
)

A.9 Hospital Operating Room Sterility & Workflow Optimization

from spaxiom import Sensor, Intent, Fusion, Metric
import math
from datetime import datetime, timedelta

class OperatingRoom:
    def __init__(self, or_id, iso_class, target_pressure_pa):
        self.or_id = or_id
        self.iso_class = iso_class  # e.g., "ISO_5"
        self.target_pressure_pa = target_pressure_pa  # +2.5 Pa minimum

        # Sensor streams
        self.particle_counter = Sensor("particle_counter")
        self.pressure_monitor = Sensor("differential_pressure")
        self.door_sensor = Sensor("door_open_close")
        self.rtls = Sensor("real_time_location_system")
        self.temp_humidity = Sensor("temp_humidity_probe")
        self.instrument_rfid = Sensor("instrument_tracking")
        self.anesthesia_ehr = Sensor("anesthesia_monitoring")

        # INTENT events
        self.sterile_breach = Intent("SterileFieldBreach")
        self.traffic_excess = Intent("TrafficExcess")
        self.turnover_delay = Intent("TurnoverDelay")
        self.equipment_missing = Intent("EquipmentMissing")
        self.pressure_anomaly = Intent("PressureAnomaly")
        self.instrument_count_mismatch = Intent("InstrumentCountMismatch")

        # Fusion metrics
        self.sas = Metric("sterility_assurance_score", range=(0, 1))
        self.oue = Metric("or_utilization_efficiency", unit="%")
        self.turnover_time = Metric("turnover_duration_min", unit="min")

        # ISO Class 5 threshold
        self.particle_limit = 3520  # particles ≥0.5μm per m³

        # State tracking
        self.procedure_active = False
        self.incision_time = None
        self.closure_time = None
        self.door_entries_current_window = []
        self.expected_instrument_count = 0

    @Fusion.rule
    def calculate_sas(self):
        """Compute Sterility Assurance Score from environmental sensors"""
        # Particle quality
        particle_data = self.particle_counter.latest()
        particle_conc = particle_data["count_per_m3"]
        Q_particle = max(0, 1 - (particle_conc / self.particle_limit))

        # Pressure quality
        pressure_data = self.pressure_monitor.latest()
        delta_p = pressure_data["differential_pa"]
        if delta_p >= 2.5:
            Q_pressure = 1.0
        elif delta_p > 0:
            Q_pressure = delta_p / 2.5
        else:
            Q_pressure = 0.0  # Negative pressure is critical failure

        # Traffic quality (door entries in last 30 minutes)
        current_time = datetime.now()
        cutoff_time = current_time - timedelta(minutes=30)
        self.door_entries_current_window = [
            t for t in self.door_entries_current_window if t > cutoff_time
        ]
        N_entries = len(self.door_entries_current_window)
        lambda_penalty = 0.15
        Q_traffic = math.exp(-lambda_penalty * N_entries)

        # Weighted combination
        w_P, w_Dp, w_T = 0.5, 0.3, 0.2
        sas_value = w_P * Q_particle + w_Dp * Q_pressure + w_T * Q_traffic
        self.sas.update(sas_value)

        # Alert on breach
        if sas_value < 0.8 and self.procedure_active:
            # Determine breach type
            breach_causes = []
            if Q_particle < 0.7:
                breach_causes.append("PARTICLE_EXCURSION")
            if Q_pressure < 0.8:
                breach_causes.append("PRESSURE_LOSS")
            if Q_traffic < 0.6:
                breach_causes.append("DOOR_TRAFFIC")

            self.sterile_breach.emit(
                or_id=self.or_id,
                breach_type=breach_causes,
                particle_spike_pct=((particle_conc / self.particle_limit) - 1) * 100,
                pressure_drop_pa=max(0, 2.5 - delta_p),
                door_open_duration_s=N_entries * 45  # Estimate avg 45s per entry
            )

        return sas_value

    @Fusion.rule
    def calculate_oue(self):
        """Compute OR Utilization Efficiency"""
        if not self.incision_time or not self.closure_time:
            return 0  # No active case

        # Surgical time (incision to closure)
        surgical_duration_min = (self.closure_time - self.incision_time).total_seconds() / 60

        # Scheduled block time (from EHR/scheduling system)
        ehr_data = self.anesthesia_ehr.latest()
        scheduled_duration_min = ehr_data.get("scheduled_duration_min", 120)

        oue_value = (surgical_duration_min / scheduled_duration_min) * 100
        self.oue.update(oue_value)

        return oue_value

    @Sensor.on_data("particle_counter")
    def monitor_particle_concentration(self, count_per_m3):
        """Detect particle excursions during procedure"""
        if count_per_m3 > self.particle_limit and self.procedure_active:
            # Immediate alert - ISO Class violation
            excess_pct = ((count_per_m3 / self.particle_limit) - 1) * 100

            self.sterile_breach.emit(
                or_id=self.or_id,
                breach_type=["PARTICLE_EXCURSION"],
                particle_spike_pct=excess_pct,
                pressure_drop_pa=0,
                door_open_duration_s=0
            )

        self.calculate_sas()

    @Sensor.on_data("differential_pressure")
    def monitor_pressure_differential(self, differential_pa, hvac_status):
        """Track OR positive pressure maintenance"""
        if differential_pa < 2.5:
            # Pressure anomaly - risk of contamination ingress
            # Check HVAC diagnostics for root cause
            time_to_recovery = self._estimate_pressure_recovery_time(hvac_status)

            self.pressure_anomaly.emit(
                or_id=self.or_id,
                current_pressure_pa=differential_pa,
                target_pressure_pa=self.target_pressure_pa,
                hvac_status=hvac_status,
                time_to_recovery_min=time_to_recovery
            )

        self.calculate_sas()

    @Sensor.on_data("door_open_close")
    def track_door_traffic(self, event_type, timestamp, personnel_type):
        """Monitor door openings during procedure"""
        if event_type == "OPEN":
            self.door_entries_current_window.append(timestamp)

            # Check traffic during critical surgical phase
            if self.procedure_active and self.incision_time:
                time_since_incision = (timestamp - self.incision_time).total_seconds() / 60
                entries_since_incision = len([
                    t for t in self.door_entries_current_window
                    if t >= self.incision_time
                ])

                # Best practice: <10 entries during surgical phase
                if entries_since_incision > 10:
                    self.traffic_excess.emit(
                        or_id=self.or_id,
                        entry_count=entries_since_incision,
                        time_window_min=time_since_incision,
                        procedure_phase="SURGICAL",
                        personnel_type=personnel_type
                    )

            self.calculate_sas()

    @Sensor.on_data("anesthesia_monitoring")
    def track_procedure_milestones(self, event_type, timestamp):
        """Track surgical timeline from anesthesia EMR integration"""
        if event_type == "INCISION":
            self.incision_time = timestamp
            self.procedure_active = True
        elif event_type == "CLOSURE":
            self.closure_time = timestamp
            self.procedure_active = False
            self.calculate_oue()
        elif event_type == "PATIENT_OUT":
            self._start_turnover_timer(timestamp)

    @Sensor.on_data("instrument_tracking")
    def verify_instrument_count(self, scan_phase, scanned_items, expected_items):
        """RFID verification of surgical instrument counts"""
        scanned_count = len(scanned_items)
        expected_count = len(expected_items)

        if scanned_count != expected_count:
            missing_items = set(expected_items) - set(scanned_items)
            extra_items = set(scanned_items) - set(expected_items)

            self.instrument_count_mismatch.emit(
                or_id=self.or_id,
                case_id=self.anesthesia_ehr.latest().get("case_id"),
                expected_count=expected_count,
                actual_count=scanned_count,
                scan_phase=scan_phase,
                item_details={
                    "missing": list(missing_items),
                    "extra": list(extra_items)
                }
            )

        # Check for expired sterilization
        for item in scanned_items:
            if item.get("sterilization_expiration") < datetime.now():
                self.equipment_missing.emit(
                    or_id=self.or_id,
                    case_id=self.anesthesia_ehr.latest().get("case_id"),
                    instrument_set=item["set_name"],
                    missing_items=[item["item_name"]],
                    sterility_expiration=item["sterilization_expiration"]
                )

    def _start_turnover_timer(self, patient_out_time):
        """Track turnover phases via RTLS"""
        # Monitor cleaning crew arrival, equipment setup, next patient arrival
        # Benchmark: 30 min standard, 45 min complex
        pass

    def _estimate_pressure_recovery_time(self, hvac_status):
        """Predict time to restore positive pressure based on HVAC diagnostics"""
        # Check damper position, fan speed, filter loading
        if hvac_status.get("filter_differential_pa") > 200:
            return 15  # Filter replacement needed
        elif hvac_status.get("damper_position_pct") < 80:
            return 5   # Damper adjustment
        else:
            return 2   # Minor fluctuation, self-correcting

# Example instantiation for OR Suite 3
or_suite_3 = OperatingRoom(
    or_id="OR_03",
    iso_class="ISO_5",
    target_pressure_pa=5.0
)

Appendix A.11: Data Center Campuses & Market Signals

Rapid growth in cloud computing and AI workloads has driven large-scale buildouts of data center campuses, high-voltage interconnects, and district cooling plants. Market participants track these trends via colocation reports, hyperscaler disclosures, and utility filings, but these sources are often lagged and coarse.

Spaxiom deployments across data centers and their supporting energy infrastructure can yield a more granular view of:

• physical build-out activity (crew presence, material flows),
• operational intensity (heat, noise, localized vibration, maintenance),
• cooling and power stress (chiller cycles, transformer loading proxies).

INTENT patterns might include:

• RackHotspot and CoolingStress from thermal and flow sensors;
• CrewDensity and ConstructionPush from floor grids and access logs in under-construction halls;
• MaintenanceIntensity from repeated occupancy of infrastructure zones (e.g., UPS rooms, substations).

For a campus c, let u_c,t denote a normalized utilization or stress proxy (e.g., weighted sum of CoolingStress events), and b_c,t a build-out activity score (e.g., from ConstructionPush events). A simple composite demand indicator might be:

where coefficients α_c, β_c reflect each campus's scale and strategic importance.

We can then explore whether D^DC_t leads:

• reported capex of hyperscalers and colocation providers,
• revenues of data center REITs,
• sales of AI accelerators, cooling systems, or power equipment.

For a given issuer (e.g., an AI hardware supplier) with revenue R_t, a stylized regression might be:

where Z_t includes standard macro factors and order backlog indicators.

Example INTENT pattern

A sketch of how a cooling stress monitor might feed into such an index:

from spaxiom.intent import CoolingMonitor
from spaxiom.logic  import on, Condition
from spaxiom.temporal import within

cooling = CoolingMonitor(
    temp_sensors=rack_thermistors,
    flow_sensors=chilled_water_flow,
)

tick_1h = within(3600, Condition(lambda: True))

@on(tick_1h)
def emit_dc_signal():
    stats = cooling.snapshot(window_s=3600)
    event = {
        "type": "CoolingStress",
        "campus_id": "dc-campus-west-1",
        "timestamp": now_iso(),
        "hotspot_area_pct": stats["hotspot_area_pct"],
        "avg_delta_T": stats["avg_delta_T"],
        "stress_index": stats["stress_index"],
    }
    bus.publish("internal.dc.signals", event)

Appendix A.12: Quick-Service Restaurant Operations & Market Signals

Quick-service restaurants are highly sensitive to operational details: queue management, service times, layout, and menu design. Investors track same-store sales, traffic counts, and average ticket sizes, but these are typically reported quarterly with limited insight into the underlying behavior.

Spaxiom deployments in QSR locations can provide a fine-grained, anonymized view of how guests actually move through the system, via:

• floor grids in the dining room and order pickup areas,
• sensors in drive-thru lanes,
• basic counters at order kiosks or counters (without capturing identities).

INTENT patterns might include:

• QueueLength and ServiceTimeDistribution for both counter and drive-thru;
• DiningDwell and TableTurnover in the seating area;
• AbandonedQueue (guests leaving the line before ordering);
• PeakStress (periods of simultaneous long queues and delayed service).

Let T^svc_i,t denote the average service time at store i during period t, and Q^len_i,t the average or p95 queue length. A simple operational throughput proxy could be:

with Spaxiom providing direct estimates of both numerator and denominator via INTENT events.

At a chain or segment level, we can construct:

where w_i captures store weights (e.g., typical volume). We can compare O^QSR_t to reported same-store sales growth ΔS_t or traffic metrics, and test whether:

where M_t captures macro variables (e.g., fuel prices, unemployment) and ν_t is noise. A stable, positive ψ₁ would suggest that Spaxiom-derived operational throughput is a leading indicator of sales performance.

Example INTENT pattern

A simplified pattern for monitoring counter queues and service times:

from spaxiom.intent import QsrFlow
from spaxiom.logic  import on, Condition
from spaxiom.temporal import within

qsr = QsrFlow(
    entry_sensor=line_entry_floor,
    order_sensor=counter_region_floor,
    exit_sensor=exit_floor,
)

tick_5m = within(300, Condition(lambda: True))

@on(tick_5m)
def emit_qsr_ops():
    stats = qsr.snapshot(window_s=300)
    event = {
        "type": "QsrOpsSnapshot",
        "store_id": "store-4821",
        "timestamp": now_iso(),
        "avg_service_time_s": stats["avg_service_time_s"],
        "p90_queue_length": stats["p90_queue_length"],
        "abandon_rate": stats["abandon_rate"],
        "throughput_per_hour": stats["throughput_per_hour"],
    }
    bus.publish("internal.qsr.ops", event)