Overview

Introduction

The Behavior Analytics microservice is implemented in Python and leverages the Python Multiprocessing library for parallel processing.

The perception layer processes camera or video input streams, performing sequential frame analysis for object detection and tracking. It generates object embeddings and sends protobuf messages containing frame metadata to a message broker. The pipeline supports multiple message brokers including Kafka, Redis Streams, and MQTT, allowing flexibility based on deployment requirements. The diagram above illustrates the input and output of the Behavior Analytics pipeline within the overall dataflow context.

The metadata from the perception layer represents a frame containing object detection bounding boxes, tracking IDs, embedding vectors, and other attributes. This input is processed by the pipeline to generate Behavior data. A Behavior represents an object tracked over time with its associated attributes. The key components of the pipeline are:

• Protobuf Message to Frame: Converts protobuf messages or byte arrays into Frame objects.
• Dynamic Configuration: Applies runtime updates to app and sensors config keys from mdx-notification, with per-key allowlist validation and ack-back semantics.
• Dynamic Calibration: Loads and manages calibration data, supporting multiple coordinate systems (image, Cartesian, geo) and monitoring for calibration file changes.
• Image to Real World Coordinate: Transforms image coordinates to real-world coordinates (latitude, longitude) or other coordinate systems such as Cartesian. Also validates if points fall within defined Regions of Interest (ROI).
• Behavior State Management: Manages the state of each object’s Behavior for individual cameras, processing objects across all camera sensors simultaneously.
• Behavior Processing: Calculates key metrics such as speed, direction, and distance based on object movement, appearance, and embeddings.
• Video Embedding Downsampling: Optional. Compresses time-series video embedding data (e.g. VisionLLM) per sensor while preserving patterns and transitions; supports Swinging Door Trending (SDT) and Sliding Window algorithms.
• Event Detection: Generates events including Tripwire events and ROI events.
• Incident Generation: Detects violations (proximity, restricted zone, confined zone, FOV count) and generates incidents when violations persist beyond configured thresholds.
• Sink: Sends Behavior data, events, and incidents to message brokers (Kafka, Redis Streams, or MQTT).

Input & Output

The pipeline stages and their corresponding inputs and outputs are shown below:

Each stage’s inputs and outputs:

Configuration

ConfigFile

The configuration file supports multiple message broker types. You can configure Kafka, Redis Streams, or MQTT based on your deployment requirements. Set sourceType and sinkType in the app section to specify which broker to use (default is kafka).

{
  "kafka": {
    "brokers": "localhost:9092",
    "group": "behavior-analytics",
    "consumer": {
      "autoOffsetReset": "latest",
      "enableAutoCommit": false,
      "maxPollIntervalMs": 900000,
      "maxPartitionFetchBytes": 10485760,
      "fetchMaxBytes": 104857600,
      "maxPollRecords": 10000,
      "timeout": 0.01
    },
    "producer": {
      "lingerMs": 0
    },
    "topics": [
      {"name": "raw", "value": "mdx-raw"},
      {"name": "frames", "value": "mdx-frames"},
      {"name": "behavior", "value": "mdx-behavior"},
      {"name": "notification", "value": "mdx-notification"},
      {"name": "events", "value": "mdx-events"},
      {"name": "incidents", "value": "mdx-incidents"}
    ]
  },
  "redisStream": {
    "host": "localhost",
    "port": 6379,
    "group": "mdx-app",
    "consumer": {
      "readCount": 8,
      "readBlockMs": 4
    },
    "producer": {
      "maxLen": 10000
    },
    "streams": [
      {"name": "raw", "value": "mdx-raw"},
      {"name": "behavior", "value": "mdx-behavior"},
      {"name": "notification", "value": "mdx-notification"},
      {"name": "incidents", "value": "mdx-incidents"}
    ]
  },
  "mqtt": {
    "host": "localhost",
    "port": 1883,
    "clientId": "mdx-app",
    "keepAliveSec": 60,
    "consumer": {
      "qos": 1,
      "maxPollCount": 16,
      "pollTimeoutSec": 1
    },
    "producer": {
      "qos": 1,
      "retain": true
    },
    "topics": [
      {"name": "raw", "value": "mdx-raw"},
      {"name": "behavior", "value": "mdx-behavior"},
      {"name": "notification", "value": "mdx-notification"},
      {"name": "incidents", "value": "mdx-incidents"}
    ]
  },
  "sensors": [
    {
      "id": "default",
      "configs": [
        {"name": "proximityDetectionEnable", "value": "true"},
        {"name": "proximityDetectionThreshold", "value": "1.8"},
        {"name": "proximityDetectionCenterClasses", "value": "[\"Person\"]"},
        {"name": "proximityDetectionSurroundingClasses", "value": "[\"Person\"]"}
      ]
    }
  ],
  "app": [
    {"name": "behaviorWatermarkSec", "value": "30"},
    {"name": "behaviorStateValidInterval", "value": "6"},
    {"name": "behaviorMaxPoints", "value": "200"},
    {"name": "objectConfidenceThreshold", "value": "0.5"},
    {"name": "sourceType", "value": "kafka"},
    {"name": "sinkType", "value": "kafka"},
    {"name": "numWorkersForBehaviorCreation", "value": "2"},
    {"name": "numWorkersForFrameEnhancement", "value": "2"},
    {"name": "proximityViolationIncidentEnable", "value": "true"},
    {"name": "proximityViolationIncidentThreshold", "value": "1"},
    {"name": "proximityViolationIncidentExpirationWindow", "value": "0.5"},
    {"name": "restrictedAreaViolationIncidentEnable", "value": "true"},
    {"name": "restrictedAreaViolationIncidentThreshold", "value": "1"},
    {"name": "restrictedAreaViolationIncidentExpirationWindow", "value": "0.5"},
    {"name": "confinedAreaViolationIncidentEnable", "value": "true"},
    {"name": "confinedAreaViolationIncidentThreshold", "value": "1"},
    {"name": "confinedAreaViolationIncidentExpirationWindow", "value": "0.5"},
    {"name": "fovCountViolationIncidentEnable", "value": "true"},
    {"name": "fovCountViolationIncidentObjectThreshold", "value": "3"},
    {"name": "fovCountViolationIncidentThreshold", "value": "1"},
    {"name": "fovCountViolationIncidentExpirationWindow", "value": "0.5"},
    {"name": "fovCountViolationIncidentObjectType", "value": "Person"}
  ]
}

Configuration Parameters

• Message Broker Selection: The pipeline supports multiple message brokers. Configure sourceType and sinkType in the app section:

  • kafka (default): Use Kafka for message streaming
  • redisStream: Use Redis Streams for lightweight deployments
  • mqtt: Use MQTT for IoT-oriented deployments

