Documentation Index
Fetch the complete documentation index at: https://api-docs.rhombus.community/llms.txt
Use this file to discover all available pages before exploring further.
Esta página fue traducida automáticamente. Si encuentra errores o tiene sugerencias, contáctenos. Tópico de eventos
Todos los eventos organizacionales se publican en un único tópico:
Cada operación de creación, actualización y eliminación en tu organización Rhombus emite un evento en este tópico.
Estructura de la carga útil del evento
Cada frame MESSAGE contiene un cuerpo JSON con los siguientes campos:
{
"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"
}
Campos principales
| Campo | Tipo | Descripción |
|---|
entity | string | El tipo de entidad que cambió |
entityUuid | string | Identificador único de la entidad específica |
type | string | Tipo de cambio: CREATE, UPDATE o DELETE |
deviceUuid | string | UUID del dispositivo asociado (cuando aplica) |
timestampMs | integer | Timestamp del evento en milisegundos Unix |
Campos específicos de alertas
Estos campos están presentes cuando entity es POLICY_ALERT:
| Campo | Tipo | Descripción |
|---|
durationSec | float | Duración del evento de alerta en segundos |
policyAlertTriggers | array | Lista de tipos de disparadores que causaron la alerta |
textDescription | string | Descripción legible de la alerta |
Tipos de cambio
| Tipo | Descripción | Ejemplo |
|---|
CREATE | Se creó una nueva entidad | Nueva alerta de política disparada |
UPDATE | Se modificó una entidad existente | Estado de la alerta cambió |
DELETE | Se eliminó una entidad | Alerta descartada o limpiada |
Tipos de entidad
El campo entity indica qué tipo de objeto cambió. Los tipos de entidad comunes incluyen:
| Entidad | Descripción |
|---|
POLICY_ALERT | Alerta por violación de política de seguridad |
DEVICE_CONFIG | Cambio de configuración de cámara o sensor |
Los tipos de entidad disponibles dependen de los dispositivos y políticas configurados en tu organización. Conéctate al WebSocket y suscríbete sin un filtro de entidad para descubrir todos los tipos de evento disponibles.
Comienza filtrando los eventos POLICY_ALERT, que son el caso de uso más común. Para recibir todos los tipos de evento, omite el filtro de tipo de entidad de tu destino SUBSCRIBE.
Disparadores de alertas de política
Cuando un evento tiene entity: "POLICY_ALERT", el array policyAlertTriggers indica qué causó la alerta:
| Disparador | Descripción |
|---|
PERSON_DETECTED | Se detectó una persona en el campo de visión de la cámara |
LOITERING | Una persona permaneció en un área más allá del umbral configurado |
VEHICLE_DETECTED | Se detectó un vehículo |
MOTION_DETECTED | Se detectó movimiento general |
LINE_CROSSING | Un objeto cruzó un límite virtual definido |
AREA_INTRUSION | Un objeto entró en un área restringida |
Filtrado de eventos
Por tipo de entidad
Filtra tipos de evento específicos para reducir el ruido:
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
Por tipo de cambio
Reacciona de forma distinta según si un evento fue creado, actualizado o eliminado:
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)
Por dispositivo
Filtra eventos de una cámara o sensor específico:
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)
Por disparador de alerta
Reacciona a tipos específicos de eventos de seguridad:
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)
Enriquecer eventos con datos de la API REST
Los eventos WebSocket contienen datos mínimos por eficiencia. Usa la API REST para obtener detalles adicionales cuando los necesites:
Obtener el nombre de la cámara desde el UUID del dispositivo
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"
Obtener detalles de la alerta
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()
Almacena los nombres de cámaras en caché localmente para evitar llamadas excesivas a la API REST. Los nombres de las cámaras cambian con poca frecuencia, así que un caché con TTL de 5 minutos funciona bien.
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()
Salida JSON sin procesar
Para canalizar a otras herramientas o sistemas de logging:
import json
def output_json(payload):
print(json.dumps(payload, indent=2))
Patrones de integración
Reenvío a webhook
Reenvía los eventos de Rhombus a tu propio endpoint webhook:
import requests
WEBHOOK_URL = "https://your-server.com/webhooks/rhombus"
def relay_to_webhook(payload):
try:
requests.post(WEBHOOK_URL, json=payload, timeout=5)
except requests.RequestException as e:
print(f"Webhook relay failed: {e}")
Notificaciones de Slack
Envía alertas de alta prioridad a un canal de Slack:
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')}"
}
try:
requests.post(webhook_url, json=message, timeout=5)
except requests.RequestException as e:
print(f"Slack notification failed: {e}")
Registro en base de datos
Persiste eventos para análisis histórico:
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()
