Prismatic Embodiment

Overview

Prismatic Embodiment bridges the physical and digital worlds by providing IoT sensor integration, physical infrastructure monitoring, and real-world data collection capabilities. While the majority of the Prismatic Platform operates on digital data sources, many intelligence and security use cases require correlation with physical-world events -- building access logs, environmental sensor readings, network hardware locations, and physical perimeter monitoring. Embodiment provides the ingestion, normalization, and correlation layer that makes physical data a first-class citizen in the platform's intelligence pipeline. In the context of EASM, physical infrastructure monitoring is essential because digital attack surface exposure often originates from physical access vectors.

The module supports standard IoT protocols including MQTT, CoAP, and AMQP, allowing it to collect data from a wide variety of commercial sensor hardware without custom driver development. Incoming sensor readings are normalized into a unified time-series format with device metadata, measurement type, and confidence scoring annotations before being stored in TimescaleDB hypertables optimized for time-range queries. The Elixir BEAM runtime's lightweight process model is particularly well-suited for this domain -- each sensor connection runs as an isolated process, meaning a failure in one sensor's data stream cannot cascade to affect others.

Beyond simple data collection, Embodiment provides anomaly detection on sensor streams. By establishing baseline behavioral profiles for each sensor, the module can detect deviations -- unexpected temperature spikes in a server room, after-hours motion in a restricted area, or environmental conditions suggesting physical tampering with infrastructure. These anomalies are emitted as platform events that downstream modules like Prismatic Influence and Prismatic Override can act upon, following the platform's event sourcing pattern for full traceability.

Architecture

IoT Devices --> Protocol Adapters --> Broadway Pipeline --> Storage + PubSub
     |              |                     |                 |
  MQTT/CoAP    Normalization         Window Functions    TimescaleDB
  AMQP/BLE     Validation           Anomaly Detection    ETS Cache
  RFID/GPIO    Enrichment           Baseline Compare     PubSub Events
  Custom       Device Registry      Multi-Sensor Fuse    Dashboard Feed

Embodiment is structured around a Device Registry and a Stream Processor. The Device Registry is an ETS-backed GenServer that maintains metadata for all connected sensors -- device ID, protocol, location, measurement types, and health status. The Stream Processor is a Broadway pipeline that consumes messages from protocol-specific producers (MQTT, CoAP, AMQP), normalizes readings, applies anomaly detection, and writes results to both TimescaleDB for persistence and PubSub for real-time distribution.

Each protocol adapter runs as a supervised child under the Embodiment supervisor, and new protocols can be added by implementing the PrismaticEmbodiment.Protocol behaviour without modifying existing code -- a clean application of the adapter pattern.

Key Modules

Key Features

Sensor Integration

• IoT protocol support for MQTT 5.0, CoAP, and AMQP with automatic reconnection and circuit breaker
• Sensor data normalization into unified %Reading{} structs with metadata
• Time-series ingestion pipeline using Broadway for backpressure-aware processing
• Device health monitoring with automatic stale-device detection and alerting via Telemetry

Physical Intelligence

• Geospatial correlation of digital events with physical sensor data using PostGIS
• Physical access point monitoring integrated with building management systems
• Environmental condition tracking for server rooms, offices, and restricted areas
• Asset location intelligence through BLE beacon and RFID reader integration

Anomaly Detection

• Real-time sensor data stream processing with configurable window functions
• Anomaly detection using rolling baseline comparison with adaptive thresholds
• Historical trend analysis with automatic downsampling for long-range queries
• Multi-sensor data fusion combining readings from co-located sensors for higher confidence

Data Management

• Data pipeline with configurable retention policies per sensor type
• Compressed archival for historical sensor data via Prismatic Compression
• Structured logging of all device lifecycle events for audit trail
• GDPR-compliant handling of location data tied to individuals

Protocol Adapter Design

The protocol adapter system follows a behaviour-based architecture that enables adding new IoT protocols without modifying the core ingestion pipeline. Each adapter implements the PrismaticEmbodiment.Protocol behaviour, which defines callbacks for connection management, message parsing, and health reporting.

