Alarms

Alarms let you define rules that trigger when a channel's aggregate data meets a condition. For example, you can create an alarm that fires when a temperature reading exceeds a threshold. The DataClient provides full CRUD operations for managing alarms on channels.

Alarm Concepts

Each alarm is defined by:

name -- a human-readable label for the alarm
key -- the field name in the channel's aggregate data to monitor
operator -- the comparison operator (eq, gt, ge, lt, le)
value -- the threshold value to compare against
enabled -- whether the alarm is active
expiry_mins -- optional duration in minutes after which the alarm auto-clears

Alarms have a state that indicates their current status:

State	Description
`NoData`	No data has been received for the monitored key
`OK`	The condition is not met (value is within bounds)
`Alarm`	The condition is met (threshold exceeded)

AlarmOperator

The AlarmOperator enum defines the comparison operators available for alarms:

Operator	Meaning	Example
`eq`	Equal to	Value equals threshold
`gt`	Greater than	Value exceeds threshold
`ge`	Greater than or equal	Value meets or exceeds threshold
`lt`	Less than	Value below threshold
`le`	Less than or equal	Value at or below threshold

Listing Alarms

Retrieve all alarms defined on a channel:

from pydoover.api import DataClient

client = DataClient(profile="default")

alarms = client.list_alarms(
    agent_id=12345,
    channel_name="telemetry",
)

for alarm in alarms:
    print(f"{alarm.name}: {alarm.key} {alarm.operator.value} {alarm.value} "
          f"[state={alarm.state.value}]")

Fetching a Single Alarm

Retrieve a specific alarm by its ID:

alarm = client.fetch_alarm(
    agent_id=12345,
    channel_name="telemetry",
    alarm_id=9876,
)

print(f"Alarm '{alarm.name}' is {alarm.state.value}")
print(f"Entered current state at: {alarm.entered_state_ts}")

Creating an Alarm

Create an alarm that triggers when a condition is met. This example creates an alarm that fires when the temperature field in the channel's aggregate exceeds 85 degrees:

from pydoover.models.data.alarm import AlarmOperator

alarm = client.create_alarm(
    agent_id=12345,
    channel_name="telemetry",
    name="High Temperature",
    key="temperature",
    operator=AlarmOperator.gt,
    value=85.0,
    description="Triggers when temperature exceeds 85 degrees",
    enabled=True,
    expiry_mins=30.0,
)

print(f"Created alarm {alarm.id}: {alarm.name}")

The expiry_mins parameter is optional. When set, the alarm automatically returns to the OK state after the specified number of minutes, even if the condition is still met. This is useful for transient conditions where you want the alarm to reset periodically.

Replacing an Alarm (PUT)

Replace an alarm's entire configuration. This requires all fields to be provided:

alarm = client.put_alarm(
    agent_id=12345,
    channel_name="telemetry",
    alarm_id=9876,
    name="High Temperature (Updated)",
    key="temperature",
    operator=AlarmOperator.ge,
    value=90.0,
    description="Updated threshold to 90 degrees",
    enabled=True,
)

Updating an Alarm (PATCH)

Partially update an alarm. Only the fields you provide are modified:

# Disable the alarm without changing anything else
alarm = client.update_alarm(
    agent_id=12345,
    channel_name="telemetry",
    alarm_id=9876,
    enabled=False,
)

The update_alarm method uses UNSET sentinel values internally to distinguish between "not provided" and None. This matters for expiry_mins, where you might want to explicitly clear it:

from pydoover.api.data._base import UNSET

# Remove the expiry by setting it to None
alarm = client.update_alarm(
    agent_id=12345,
    channel_name="telemetry",
    alarm_id=9876,
    expiry_mins=None,  # Explicitly clears the expiry
)

# Compare with omitting it entirely (leaves the current value unchanged)
alarm = client.update_alarm(
    agent_id=12345,
    channel_name="telemetry",
    alarm_id=9876,
    name="Renamed Alarm",
    # expiry_mins is UNSET by default, so not modified
)

Deleting an Alarm

Remove an alarm from a channel:

client.delete_alarm(
    agent_id=12345,
    channel_name="telemetry",
    alarm_id=9876,
)

Alarm Method Reference

Method	Returns	Description
`list_alarms(agent_id, channel_name)`	`list[Alarm]`	List all alarms on a channel
`fetch_alarm(agent_id, channel_name, alarm_id)`	`Alarm`	Fetch a single alarm by ID
`create_alarm(agent_id, channel_name, ...)`	`Alarm`	Create a new alarm
`put_alarm(agent_id, channel_name, alarm_id, ...)`	`Alarm`	Replace an alarm entirely
`update_alarm(agent_id, channel_name, alarm_id, ...)`	`Alarm`	Partially update an alarm
`delete_alarm(agent_id, channel_name, alarm_id)`	--	Delete an alarm

Connections

The DataClient also provides methods for monitoring WebSocket connections. Connections represent active or historical WebSocket sessions between agents and the Doover data plane.

Listing Connections

Retrieve all active WebSocket connections for an agent:

connections = client.list_connections(agent_id=12345)

for conn in connections:
    print(f"Connection {conn.id}")

Fetching a Connection

Retrieve details for a specific connection by its ID:

connection = client.fetch_connection(connection_id=5555)

Connection History

Retrieve historical connection records for an agent, optionally filtered by time range:

from datetime import datetime, timezone, timedelta

now = datetime.now(timezone.utc)
one_week_ago = now - timedelta(days=7)

history = client.fetch_connection_history(
    agent_id=12345,
    default_connection=True,
    after=one_week_ago,
    before=now,
    limit=100,
)

for conn in history:
    print(f"Connection {conn.id}")

Subscription History

View the subscription history for a specific channel on a specific connection:

sub_history = client.fetch_subscription_history(
    agent_id=12345,
    channel_agent_id=12345,
    channel_name="telemetry",
    limit=50,
)

for entry in sub_history:
    print(entry)

Channel Subscriptions

List active WebSocket subscriptions for a specific channel:

subscriptions = client.fetch_channel_subscriptions(
    agent_id=12345,
    channel_name="telemetry",
)

for sub in subscriptions:
    print(sub)

TURN Token

Fetch a TURN server credential for WebRTC camera streaming:

turn = client.fetch_turn_token(
    role="viewer",
    camera_name="front_camera",
    device_id=12345,
)

print(f"TURN URL: {turn.url}")

Connection Method Reference

Method	Returns	Description
`list_connections(agent_id)`	`list[ConnectionDetail]`	List active connections
`fetch_connection(connection_id)`	`ConnectionDetail`	Fetch a single connection
`fetch_connection_history(agent_id, ...)`	`list[ConnectionDetail]`	Historical connection records
`fetch_subscription_history(agent_id, ...)`	`list[ConnectionSubscriptionLog]`	Subscription change history
`fetch_channel_subscriptions(agent_id, channel_name)`	`list[ConnectionSubscription]`	Active channel subscriptions
`fetch_turn_token(role, camera_name, ...)`	`TurnCredential`	TURN server credentials

Channels -- channel management (alarms are defined per channel)
Notifications -- notification system for alerting users
Data Client -- DataClient constructor and configuration