Skip to main content

​
Event Monitoring

The Rhombus WebSocket event stream delivers real-time notifications about everything happening in your organization. This guide covers the event structure, available entity types, and patterns for filtering and processing events.

​
Event Topic

All organizational events are published to a single topic: 
/topic/change/{orgUuid}
Every create, update, and delete operation across your Rhombus organization emits an event on this topic.

​
Event Payload Structure

Each MESSAGE frame contains a JSON body with the following fields: 
{
  "entity": "POLICY_ALERT",
  "entityUuid": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "type": "CREATE",
  "deviceUuid": "d1e2f3a4-b5c6-7890-abcd-ef1234567890",
  "timestampMs": 1711843200000,
  "durationSec": 5.2,
  "policyAlertTriggers": ["PERSON_DETECTED", "LOITERING"],
  "textDescription": "Person detected near entrance"
}

​
Core Fields

FieldTypeDescription
entitystringThe type of entity that changed
entityUuidstringUnique identifier for the specific entity
typestringChange type: CREATE, UPDATE, or DELETE
deviceUuidstringAssociated device UUID (when applicable)
timestampMsintegerEvent timestamp in Unix milliseconds

​
Alert-Specific Fields

These fields are present when entity is POLICY_ALERT:
FieldTypeDescription
durationSecfloatDuration of the alert event in seconds
policyAlertTriggersarrayList of trigger types that caused the alert
textDescriptionstringHuman-readable description of the alert

​
Change Types

TypeDescriptionExample
CREATEA new entity was createdNew policy alert triggered
UPDATEAn existing entity was modifiedAlert status changed
DELETEAn entity was removedAlert dismissed or cleared

​
Entity Types

The entity field indicates what kind of object changed. Common entity types include:
EntityDescription
POLICY_ALERTSecurity policy violation alert
DEVICE_CONFIGCamera or sensor configuration change
Other entitiesAdditional entity types may appear as new features are added
Start by filtering for POLICY_ALERT events, which are the most common use case. Use the --all-events flag equivalent in your code when you need to capture every event type.

​
Policy Alert Triggers

When an event has entity: "POLICY_ALERT", the policyAlertTriggers array indicates what caused the alert:
TriggerDescription
PERSON_DETECTEDA person was detected in the camera’s field of view
LOITERINGA person remained in an area beyond the configured threshold
VEHICLE_DETECTEDA vehicle was detected
MOTION_DETECTEDGeneral motion was detected
LINE_CROSSINGAn object crossed a defined virtual boundary
AREA_INTRUSIONAn object entered a restricted area

​
Filtering Events

​
By Entity Type

Filter for specific event types to reduce noise: 
def handle_message(payload):
    entity = payload.get("entity")

    if entity == "POLICY_ALERT":
        handle_alert(payload)
    elif entity == "DEVICE_CONFIG":
        handle_device_change(payload)
    else:
        # Log or ignore other entity types
        pass

​
By Change Type

React differently based on whether an event was created, updated, or deleted: 
def handle_alert(payload):
    change_type = payload.get("type")

    if change_type == "CREATE":
        # New alert - send notification
        send_notification(payload)
    elif change_type == "UPDATE":
        # Alert updated - refresh dashboard
        refresh_dashboard(payload)
    elif change_type == "DELETE":
        # Alert cleared - close ticket
        close_ticket(payload)

​
By Device

Filter events for a specific camera or sensor: 
WATCHED_DEVICES = {
    "camera-uuid-lobby",
    "camera-uuid-parking-lot",
}

def handle_message(payload):
    device_uuid = payload.get("deviceUuid")
    if device_uuid in WATCHED_DEVICES:
        process_event(payload)

​
By Alert Trigger

React to specific types of security events: 
HIGH_PRIORITY_TRIGGERS = {"PERSON_DETECTED", "AREA_INTRUSION", "LINE_CROSSING"}

def handle_alert(payload):
    triggers = set(payload.get("policyAlertTriggers", []))
    if triggers & HIGH_PRIORITY_TRIGGERS:
        send_urgent_notification(payload)

​
Enriching Events with REST API Data

WebSocket events contain minimal data for efficiency. Use the REST API to fetch additional details when needed:

​
Get Camera Name from Device UUID

import requests

def get_camera_name(api_token, device_uuid):
    """Fetch camera name for display purposes."""
    response = requests.post(
        "https://api2.rhombussystems.com/api/camera/getMinimalCameraStateList",
        headers={
            "x-auth-apikey": api_token,
            "Content-Type": "application/json"
        },
        json={}
    )
    cameras = response.json().get("cameraStates", [])
    for camera in cameras:
        if camera.get("uuid") == device_uuid:
            return camera.get("name", "Unknown Camera")
    return "Unknown Camera"

​
Get Alert Details

def get_alert_details(api_token, alert_uuid):
    """Fetch full alert details from the REST API."""
    response = requests.post(
        "https://api2.rhombussystems.com/api/alert/getAlert",
        headers={
            "x-auth-apikey": api_token,
            "Content-Type": "application/json"
        },
        json={"alertUuid": alert_uuid}
    )
    return response.json()
Cache camera names locally to avoid excessive REST API calls. Camera names change infrequently, so a cache with a 5-minute TTL works well.

​
Output Formats

​
Formatted Alert Display

from datetime import datetime

def display_alert(payload, camera_name=""):
    ts = datetime.fromtimestamp(payload["timestampMs"] / 1000)
    triggers = ", ".join(payload.get("policyAlertTriggers", []))
    duration = payload.get("durationSec", 0)
    description = payload.get("textDescription", "")

    print(f"[{ts:%H:%M:%S}] ALERT  camera={camera_name}")
    print(f"  triggers={triggers}  duration={duration}s")
    if description:
        print(f"  {description}")
    print(f"  uuid={payload['entityUuid']}")
    print()

​
Raw JSON Output

For piping to other tools or logging systems: 
import json

def output_json(payload):
    print(json.dumps(payload, indent=2))

​
Integration Patterns

​
Webhook Relay

Forward Rhombus events to your own webhook endpoint: 
import requests

WEBHOOK_URL = "https://your-server.com/webhooks/rhombus"

def relay_to_webhook(payload):
    requests.post(WEBHOOK_URL, json=payload, timeout=5)

​
Slack Notifications

Send high-priority alerts to a Slack channel: 
def send_slack_alert(payload, webhook_url):
    triggers = ", ".join(payload.get("policyAlertTriggers", []))
    message = {
        "text": f":rotating_light: *Security Alert*\n"
                f"Triggers: {triggers}\n"
                f"{payload.get('textDescription', 'No description')}"
    }
    requests.post(webhook_url, json=message)

​
Database Logging

Persist events for historical analysis: 
import sqlite3
import json

def log_event(db_path, payload):
    conn = sqlite3.connect(db_path)
    conn.execute("""
        INSERT INTO events (entity, entity_uuid, change_type, device_uuid, timestamp_ms, payload)
        VALUES (?, ?, ?, ?, ?, ?)
    """, (
        payload.get("entity"),
        payload.get("entityUuid"),
        payload.get("type"),
        payload.get("deviceUuid"),
        payload.get("timestampMs"),
        json.dumps(payload)
    ))
    conn.commit()
    conn.close()