defmodule PrismaticEmbodiment.Protocol do
  @doc "Behaviour for IoT protocol adapters"
  @callback connect(config :: map()) :: {:ok, connection()} | {:error, term()}
  @callback disconnect(connection()) :: :ok
  @callback parse_message(raw :: binary(), connection()) :: {:ok, Reading.t()} | {:error, term()}
  @callback health_check(connection()) :: :healthy | :degraded | :disconnected

  @type connection :: term()
end

defmodule PrismaticEmbodiment.Protocol.Mqtt do
  @behaviour PrismaticEmbodiment.Protocol

  @impl true
  def connect(%{broker: broker, port: port, topics: topics} = config) do
    with {:ok, client} <- Tortoise311.Connection.start_link(config),
         :ok <- subscribe_topics(client, topics) do
      {:ok, %{client: client, topics: topics}}
    end
  end

  @impl true
  def parse_message(raw, _conn) do
    with {:ok, decoded} <- Jason.decode(raw),
         {:ok, reading} <- normalize_reading(decoded) do
      {:ok, reading}
    end
  end
end

Broadway Pipeline Configuration

The stream processor uses Broadway for reliable, acknowledgment-based message processing with automatic batching and rate limiting. The pipeline configuration controls throughput, concurrency, and error handling for each protocol:

defmodule PrismaticEmbodiment.StreamProcessor do
  use Broadway

  @impl true
  def start_link(opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [module: {PrismaticEmbodiment.Producer, opts}],
      processors: [default: [concurrency: 10, max_demand: 50]],
      batchers: [
        timescaledb: [concurrency: 5, batch_size: 100, batch_timeout: 1_000],
        pubsub: [concurrency: 2, batch_size: 50, batch_timeout: 500]
      ]
    )
  end
end

Usage

# Register a new sensor device
{:ok, device} = PrismaticEmbodiment.register_device(%{
  id: "temp_sensor_dc1_rack3",
  protocol: :mqtt,
  location: %{building: "DC1", floor: 2, rack: "R3"},
  measurements: [:temperature, :humidity]
})

# Subscribe to real-time sensor data with filtering
PrismaticEmbodiment.subscribe(:temperature, location: "DC1", fn reading ->
  if reading.value > 35.0, do: trigger_cooling_alert(reading)
end)

# Query historical sensor data with downsampling
{:ok, data} = PrismaticEmbodiment.query("temp_sensor_dc1_rack3",
  from: ~U[2026-01-01 00:00:00Z],
  to: ~U[2026-01-31 23:59:59Z],
  resolution: :hour)

# Get anomaly report for a location
{:ok, anomalies} = PrismaticEmbodiment.anomalies(location: "DC1", period: :last_24h)

# List all registered devices with health status
{:ok, devices} = PrismaticEmbodiment.list_devices(status: :all)
# => [%{id: "temp_sensor_dc1_rack3", status: :healthy, last_reading: ~U[...]}, ...]

NABLA Compliance

Testing

Protocol adapter tests verify connection management, message parsing, and reconnection behavior using mock IoT broker implementations. Anomaly detection tests verify baseline computation accuracy and threshold sensitivity using synthetic sensor data with known anomaly points. Broadway pipeline tests verify backpressure handling, message ordering, and error recovery.

Integration tests exercise the full pipeline from protocol adapter through normalization, anomaly detection, and storage. Property-based tests generate random sensor readings to verify normalization consistency and anomaly detection stability under varied input distributions.

Integration Points

Performance

Telemetry events: [:prismatic, :embodiment, :reading_ingested], [:prismatic, :embodiment, :anomaly_detected], [:prismatic, :embodiment, :device_stale].

Related Resources

• Prismatic Override -- Emergency response triggered by physical anomalies
• Prismatic Suppression -- Noise filtering for high-volume sensor alerts
• Alert Management Specialist -- Manages sensor anomaly alert routing and escalation
• Adapter Pattern Specialist -- Reviews IoT protocol adapter implementations
• Elixir Architect -- Ensures Broadway pipeline and OTP supervision design follows best practices
• Real-Time Monitoring -- Sensor data feeds into platform-wide monitoring dashboards
• Telemetry Integration -- Device health and pipeline metrics instrumented through Telemetry
• Intelligence Synthesis -- Physical-digital data fusion for comprehensive situational awareness

Connect & Contribute

Created by Tomáš Korcak (korczis) | Open Source under GHL

• GitHub | GitLab | LinkedIn | Contact
• Developer Portal | Architecture | Meet the Creator