• Kafka Broker: Configure Kafka connection for message streaming:

  • brokers: List of broker host:port pairs (e.g., 10.137.149.51:9092). The sensor-processing layer sends all metadata to this Kafka broker, and the Python streaming pipeline consumes the metadata.
  • group: Consumer group name (required, no default)
  • consumer.autoOffsetReset: Where to start reading when no offset exists (default: latest)
  • consumer.enableAutoCommit: Enable automatic offset commits (default: false)
  • consumer.maxPollIntervalMs: Maximum delay between poll invocations in milliseconds (default: 900000)
  • consumer.maxPartitionFetchBytes: Maximum bytes per partition per fetch request (default: 10485760)
  • consumer.fetchMaxBytes: Maximum bytes per fetch request (default: 104857600)
  • consumer.maxPollRecords: Maximum records returned per poll (default: 1000)
  • consumer.timeout: Consumer timeout in seconds (default: 0.01)
  • consumer.retryMaxAttempts: Maximum attempts to create a consumer / connect to the broker before giving up (default: 3)
  • consumer.retryIntervalSec: Seconds to wait between consumer-creation attempts (default: 30.0)
  • producer.lingerMs: Time to wait before sending a batch in milliseconds (default: 0)
  • producer.messageMaxBytes: Maximum size of a single produced message in bytes (default: 10485760)
  • producer.batchSize: Soft target for batching multiple messages together, in bytes — single messages can exceed this (default: 16384)
  • topics: List of key-value pairs mapping logical names to actual topic names (see table below)

• Redis Stream: Configure Redis connection for Redis Streams-based messaging:

  • host: Redis server hostname (default: localhost)
  • port: Redis server port (default: 6379)
  • db: Redis database number (default: 0)
  • group: Consumer group name (required, no default)
  • consumer.readCount: Maximum entries returned per read (default: 1000)
  • consumer.readBlockMs: Block timeout in milliseconds for stream reads (default: 5)
  • consumer.mkstream: Create stream if it doesn’t exist (default: true)
  • consumer.retryMaxAttempts: Maximum attempts to set up the Redis consumer before giving up (default: 3)
  • consumer.retryIntervalSec: Seconds to wait between connection attempts (default: 30.0)
  • producer.maxLen: Maximum stream length, oldest entries trimmed beyond this (default: 10000)
  • streams: List of key-value pairs mapping logical names to actual stream names (see table below)

• MQTT: Configure MQTT broker connection:

  • host: MQTT broker hostname (required, no default)
  • port: MQTT broker port (required, no default)
  • clientId: Client identifier for MQTT connection (required, no default)
  • keepAliveSec: Keep-alive interval in seconds (default: 60)
  • consumer.qos: Consumer Quality of Service level (default: 1)
  • consumer.maxPollCount: Maximum messages per poll (default: 1000)
  • consumer.pollTimeoutSec: Poll timeout in seconds (default: 0.01)
  • consumer.retryMaxAttempts: Maximum attempts to connect to the MQTT broker before giving up (default: 3)
  • consumer.retryIntervalSec: Seconds to wait between connection attempts (default: 30.0)
  • producer.qos: Producer Quality of Service level (default: 1)
  • producer.retain: Retain published messages on the broker (default: true)
  • topics: List of key-value pairs mapping logical names to actual topic names (see table below)

• Topics/Streams: Each entry is a key-value pair where name is the logical key used by the pipeline and value is the actual topic or stream name:

• App Pipeline Config

  • numWorkersForBehaviorCreation: Number of workers to run behavior creation. If set to 0, behavior creation will not run. Read with no default by the analytics 2D/3D, public-safety and smart-city apps — a config that omits the key crashes those apps at startup. The composite and search/alerts apps default it to 0.
  • numWorkersForFrameEnhancement: Number of workers to run frame enhancement. If set to 0, frame enhancement will not run. No default for the analytics 2D/3D, public-safety and robot-speed-control apps; 0 for the composite app.
  • numWorkersForSpaceEstimation: Number of workers to run the periodic space-utilization estimator. No default for the analytics 3D app; 0 for the composite app. Requires cartesian 3D input (it reads object.bbox3d).
  • numWorkersForEmbedFiltering: Number of workers to run video-embedding downsampling (search/alerts and composite apps). Default: 0
  • numWorkersForBehaviorClustering: Number of workers to run trajectory-clustering enrichment for behaviors (Smart Cities Application). No default for the smart-city app; 0 for the composite app. The smart-city app registers the processor only when inference.enable is true.
  • numWorkersForIncidentGeneration: Number of workers to run incident generation (SearchAndAlertsApp). Default: 0
  • in3dMode: If true, the pipeline runs in 3D mode. Supports env-var indirection: a value starting with $ is resolved from the environment. Default: false
  • compactFrame: If true, the pipeline compacts the frame metadata. Recommended for 3D / multi-sensor frames. Default: false
  • objectConfidenceThreshold: Minimum confidence score for object detection. Objects below this threshold are filtered out. Default: 0.5
  • clusterThreshold: Cosine similarity threshold for per-object embedding clustering (Chinese Restaurant Process) during state management. Default: 0.9
  • imageLocationMode: (Image calibration only.) Bbox reference point: bottom_center or center. Default: bottom_center
  • roiEventDetectionMode: How ROI ENTRY/EXIT is decided — coordinate or bbox. See ROI event detection mode. Unrecognized values fall back to coordinate. Default: coordinate
  • advancedOverlay: If true, the space-utilization visualization draws transparent patches that avoid overlapping existing objects. Costs roughly 0.04 s → 1.7 s per visualization. Default: false

• Tripwire (Sensor-level config)

  • tripwireMinPoints: Minimum number of points an object must be detected before and after the tripwire to generate a trip event. Default: 5

• Proximity Detection (Sensor-level config)

  • proximityDetectionEnable: If true, the pipeline checks for proximity violations between objects. Default: false
  • proximityDetectionThreshold: Distance threshold (in meters) for proximity detection. Default: 1.8
  • proximityDetectionCenterClasses: List of object types to use as center objects for proximity detection (e.g., AMR, humanoid). Default: []
  • proximityDetectionSurroundingClasses: List of object types to check proximity against the center objects. Default: []

• State Management

  • behaviorStateValidInterval: Maximum gap (in seconds) between an object’s previous state end and next state start before treating them as separate behaviors. This is also the inactivity gap after which a track’s state is reclaimed: if an object with the same id reappears after the gap, a new behavior is created. The uniqueness of the behavior is based on sensor + id + start timestamp. The gap is measured on the sensor’s own event clock, never the wall clock, so a lagging or replayed pipeline is not mistaken for a quiet camera. Default: 6
  • behaviorWatermarkSec: The watermark for the behavior in seconds. Default: 30
  • behaviorMaxPoints: Maximum number of behavior points to store. Default: 200
  • behaviorEmitOnce: If true, only ended tracks are written — one behaviorStateValidInterval after a track goes quiet — instead of writing every live track on every batch. A track that ends is written in either mode, carrying info["isComplete"] = "true"; the key decides only whether live tracks accompany it. Tracks still live at shutdown are flushed and marked complete in either mode. Events, anomalies and incidents are always built from the per-batch behaviors and are unaffected. Default: false
  • behaviorStateEndToleranceSec: Grace window (in seconds) past a state’s end within which late messages are still accepted into that state. Default: 0.1
  • behaviorTimeThreshold: ISO-8601 instant; messages at or before it are filtered out. Default: 1970-01-01T00:00:00.000Z
  • stateManagementFilter: Whitelist of object types to admit into the pipeline; other types are dropped at frame ingest, before state management. If the list is empty, no filter is applied and all types pass through. Default: []

