The Rules resource allows you to create and manage automation rules that monitor your fleet for specific conditions and trigger actions (such as creating alerts) when those conditions are met.
Methods
list
List all monitoring rules in your fleet.
rules = client.fleet.rules.list(
enabled=True,
limit=50,
cursor=None
)
Parameters
Filter rules by enabled status (True for active rules, False for disabled)
Maximum number of rules to return per page
Pagination cursor for fetching the next page
Returns
List of rule objectsDetailed description of what the rule monitors
Whether the rule is currently active
Condition configuration that triggers the rule
Actions to perform when the condition is met
Optional scope limiting which devices/recordings this rule applies to
Number of times this rule has been triggered
Timestamp of the most recent trigger
Whether more results are available
Cursor for fetching the next page
create
Create a new monitoring rule.
rule = client.fleet.rules.create(
name="High Temperature Alert",
condition={
"type": "threshold",
"field": "temperature",
"operator": ">",
"value": 80
},
actions=[
{
"type": "create_alert",
"severity": "critical",
"message": "Device temperature exceeded 80°C"
},
{
"type": "notify_channel",
"channel_id": "chan_abc123"
}
],
description="Alert when device temperature exceeds safe operating limits",
enabled=True,
scope={
"device_types": ["autonomous_vehicle"],
"tags": ["production"]
}
)
Parameters
Condition configuration defining when the rule triggers. Structure varies by condition type.
List of actions to perform when the condition is met. Each action is a dictionary with a type field and type-specific parameters.
Detailed description of the rule’s purpose
Whether the rule should be active immediately (default: true)
Scope limiting which devices/recordings this rule applies to. Can include:
device_ids: List of specific device IDs
device_types: List of device types
tags: List of required tags
Returns
Returns a Rule object with all fields populated.
get
Retrieve details for a specific rule.
rule = client.fleet.rules.get("rule_def456")
Parameters
The unique identifier of the rule
Returns
Returns a Rule object.
update
Update an existing rule’s configuration.
rule = client.fleet.rules.update(
"rule_def456",
name="High Temperature Alert - Updated",
enabled=False,
condition={
"type": "threshold",
"field": "temperature",
"operator": ">",
"value": 85 # Increased threshold
},
description="Updated temperature threshold to 85°C"
)
Parameters
The unique identifier of the rule to update
Updated name for the rule
Enable or disable the rule
Updated condition configuration
Updated scope configuration
Returns
Returns the updated Rule object.
delete
Permanently delete a rule.
client.fleet.rules.delete("rule_def456")
Parameters
The unique identifier of the rule to delete
Returns
Returns None on success.
Example Usage
from avala import Avala
client = Avala(api_key="your-api-key")
# Create a rule for detecting hard braking events
hard_braking_rule = client.fleet.rules.create(
name="Hard Braking Detection",
description="Alert when hard braking is detected",
condition={
"type": "event_detection",
"event_type": "safety",
"event_label": "Hard Braking",
"severity": ["high", "critical"]
},
actions=[
{
"type": "create_alert",
"severity": "high",
"message": "Hard braking event detected"
}
],
enabled=True,
scope={
"device_types": ["autonomous_vehicle"],
"tags": ["production"]
}
)
print(f"Created rule: {hard_braking_rule.uid}")
# Create a rule for device offline detection
offline_rule = client.fleet.rules.create(
name="Device Offline",
description="Alert when a device hasn't communicated in 1 hour",
condition={
"type": "device_offline",
"duration_minutes": 60
},
actions=[
{
"type": "create_alert",
"severity": "critical",
"message": "Device has been offline for over 1 hour"
},
{
"type": "notify_channel",
"channel_id": "chan_ops_team"
}
],
enabled=True
)
# List all enabled rules
active_rules = client.fleet.rules.list(enabled=True)
print(f"\nActive rules ({len(active_rules.data)}):")
for rule in active_rules.data:
print(f" - {rule.name}")
print(f" Hit count: {rule.hit_count}")
if rule.last_hit_at:
print(f" Last triggered: {rule.last_hit_at}")
# Update a rule to temporarily disable it
updated = client.fleet.rules.update(
hard_braking_rule.uid,
enabled=False,
description="Temporarily disabled for testing"
)
print(f"\nRule '{updated.name}' disabled")
# Get detailed rule information
rule_detail = client.fleet.rules.get(offline_rule.uid)
print(f"\nRule: {rule_detail.name}")
print(f"Condition: {rule_detail.condition}")
print(f"Actions: {rule_detail.actions}")
print(f"Scope: {rule_detail.scope}")
Async Usage
from avala import AsyncAvala
import asyncio
async def manage_rules():
client = AsyncAvala(api_key="your-api-key")
# Create rule
rule = await client.fleet.rules.create(
name="Async Rule",
condition={"type": "threshold", "field": "speed", "operator": ">", "value": 70},
actions=[{"type": "create_alert", "severity": "medium"}]
)
# List rules
rules = await client.fleet.rules.list()
# Update rule
updated = await client.fleet.rules.update(
rule.uid,
enabled=False
)
# Delete rule
await client.fleet.rules.delete(rule.uid)
asyncio.run(manage_rules())
Condition Types
Rules support various condition types:
Threshold Condition
Triggers when a metric crosses a threshold:
condition = {
"type": "threshold",
"field": "temperature",
"operator": ">", # >, <, >=, <=, ==, !=
"value": 80
}
Event Detection Condition
Triggers when specific events are detected:
condition = {
"type": "event_detection",
"event_type": "safety",
"event_label": "Hard Braking",
"severity": ["high", "critical"]
}
Device Offline Condition
Triggers when a device stops communicating:
condition = {
"type": "device_offline",
"duration_minutes": 60
}
Action Types
Rules can trigger various actions:
Create Alert
action = {
"type": "create_alert",
"severity": "critical",
"message": "Custom alert message"
}
Notify Channel
action = {
"type": "notify_channel",
"channel_id": "chan_abc123"
}
Webhook
action = {
"type": "webhook",
"url": "https://example.com/webhook",
"method": "POST"
}