• Incident Detection (App-level config)

  • incidentObjectTtl: TTL (seconds) for retained object presence data used during incident confirmation. Default: 3600
  • proximityViolationIncidentEnable: Enable proximity violation incident generation. Requires proximityDetectionEnable at the sensor level. Default: false
  • proximityViolationIncidentThreshold: Duration (seconds) the violation must persist before an incident is confirmed. Default: 1
  • proximityViolationIncidentExpirationWindow: Largest detection gap (seconds) tolerated before the current violation run is closed. Accepts fractional values. Keep it below the matching threshold. Default: 0.5
  • restrictedAreaViolationIncidentEnable: Enable restricted area violation incidents. Default: false
  • restrictedAreaViolationIncidentThreshold: Minimum violation duration (seconds) before an incident is reported. Accepts fractional values. Default: 1
  • restrictedAreaViolationIncidentExpirationWindow: Largest detection gap (seconds) tolerated before the current violation run is closed. Accepts fractional values. Keep it below the matching threshold. Default: 0.5
  • confinedAreaViolationIncidentEnable: Enable confined area violation incidents. Default: false
  • confinedAreaViolationIncidentThreshold: Minimum violation duration (seconds) before an incident is reported. Accepts fractional values. Default: 1
  • confinedAreaViolationIncidentExpirationWindow: Largest detection gap (seconds) tolerated before the current violation run is closed. Accepts fractional values. Keep it below the matching threshold. Default: 0.5
  • fovCountViolationIncidentEnable: Enable FOV count violation incidents. Default: false
  • fovCountViolationIncidentObjectThreshold: Object count at which a violation fires. The comparison is count >= threshold, not >, so the default of 1 makes a single object a violation. To alert on “more than N”, set it to N + 1. Default: 1
  • fovCountViolationIncidentThreshold: Minimum violation duration (seconds) before an incident is reported. Accepts fractional values. Default: 1
  • fovCountViolationIncidentExpirationWindow: Largest detection gap (seconds) tolerated before the current violation run is closed. Accepts fractional values. Keep it below the matching threshold. Default: 0.5
  • fovCountViolationIncidentObjectType: Object type to count for FOV violations. Default: Person

• Frame Management (Sensor-level config)

  • sensorMinFrames: Minimum number of frames to process for frame state management. Default: 5

• Video Embedding (App-level config)

  • embedEnableDownsampling: If true, enable video embedding downsampling; if false, embeddings are passed through unchanged. Default: false
  • embedDownsamplerType: Downsampler algorithm: sdt or window. Default: window
  • embedSensorTTLSec: Seconds of inactivity after which a sensor’s downsampler state is purged. Default: 3600
  • embedDownsampleToleranceMode: Tolerance metric: distance or cosine. Default: cosine
  • embedDownsampleSimilarityThreshold: Cosine similarity threshold (cosine mode). Default: 0.90
  • embedDownsampleDistanceThreshold: Euclidean distance threshold in normalized space (distance mode). Default: 0.15
  • embedDownsampleMaxIntervalSec: Maximum interval (seconds) between saved points; forces a save if exceeded. Default: 60
  • embedDownsampleWindowSize: Sliding window size in points (window algorithm only). Default: 60
  • embedDownsampleMinNeighbours: Minimum consecutive similar neighbors to skip a point (window algorithm only). Default: 3

• Space Analytics

  • spaceAnalyticsIntervalSec: Interval (in seconds) to invoke space analytics. Default: 5.0
  • spaceAnalyticsGridSize: Size of the grid for space analytics. The target area is treated as a rectangle and divided into smaller square grids. The grid size is the length of one side of the square grid. Pallet placement is based on the grids. Smaller grid sizes increase the search space. Default: 0.2
  • spaceAnalyticsUnsafeSize: Size threshold (in meters) to determine unsafe pallets. If a pallet is placed partially outside the designated area (buffer zone) and the size of the outside portion exceeds this value, it is flagged as unsafe. Default: 0.5
  • spaceAnalyticsTargetClasses: List of object types to target for space analytics. Default: [“Box”, “Pallet”]
  • spaceAnalyticsUseGA: If true, the pipeline uses a genetic algorithm to optimize object placements for maximum space utilization; otherwise, a greedy search algorithm is used. Default: false
  • spaceAnalyticsPopulationSizeGA: Population size for the genetic algorithm. Effective only if useGA is true. Default: 200
  • spaceAnalyticsNumGenerationsGA: Number of generations for the genetic algorithm. Effective only if useGA is true. Default: 300

• Playback Configs

  • playbackSensors: List of sensor IDs to include in playback. If empty, all sensors are used. Default: []
  • playbackLoop: Number of times to loop through playback data. Default: 1
  • playbackFilterEmptyObjects: If true, filters out frames with no objects. Default: true
  • playbackInSimulationMode: If true, uses simulated time for playback. Default: false
  • playbackStartUpDelaySec: Delay in seconds before playback starts. Default: 0
  • playbackSimulationTimedeltaInMin: Time delta in minutes for simulation mode. Default: 60
  • playbackSimulationFps: Frames per second for simulation playback. Default: 1
  • playbackSimulationBaseTime: Base timestamp for simulation (ISO format). Default: ""

• Trajectory Configs (Smart Cities Application)

  • trajGeoCoordEnable: Enable geo coordinates for trajectory processing. Default: true
  • trajDirectionMode: Direction calculation mode (0=bearing-based). Default: 0
  • trajDirectionClusterMode: Direction clustering mode. Default: 1
  • trajSmoothMinPoints: Minimum points required for trajectory smoothing. Default: 20
  • trajSmoothWindowSize: Window size for trajectory smoothing. Default: 5
  • trajDistanceStride: Stride for distance computation. Default: 5
  • trajSpeedSegmentSize: Segment size for speed computation. Default: 10
  • mapMatchingMaxPoints: Maximum points for map matching. Default: 5
  • mapMatchingClasses: Object classes to apply map matching. Default: []

• Anomaly Detection (Sensor-level config, Smart Cities Application)

  • anomalyActionThreshold: Confidence threshold for anomaly action detection. Default: 0.5
  • anomalyIgnoreSensors: List of sensor IDs to ignore for anomaly detection. Default: []
  • anomalyClasses: List of object classes for anomaly detection. Default: []
  • anomalySpeedViolation: JSON config for speed violation detection. Includes enable, mphThreshold, timeIntervalSecThreshold. Despite its name, mphThreshold is interpreted as pixels per second for image-coordinate calibration, including when no calibration file is provided.
  • anomalyUnexpectedStop: JSON config for unexpected stop detection. Includes enable, mphThreshold, timeIntervalSecThreshold. Despite its name, mphThreshold is interpreted as pixels per second for image-coordinate calibration, including when no calibration file is provided.
  • anomalyAbnormalMovement: JSON config for abnormal movement detection. Includes enable, distanceMetersThreshold, timeIntervalSecThreshold, changeInDirectionDegree.
  • anomalyFallRisk: JSON config for fall risk detection. Includes enable (default true), nConfirmStart (default 10), nConfirmEnd (default 50).
  • anomalyLackMovement: JSON config for lack-of-movement detection. Includes enable (default true), durationThreshold in seconds (default 120.0).
  • anomalyCollisionDetection: JSON config for collision detection. Includes enable, targetClasses, distanceMetersThreshold.

• Inference Config

  • inference.enable: Enable Triton Inference Server integration. Default: false
  • inference.url: URL of the Triton Inference Server. Default: localhost:8000

• Overriding Default Config

  • All sensors use the configuration defined with sensor “id”: “default”.

    {
      "sensors": [
        {
          "id": "default",
          "configs": [
              ...
          ]
       },
       ...
    }

  • You can override one or more default configurations for a given sensor by adding a new entry with the sensor name. Configuration keys in JSON use camelCase (e.g. tripwireMinPoints). The following overrides the default tripwire min points for sensor-id Warehouse_Cam1:

    {
      "sensors": [
      {
          "id": "default",
          "configs": [
              ...
          ]
      },
      {
          "id": "Warehouse_Cam1",
          "configs": [
              {
                  "name": "tripwireMinPoints",
                  "value": "11"
              },
              ...
          ]
        }
      ]
    }

CalibrationFile

Use the Calibration tool to generate the JSON file. For more details, refer to the Calibration. For the calibration JSON structure, see Calibration Schema.

Components & Customization

To help understand more and potentially customize the microservice, here’s a deep dive of the components’ implementation and the data flow between them:

Application Framework

The recommended way to build analytics applications is using the BaseApp framework. When you build a new pipeline (by modifying ours or creating one from scratch), the library dependencies will be fetched from our artifact repository.

BaseApp provides:

• Automatic source/sink management for multiple message broker types (Kafka, Redis Streams, MQTT)
• Worker-based multi-processing with configurable parallelism
• Dynamic calibration support with automatic type detection (image, cartesian, geo)
• Graceful shutdown handling with signal handlers (SIGINT, SIGTERM)
• Performance monitoring with BatchStats
• Built-in write methods for behaviors, events, frames, incidents, and anomalies

Example

from mdx.analytics.core.app.app_base import BaseApp
from mdx.analytics.core.app.app_runner import run
class MyApp(BaseApp):
    def __init__(self, config, calibration_path):
        super().__init__(config, calibration_path)
        self.register_processor(self.read_raw, self.process, num_workers=2)
    def process(self, frames, stats):
        enhanced = [self.calibration.transform_frame(f) for f in frames]
        self.write_frames(enhanced)
if __name__ == '__main__':
    run(MyApp)

For a complete example, refer to apps/analytics/main_analytics_2d_app.py.

Built-in Methods

Protobuf to Object

The protobuf message payload is sent to a message broker (Kafka, Redis Streams, or MQTT) as a ByteArray. For example, a Kafka ProducerRecord contains (topic, key, value), with topic=mdx-raw, key=sensor-id, and value=protobuf-byte-array. When the Behavior Analytics microservice receives the message, it transforms the byte array into a Frame object. The Frame object corresponds to the video frame being processed by the perception layer. It contains all detected objects for a given frame and their corresponding bounding boxes in a single message. Each object may include additional attributes like embedding vectors, pose data, etc.

To read more about the schema, refer to the Protobuf-Schema.

Usage

When using the BaseApp framework, data reading is handled automatically via self.read_raw():

from mdx.analytics.core.app.app_base import BaseApp
from mdx.analytics.core.schema.proto import schema_pb2 as nvSchema
from mdx.analytics.core.utils.processing_stats import BatchStats
class MyApp(BaseApp):
    def __init__(self, config, calibration_path):
        super().__init__(config, calibration_path)
        self.register_processor(self.read_raw, self.process, num_workers=2)
    def process(self, frames: list[nvSchema.Frame], stats: BatchStats) -> None:
        # frames are automatically deserialized from protobuf
        for frame in frames:
            print(f"Frame {frame.id} from sensor {frame.sensorId}")

For standalone source usage (without BaseApp), use the source factory:

from mdx.analytics.core.stream.source.source_factory import get_source
from mdx.analytics.core.stream.source.source_base import StreamMessageProtoDeserializer
from mdx.analytics.core.schema.proto import schema_pb2 as nvSchema
from mdx.analytics.core.schema.config import AppConfig
# Load config and create source based on sourceType (kafka/redisStream/mqtt)
config = AppConfig(**config_data)
source = get_source(config)
# Poll raw frames from the source
frames = source.poll(
    src_key="raw",
    msg_deserializer=StreamMessageProtoDeserializer(nvSchema.Frame)
)

Definition

The raw frame object is translated to JSON before being stored in Elasticsearch, the JSON representation is as below.

{
  "version": "4.0",
  "id": "252",
  "timestamp": "2022-02-09T10:45:10.170Z",
  "sensorId": "xyz",
  "objects": [
    {
      "id": "3",
      "bbox": {
        "leftX": 285.0,
        "topY": 238.0,
        "rightX": 622.0,
        "bottomY": 687.0
      },
      "type": "Person",
      "confidence": 0.9779,
      "info": {
        "gender": "male",
        "age": 45,
        "hair": "black",
        "cap": "none",
        "apparel": "formal"
      },
      "embedding": {
        "vector": [
          1.4162299633026123,
          -0.28852298855781555,
          1.1123499870300293,
          -0.047587499022483826,
          -2.293760061264038,
          0.8388320207595825
        ]
      }
    }
  ]
}

API

BaseApp provides the following read methods that internally use the configured source:

# BaseApp read methods (use these when extending BaseApp)
class BaseApp:
    # Read raw frames from the configured source
    # group_id_suffix is used when multiple processors read from the same topic
    def read_raw(self, group_id_suffix: str | None = None) -> list[nvSchema.Frame]
    # Read behavior data from the source
    def read_behavior(self, group_id_suffix: str | None = None) -> list[Behavior]
    # Read event data from the source
    def read_events(self, group_id_suffix: str | None = None) -> list[Behavior]
    # Read anomaly data from the source
    def read_anomaly(self, group_id_suffix: str | None = None) -> list[Behavior]

For low-level source access, use the Source interface:

# Source base class interface
class Source:
    # Poll and deserialize messages from a source
    def poll(
        self,
        src_key: str,                    # Stream key (e.g., "raw", "behavior")
        msg_deserializer: Callable,      # Function to deserialize messages
        group_id_suffix: str | None = None
    ) -> list[Any]
    # Read raw stream messages (before deserialization)
    def read(
        self,
        src_key: str,
        group_id_suffix: str | None = None
    ) -> list[StreamMessage]
    # Close the source and release resources
    def close(self) -> None

Image to Global Coordinate Transformation

Frame metadata includes object-id, bbox in image/pixel coordinates, object-type, and other attributes.

Cartesian or geo calibration: Diagram: image coordinates (left) transformed via homography to real-world—geo (lat/lon) or Cartesian (meters)—for speed, distance, ROI, etc. Bbox reference is configurable for image calibration (see Image calibration (no transformation)).

Image calibration: No transformation; output stays in image/pixel. Default when no calibration file or calibrationType is "image".

Usage

With BaseApp, calibration is initialized as self.calibration. DynamicCalibration detects type (image, cartesian, geo) from the JSON and creates the right implementation.

from mdx.analytics.core.app.app_base import BaseApp
from mdx.analytics.core.utils.schema_util import nv_frame_to_messages
class MyApp(BaseApp):
    def process(self, frames, stats):
        # self.calibration is automatically initialized (DynamicCalibration)
        # Filter frames to only include sensors defined in calibration
        frames = self.calibration.filter_frames_by_sensor_id(frames)
        # Transform frames to enhanced frames (with real world coordinates, ROI/FOV count, violations)
        enhanced_frames = [self.calibration.transform_frame(frame) for frame in frames]
        # For behavior processing, convert frames to messages
        messages = [
            msg for frame in frames
            for msg in nv_frame_to_messages(frame, object_filter=self.config.state_mgmt_filter)
        ]
        # Transform messages with calibration (updates locations)
        updated_messages = [self.calibration.transform(msg) for msg in messages]

For standalone usage without BaseApp:

from mdx.analytics.core.transform.calibration.calibration_dynamic import DynamicCalibration
from mdx.analytics.core.schema.config import AppConfig
config = AppConfig(**config_data)
calibration = DynamicCalibration(config, calibration_path)
calibration.start_listen()  # Start monitoring for calibration file changes
# Transform frames
enhanced_frames = [calibration.transform_frame(frame) for frame in frames]

Calibration types

Three types are supported, selected by calibrationType in the calibration JSON (defaulting to image when no file is provided). See the subsections below.

Image calibration (no transformation)

CalibrationI does not transform coordinates; output stays in image/pixel space (no homography). Use when:

• No calibration file at startup (or one will be added later)
• calibrationType is "image"
• Analytics are image-space only

No homography; transform_bbox returns the chosen bbox reference point. 2D bbox only; 3D not supported. filter_frames_by_sensor_id returns all frames (no sensor filtering).

Image location mode (image calibration only)

Config imageLocationMode (JSON key: imageLocationMode) sets the bbox reference point:

Applies only to image calibration (CalibrationI). Cartesian/geo use a fixed convention.

Euclidean/Cartesian calibration

CalibrationE transforms image coordinates to real-world Cartesian (meters) via a per-sensor homography matrix. Use for building or floor maps. Supports 2D bbox (bottom-center reference) and 3D bbox when in3dMode is true; transform_bbox uses perspective transform. Frames are filtered by sensor ID (only sensors defined in calibration are kept).

Geo calibration

Calibration transforms image coordinates to real-world; output is lat/lon (WGS-84) or Cartesian depending on config (trajGeoCoordEnable, coordinate reference system, origin). Handles four cases: latlon→latlon, latlon→cartesian, cartesian→latlon, cartesian→cartesian. 2D bbox only (bottom-center); homography from image to geo/cartesian. Frames are filtered by sensor ID.

Definition

Message represents a single object + contextual data:

class Message(BaseModel):
  messageid: str  # Unique identifier for the message, sensorId + "#-#" + objectId
  timestamp: datetime  # Timestamp of the message
  sensor: Sensor  # Sensor metadata
  object: Optional[Object] = None  # Object metadata
  mdsversion: str = ""  # Version
  place: Place = Field(default_factory=Place)  # Place metadata
  analyticsModule: Optional[AnalyticsModule] = None  # Analytics module that processed the message
  event: Optional[Event] = None  # Event information associated with the message
  videoPath: str = ""  # Video path

API

# Checks if the given point falls within the ROI for the given sensor and roi_id.
def point_in_polygon(self, point: Coordinate | Point2D | nvSchema.Coordinate | nvSchema.Point2D, sensor_id: str, roi_id: str) -> bool
# Transform a 2D or 3D bbox in image coordinates to a single real-world coordinate
# based on the per-sensor homography matrix (2D bbox uses the 4 image-plane corners;
# 3D bbox uses the 8 corners and does not require a transformation matrix).
# Also transforms real-world Cartesian coordinates to lat/lon with respect to a given
# origin. Both the transformation matrix and the origin are initialized from external
# configuration. If the transformation matrix is not present, image coordinates are returned.
# Dispatches on the bbox type (Bbox | Bbox3d).
def transform_bbox(self, bbox: Bbox | Bbox3d, sensor_id: str) -> tuple[Coordinate, Location]
# Transform the message to a new message with updated location in real-world coordinates.
# Internally invokes transform_bbox and returns the updated Message.
def transform(self, msg: Message) -> Message
# Transform the raw frame to enhanced frames (with object locations in real-world coordinates,
# ROI/FOV counts, zone violations, etc.). Internally invokes transform_bbox and returns the
# updated Frame.
def transform_frame(self, frame: nvSchema.Frame) -> nvSchema.Frame

Behavior State Management (BSM)

As the perception layer tracks a given object, the object-id remains the same during the time the object is seen by a single camera. This helps in forming a Behavior for a given object. Behavior defines the sequence of real world coordinates an object was detected at, its appearance, embeddings, pose etc., The span of the Behavior may be short lived based on how long it was seen with respect to a given camera. There are cases where the span of the Behavior can be very long. An example is a car stopped due to a red signal or a car parked on the roadside or a person standing in a queue.

The above diagram depicts a few tracks with respect to vehicle and people movement as seen by the camera. However, this can be any kind of object, like a robot, a bicycle, etc. The BSM maintains the Behavior state on live streaming data. As the objects are detected by the perception layer and frame metadata is sent over the network to the Kafka broker, the Behavior Analytics pipeline consumes all detections corresponding to an object and maintains the Behavior state in memory. The in-memory copy of the Behavior is backed up by a persistent store for reliability. The Behavior gets updated when the perception layer sends more detection metadata with respect to the same object. At some point in time, the object gets out of the Field of View of the camera and no more metadata is sent to the BSM. The BSM then cleans up the Behavior corresponding to the object that is no longer seen. The period of time after which the BSM cleans up the state is configurable. To keep the memory footprint of the Behavior state low, only minimal information is stored during the life of the Behavior. The state comprises the object ID, sensor ID, locations, start and end timestamps, object embeddings, etc.

Behavior State Management API

Usage

State management is driven one batch at a time, not one track at a time. process_batch takes the whole messages_map because deciding which tracks have fallen silent is only sound once every message in the batch has reached its state and every sensor clock has advanced — looping per key would end tracks whose own messages are still sitting later in the same batch.

from mdx.analytics.core.stream.state.behavior.state_management import StateMgmt
from mdx.analytics.core.utils.schema_util import messages_to_map
# Initialize state management with config and calibration
state_mgmt = StateMgmt(config, calibration)
# Convert messages to map: key = sensorId + "#-#" + objectId
messages_map = messages_to_map(updated_messages)
# One call processes the entire batch
behavior_batch = state_mgmt.process_batch(messages_map)
# Write the behaviors the emission policy selected
app.write_behaviors(behavior_batch.behaviors_to_write)
# Turn trip states into tripwire / ROI events
events = []
for trip in behavior_batch.trip_behaviors:
    events.extend(tripwire_event.get_events(trip))
    events.extend(roi_event.get_events(trip))

At shutdown, call flush_behaviors() to release every track still held — live tracks are retained in both emission modes — otherwise those behaviors are never written. Everything it returns is marked info["isComplete"] = "true", since shutdown ends every remaining track:

def close(self) -> None:
    pending = self.state_mgmt.flush_behaviors()
    if pending:
        self.write_behaviors(pending)
    super().close()

Definition

process_batch returns a BehaviorBatch. The three lists hold live references, so enriching an entry in place — adding anomaly info, compacting — is visible to whatever is written:

@dataclass
class BehaviorBatch:
  active_behaviors: list[Behavior]     # One per track this batch updated; feed these to detection
  trip_behaviors: list[Behavior]       # One per track, previous tail + this batch's points; for tripwire/ROI events
  behaviors_to_write: list[Behavior]   # What to write to the behavior stream; emission policy already resolved

Because behaviors_to_write already accounts for behaviorEmitOnce, no caller needs to know whether emit-once is on.

The per-track state itself:

class ObjectState(BaseModel):
  id: str  # Unique identifier for the object, sensorId + "#-#" + objectId
  start: datetime  # Start timestamp of the object state
  end: datetime  # End timestamp of the object state
  points: list[Coordinate] = Field(default_factory=list)  # Sequence of points' coordinates
  bboxes: list[Bbox] = Field(default_factory=list)  # Per-frame bboxes aligned 1:1 with points
  sampling: int = 1  # Sampling stride (1-in-N kept from raw observations)
  sample_phase: int = 0  # Phase for the next incoming raw observation, carried across batches for true 1-in-N sampling
  tail_ts: list[datetime] = Field(default_factory=list)  # Timestamps aligned to points[-len(tail_ts):]; used to bisect-insert tolerance-window messages while sampling == 1
  lastXpoints: list[Coordinate] = Field(default_factory=list)  # Last X points for trajectory analysis
  lastXbboxes: list[Bbox] = Field(default_factory=list)  # Bboxes aligned to lastXpoints
  object: Object | None = None  # Object metadata
  model: Model | None = None  # Model including clustered embeddings

API

from mdx.analytics.core.stream.state.behavior.state_management import StateMgmt, BehaviorBatch
class StateMgmt:
    """Object tracking over time. Base class; use StateMgmtG for geographic coordinates."""
    def __init__(self, config: AppConfig, calibration: CalibrationBase) -> None:
        """Initialize with config and calibration."""
    def process_batch(self, messages_map: dict[str, list[Message]]) -> BehaviorBatch:
        """Process one whole batch of messages grouped by track key. One call is one batch."""
    def flush_behaviors(self) -> list[Behavior]:
        """Release behaviors still held back by emit-once mode. Call at shutdown."""
    def live_object_ids(self) -> list[str]:
        """IDs of tracks currently held in state."""

Embeddings Cluster

Behavior comprises of sequence of real world coordinates an object was detected at, its appearance, embeddings, pose etc., The embeddings or the feature vector represents the object appearance, these embeddings are generated continuously and may vary based on factors like lighting, camera angle and field of view. Embeddings cluster tries to summarize 100s of feature vectors into few salient ones that can be used to represent an object appearance. The cluster algorithm used is CRP, it is a non-parametric generative Bayesian model, the data is processed in a streaming mode, clustering is done in an incremental model. As the sequence of embeddings is processed by the pipeline, it simultaneously learns the number of clusters, the model of each cluster, and entity assignments into clusters. The state of the cluster model is maintained by computing the cluster centers across every micro batch.

The diagram above shows only a few embeddings instead of hundreds; the idea is to create a few representative embeddings from all the detections for a given object. The clustering distance used is cosine similarity, and the default threshold is 0.9. The clustering process completes when an object is no longer detected by a sensor.

Usage

from mdx.analytics.core.utils.crp import CRP
# Given a list of normalized vectors create the CRP cluster model.
# default pnew=0.9, threshold for creation of new cluster
# if given a new vector is not close enough to any of the existing cluster.
# takes two inputs,  vecs = Array[Embedding], pnew = cosine similarity threshold
model = CRP().cluster(embeddings, cluster_threshold)
# Given a list of normalized vectors and a existing model, create a new CRP cluster model.
# default pnew=0.9, threshold for creation of new cluster
updated_model = CRP().update_model(model, embeddings, cluster_threshold)

Behavior Processing

Behavior Attributes

The key attributes of the Behavior are based on object movement + Object appearance with highest confidence + Sensor + Place + Embeddings + Pose + Gesture + Gaze:

Movement Attributes

For image and Cartesian coordinates, Trajectory is used:

from mdx.analytics.core.schema.trajectory.trajectory import Trajectory
tr = Trajectory(id=state.id, start=state.start, end=state.end, points=state.points)
# compute movement behavior or attributes (computed properties on the model)
tr.bearing
tr.direction
tr.distance
tr.linear_distance
tr.time_interval
tr.speed
tr.speed_over_time
tr.smooth_trajectory

Average Speed Computation

The distance computation is made with respect to two consecutive points in the smoothened trajectory. The trajectory distance is the sum of all the individual distances obtained between consecutive points.

Compute distance d between two consecutive points:

d = √((p1.x - p2.x)^2 + (p1.y - p2.y)^2)
distance = Σd
average_speed = distance / time_interval

Speed Computation Over Time

Here is an example of Vehicle moving East with an average speed of 29.95 miles/hr, covering 212.87 meters in 15.90 seconds. However, the average speed may not exactly depict the vehicle speed, how it changes over time. If we consider the speed over time, it looks as below:

speedsOverTime = [35.37, 26.53, 28.63, 30.53, 27.51, 28.34, 29.4, 30.33, 28.79, 26.76, 25.54, 24.71, 23.98, 23.39, 24.32]

The speed computation over time is computed based on smaller segments of the trajectory. default is every segment with a span of 1 seconds speed is computed.

Direction

The Direction is computed based on the bearing of the Object.

The formula is to calculate the bearing of the trajectory from the first to the last point. For Cartesian and image coordinates the angle with the x axis is used.

bearing = math.atan2(p1.y - p2.y, p1.x - p2.x) * 180 / math.pi
bearing = (bearing + 360) % 360

For Cartesian and image coordinates left, right, up & down is used. The direction definition based on bearing is as follows:

# 4-way direction.
directions = ["Right", "Up", "Left", "Down"]
direction = directions[int((bearing / 90) + 0.5) % 4]

Video Embedding Downsampling

Video embedding downsampling reduces the volume of time-series video embedding data (e.g. VisionLLM protobuf messages) while preserving important patterns and transitions. It is useful for pipelines that consume high-dimensional embedding streams and need to store or forward a compressed subset. The implementation supports two algorithms and per-sensor state with configurable tolerance and lifecycle.

Algorithms

• Swinging Door Trending (SDT): Interpolation-based compression. Uses a one-point look-ahead: the last committed point (anchor), a pending point (candidate), and the current point. The candidate is skipped if it lies within a tolerance band of the line from anchor to current. Good for smooth, gradual embedding changes; O(1) memory; one-point latency. Call get_pending_video_embeddings() at shutdown to flush the pending candidate.
• Sliding Window: Neighbor-based novelty detection. Keeps a fixed-size window of recent embeddings; a new point is saved only if it has fewer than min_neighbours consecutive similar neighbors in the window. Good for cyclical or repetitive patterns; O(window size) memory; no look-ahead delay.

Both algorithms support distance (Euclidean in normalized vector space) or cosine (similarity) tolerance metrics. Embeddings are normalized before comparison. Typical thresholds: SDT distance ~0.15, cosine ~0.91; Window distance ~0.45, cosine ~0.90 (looser thresholds for pattern matching).

Configuration

Video embedding is configured via AppConfig.video_embedding (VideoEmbeddingConfig). App-level keys use the embed prefix (camelCase in JSON). For the full list of parameters and defaults, see Configuration Parameters (Video Embedding, App-level config).

Usage

Use VideoEmbeddingStateMgmt with VideoEmbeddingConfig (e.g. from config.video_embedding). Create one manager and call update_video_embeddings(sensor_id, video_embeddings) per sensor for each batch; at shutdown call get_pending_video_embeddings() to flush pending candidates (important for SDT). The class creates downsamplers lazily per sensor and purges sensors inactive beyond sensor_ttl_sec.

from mdx.analytics.core.stream.state.video_embedding.video_embedding_state_mgmt import VideoEmbeddingStateMgmt
from mdx.analytics.core.schema.config import AppConfig
from mdx.analytics.core.utils.schema_util import group_video_embeddings_by_sensor_id
config = AppConfig(**config_data)
state_mgmt = VideoEmbeddingStateMgmt(config.video_embedding)
# Per batch: group by sensor and update
for sensor_id, vid_embeddings in group_video_embeddings_by_sensor_id(video_embeddings).items():
    downsampled = state_mgmt.update_video_embeddings(sensor_id, vid_embeddings)
    # Use downsampled list (e.g. write to sink)
# At shutdown: flush pending (e.g. SDT candidate)
pending = state_mgmt.get_pending_video_embeddings()

Applications such as the fusion search app read video embeddings from a source, pass them through VideoEmbeddingStateMgmt, and write filtered/downsampled results via write_embed_filtered (or a custom sink).

API

from mdx.analytics.core.stream.state.video_embedding.video_embedding_state_mgmt import VideoEmbeddingStateMgmt
class VideoEmbeddingStateMgmt:
    def __init__(self, config: VideoEmbeddingConfig) -> None:
        """Initialize with video embedding config (e.g. AppConfig.video_embedding)."""
    def update_video_embeddings(
        self,
        sensor_id: str,
        video_embeddings: list[nvSchema.VisionLLM]
    ) -> list[nvSchema.VisionLLM]:
        """Process and downsample embeddings for the sensor; returns downsampled list. Skips invalid (no visionEmbeddings)."""
    def get_pending_video_embeddings(self) -> list[nvSchema.VisionLLM]:
        """Flush pending candidates from all sensors (call at shutdown). Critical for SDT."""

Tolerance and algorithm choice can be tuned via the parameters in the Configuration table above.

Event Detection

The event detection system supports two types of events:

• Tripwire Events: It was triggered when objects cross the tripwire line segment.
• ROI Events: It was triggered when objects enter or exit the ROI polygon boundary.

Both tripwire and ROI are defined in the camera/sensor calibration JSON. Multiple tripwires or ROIs can be configured for each sensor.

The wire is defined as a line segment between two points. The direction specifies IN point (p1) and OUT point (p2).

The roiCoordinates define the vertices of the ROI polygon.

Usage

from mdx.analytics.core.app.app_base import BaseApp
from mdx.analytics.core.stream.state.behavior.state_management import StateMgmt
from mdx.analytics.core.transform.event.tripwire_event import TripwireEvent
from mdx.analytics.core.transform.event.roi_event import ROIEvent
from mdx.analytics.core.utils.schema_util import messages_to_map
class MyApp(BaseApp):
    def __init__(self, config, calibration_path):
        super().__init__(config, calibration_path)
        self.state_mgmt = StateMgmt(self.config, self.calibration)
        self.tripwire_event = TripwireEvent(self.config, self.calibration)
        self.roi_event = ROIEvent(self.config, self.calibration)
    def process(self, frames, stats):
        behavior_batch = self.state_mgmt.process_batch(messages_to_map(messages))
        # Generate events from every trip state the batch produced
        events = []
        for trip in behavior_batch.trip_behaviors:
            events.extend(self.tripwire_event.get_events(trip))
            events.extend(self.roi_event.get_events(trip))
        self.write_behaviors(behavior_batch.behaviors_to_write)
        self.write_events(events)

Definition

The Behavior object (an entry of BehaviorBatch.trip_behaviors) contains the trajectory information including locations, start and end timestamps for each tracked object. The system processes this trajectory by:

1. Splitting it into overlapping tracklets of length 2 × tripwireMinPoints (the sensor-level tripwireMinPoints, default 5, so 10 points per tracklet)
2. For tripwires: Checking if tracklets have sufficient points before and after crossing the wire
3. For ROIs: Checking if tracklets have sufficient points both inside and outside the ROI boundary

ROI event detection mode

The app-level roiEventDetectionMode key controls how ROIEvent decides whether an object is inside an ROI:

API

def get_events(self, behavior: Behavior) -> List[Behavior]

Incidents

The Incidents framework detects violations in video frames and tracks them over time to generate confirmed incidents when violations persist beyond configured thresholds. This transforms transient violations (which may be false positives) into confirmed incidents that warrant attention.

Key Concepts

• Violation: A detected event from an enhanced frame (e.g., objects too close, unauthorized area access). Violations are instantaneous detections within a single frame.
• Violation State: Internal tracking of violation duration and continuity across frames.
• Incident: A violation that exceeds the configured time threshold and needs reporting.

For configuration details, see Configuration Parameters. For calibration schema details, refer to Calibration Schema.

Violation Types

The framework supports four types of violations that can be promoted to incidents:

Proximity Violation

The objects detected are clustered via distance calculation and reports the cluster of points/objectIds involved in the violation.

The image below shows a proximity violation between a humanoid and a person. The distance between the two objects is less than the proximity threshold.

Confined Area Violation

The confined area violation is detected via point-in-polygon calculation. The system detects if any object is outside their designated confined area.

The image below shows the confined area violation. There are two humanoids present outside their designated confined areas. confinedObjectTypes for the area is set to “Humanoid” object type.

Restricted Area Violation

The restricted area violation is also detected via point-in-polygon calculation. The system detects if any object is inside their designated restricted area.

The image below shows the restricted area violation. One person is not allowed to be in the restricted area but is present inside it. restrictedObjectTypes for the same area is set to “Person” object type.

FOV Count Violation

The FOV count violation is triggered when the number of objects of a specific type in the field of view exceeds the configured threshold. This is useful for crowd control and capacity management scenarios.

The violation is configured via fovCountViolationIncidentObjectThreshold (the count at which the violation fires, compared as count >= threshold) and fovCountViolationIncidentObjectType (object type to count, e.g., “Person”). Because the comparison is inclusive, “no more than N in view” is expressed as a threshold of N + 1.

Violation Detection API

from mdx.analytics.core.transform.calibration.calibration_e import CalibrationE
from mdx.analytics.core.transform.detection.proximity_detection import ProximityDetection
calibration = CalibrationE(app_config, calibration_file_path)
# Inside transform_frame:
# - Zone violations (restricted/confined) detected based on "restrictedObjectTypes" and "confinedObjectTypes"
# - FOV count computed per object type
enhanced_frames = [calibration.transform_frame(frame) for frame in batch_frames]
# Proximity violations detected separately via ProximityDetection
clusters, object_id_groups = ProximityDetection.cluster(center_objs, surrounding_objs, threshold)

Enhanced Frame Data

The transform_frame method enriches the frame with FOV counts, ROI counts, and social-distancing (proximity) data. Proximity violations are written to the socialDistancing field as an SD object: proximityDetections is the number of violating clusters, clusters holds the world-coordinate points of each cluster, and info carries the proximityViolation flag plus the proximityViolationObjects groups (object IDs joined with , within a cluster and | across clusters). When no violation is present, proximityDetections is 0, clusters is empty, and info is {"proximityViolation": "false"}.

{
  "fov": [
    {"type": "Person", "count": 5},
    {"type": "Vehicle", "count": 3}
  ],
  "rois": [
    {"id": "roi-1", "type": "Person", "count": 2},
    {"id": "roi-2", "type": "Vehicle", "count": 3}
  ],
  "socialDistancing": {
    "threshold": 1.8,
    "proximityDetections": 2,
    "clusters": [
      {"points": [{"x": 10.2, "y": 4.5}, {"x": 10.9, "y": 4.8}]},
      {"points": [{"x": 22.1, "y": 9.0}, {"x": 22.6, "y": 9.3}, {"x": 23.0, "y": 8.7}]}
    ],
    "info": {
      "proximityViolation": "true",
      "proximityViolationObjects": "1,2|3,4,5"
    }
  }
}

Incident Generation

Violations are tracked over time, and when they persist beyond the configured threshold, they become incidents.

Time Parameters

Each violation type has configurable timing parameters:

Configuration

{
  "app": [
    {"name": "incidentObjectTtl", "value": "3600"},
    {"name": "proximityViolationIncidentEnable", "value": "true"},
    {"name": "proximityViolationIncidentThreshold", "value": "1"},
    {"name": "proximityViolationIncidentExpirationWindow", "value": "0.5"},
    {"name": "restrictedAreaViolationIncidentEnable", "value": "true"},
    {"name": "restrictedAreaViolationIncidentThreshold", "value": "1"},
    {"name": "restrictedAreaViolationIncidentExpirationWindow", "value": "0.5"},
    {"name": "confinedAreaViolationIncidentEnable", "value": "true"},
    {"name": "confinedAreaViolationIncidentThreshold", "value": "1"},
    {"name": "confinedAreaViolationIncidentExpirationWindow", "value": "0.5"},
    {"name": "fovCountViolationIncidentEnable", "value": "true"},
    {"name": "fovCountViolationIncidentObjectThreshold", "value": "1"},
    {"name": "fovCountViolationIncidentThreshold", "value": "1"},
    {"name": "fovCountViolationIncidentExpirationWindow", "value": "0.5"},
    {"name": "fovCountViolationIncidentObjectType", "value": "Person"}
  ]
}

Incident Configuration Parameters

Usage

from mdx.analytics.core.stream.state.frame.frame_state_management import FrameStateMgmt
# Initialize (within BaseApp context)
frame_manager = FrameStateMgmt(self.config)
# Enhanced frames from calibration.transform_frame()
enhanced_frames = [self.calibration.transform_frame(f) for f in frames]
# Update with enhanced frames containing violation data
frame_manager.update_frames("sensor_001", enhanced_frames)
# Get all incidents
incidents = frame_manager.get_incidents("sensor_001")
# Get specific incident types
proximity_incidents = frame_manager.get_proximity_violation_incidents("sensor_001")
restricted_incidents = frame_manager.get_restricted_area_violation_incidents("sensor_001")
confined_incidents = frame_manager.get_confined_area_violation_incidents("sensor_001")
fov_count_incidents = frame_manager.get_fov_count_violation_incidents("sensor_001")

Incident Structure

For object-specific violations (Proximity, Restricted Area, Confined Area):

Incident(
    timestamp=datetime,              # Start time
    end=datetime,                    # End time
    sensorId="sensor_001",           # Sensor ID
    objectIds=["obj1", "obj2"],      # Objects involved
    category="Proximity Violation",  # Incident type
    place=Place(name="Location"),    # Location name
    info={
        "primaryObjectId": "obj1",   # Primary object (for object-specific violations)
        "roiId": "zone_1",           # ROI ID (for area violations)
        "isComplete": "true"         # Present for completed violations
    }
)

When object presence is tracked (e.g. Proximity Violation, FOV Count Violation), info may also include objectTimeline: a JSON string mapping object IDs to lists of {start, end} intervals (ISO 8601) during the violation window.

FOV Count Violation incidents use a single state per sensor (no primary object). objectIds lists all objects that contributed to the violation; info has no primaryObjectId and may include objectTimeline. Call get_incidents() regularly so completed states are cleared and memory stays bounded.

Data Flow

Raw Frames (objects) → Frame Enhancement (calculates violations) →
Enhanced Frames → FrameStateMgmt.update_frames():
  1. Filter frames by time
  2. Complete expired violations (gap > expiration_window)
  3. Update violation states from frame fields
→ get_incidents():
  1. Generate incidents from active states (duration >= threshold)
  2. Generate incidents from completed states
  3. Clear processed completed states

Sink

The pipeline supports multiple sink types for publishing processed data:

• Kafka Sink: Publishes data to Kafka topics.
• Redis Streams Sink: Publishes data to Redis Streams for lightweight deployments.
• MQTT Sink: Publishes data to MQTT topics for IoT-oriented deployments.

The sink type is configured via the sinkType parameter in the app configuration section. Set it to kafka, redisStream, or mqtt based on your deployment requirements.

Usage with BaseApp

When using BaseApp, the sink is automatically initialized and available via built-in write methods:

from mdx.analytics.core.app.app_base import BaseApp
from mdx.analytics.core.utils.processing_stats import BatchStats
class MyApp(BaseApp):
    def process(self, frames, stats: BatchStats):
        # Write enhanced frames
        enhanced = [self.calibration.transform_frame(f) for f in frames]
        self.write_frames(enhanced)
        # Write behaviors
        self.write_behaviors(behaviors)
        # Write events (tripwire, ROI)
        self.write_events(events)
        # Write incidents from violation detection
        self.write_incidents(incidents)
        # Write anomalies
        self.write_anomalies(anomalies)
        # Write behaviors with trajectory clustering
        self.write_behaviors_with_clustering(behaviors)
        # Write space utilization data
        self.write_space_utilization(space_utilization_data)

Standalone Sink Usage

For standalone usage without BaseApp, use the sink factory:

from mdx.analytics.core.stream.sink.sink_factory import get_sink
from mdx.analytics.core.stream.sink.sink_base import ProtoBytesSerializer, StrBytesSerializer
from mdx.analytics.core.schema.config import AppConfig
config = AppConfig(**config_data)
sink = get_sink(config)  # Returns SinkKafka, SinkRedisStream, or SinkMQTT
# Write protobuf messages to a destination
sink.write(
    dest_key="behavior",
    messages=behavior_protos,
    value_serializer=ProtoBytesSerializer,
    key_extractor=lambda b: b.sensor.id,
    key_serializer=StrBytesSerializer
)
# Close when done
sink.close()

API

BaseApp write methods (recommended):

class BaseApp:
    # Write enhanced frames to the frames topic
    def write_frames(self, frames: list[nvSchema.Frame]) -> None
    # Write behaviors to the behavior topic
    def write_behaviors(self, behaviors: list[Behavior]) -> None
    # Write events to the events topic
    def write_events(self, events: list[Behavior]) -> None
    # Write incidents to the incidents topic
    def write_incidents(self, incidents: list[Incident]) -> None
    # Write anomalies to the anomaly topic
    def write_anomalies(self, anomalies: list[Behavior]) -> None
    # Write space utilization data
    def write_space_utilization(self, output: list[SpaceUtilization]) -> None
    # Write behaviors with clustering to behaviorPlus topic
    def write_behaviors_with_clustering(self, behaviors: list[Behavior]) -> None
    # Write filtered/downsampled video embeddings to the embedFiltered destination
    def write_embed_filtered(self, video_embeddings: list[nvSchema.VisionLLM]) -> None
    # Generic escape hatch: write any protobuf list to any logical destination key
    def write_proto(self, dest_key: str, msgs: list[Any], key_extractor: Callable | None = None) -> None