Skip to main content
The Events resource allows you to create, retrieve, and manage events that occur within your fleet recordings. Events represent significant moments or occurrences that you want to track and analyze.

Methods

list

Query events with flexible filtering options.
events = client.fleet.events.list(
    recording_id="rec_abc123",
    device_id="dev_xyz789",
    type="safety",
    severity="high",
    limit=50,
    cursor=None
)

Parameters

recording_id
string
Filter events by recording ID
device_id
string
Filter events by device ID
type
string
Filter events by type (e.g., “safety”, “anomaly”, “detection”)
severity
string
Filter events by severity level (e.g., “low”, “medium”, “high”, “critical”)
limit
integer
Maximum number of events to return per page
cursor
string
Pagination cursor for fetching the next page

Returns

data
list[FleetEvent]
List of event objects
uid
string
Unique event identifier
recording
string
Recording ID this event belongs to
device
string
Device ID that created this event
type
string
Event type category
label
string
Human-readable event label
description
string
Detailed event description
timestamp
datetime
When the event occurred
duration_ms
integer
Event duration in milliseconds
severity
string
Event severity level
tags
list[string]
Tags associated with the event
metadata
dict
Custom metadata dictionary
created_at
datetime
Event creation timestamp
updated_at
datetime
Last update timestamp
has_more
boolean
Whether more results are available
next_cursor
string
Cursor for fetching the next page

create

Create a new event for a recording.
event = client.fleet.events.create(
    recording="rec_abc123",
    device="dev_xyz789",
    label="Hard Braking Detected",
    type="safety",
    timestamp="2024-01-15T14:32:15Z",
    severity="high",
    description="Sudden deceleration detected at 45 mph to 15 mph",
    duration_ms=2500,
    tags=["automated", "safety-critical"],
    metadata={
        "initial_speed": 45,
        "final_speed": 15,
        "g_force": -0.8
    }
)

Parameters

recording
string
required
ID of the recording this event belongs to
device
string
required
ID of the device that generated this event
label
string
required
Short, human-readable label for the event
type
string
required
Event type category
timestamp
string
required
ISO 8601 timestamp when the event occurred
severity
string
Severity level (e.g., “low”, “medium”, “high”, “critical”)
description
string
Detailed description of the event
duration_ms
integer
Duration of the event in milliseconds
tags
list[string]
Tags for categorizing the event
metadata
dict
Custom metadata as key-value pairs

Returns

Returns a FleetEvent object with all fields populated.

create_batch

Create multiple events in a single request for better performance.
events = [
    {
        "recording": "rec_abc123",
        "device": "dev_xyz789",
        "label": "Object Detected",
        "type": "detection",
        "timestamp": "2024-01-15T14:30:00Z",
        "severity": "low"
    },
    {
        "recording": "rec_abc123",
        "device": "dev_xyz789",
        "label": "Lane Departure",
        "type": "safety",
        "timestamp": "2024-01-15T14:31:30Z",
        "severity": "medium"
    }
]

created_count = client.fleet.events.create_batch(events=events)
print(f"Created {created_count} events")

Parameters

events
list[BatchEventParams]
required
List of event parameter dictionaries. Each event must have:
  • recording (string, required): Recording ID
  • device (string, required): Device ID
  • label (string, required): Event label
  • type (string, required): Event type
  • timestamp (string, required): ISO 8601 timestamp
  • severity (string, optional): Severity level
  • description (string, optional): Description
  • duration_ms (integer, optional): Duration in milliseconds
  • tags (list[string], optional): Tags
  • metadata (dict, optional): Custom metadata

Returns

Returns an integer representing the number of events successfully created.

get

Retrieve details for a specific event.
event = client.fleet.events.get("evt_def456")

Parameters

event_id
string
required
The unique identifier of the event

Returns

Returns a FleetEvent object.

delete

Permanently delete an event.
client.fleet.events.delete("evt_def456")

Parameters

event_id
string
required
The unique identifier of the event to delete

Returns

Returns None on success.

Example Usage

from avala import Avala

client = Avala(api_key="your-api-key")

# Create a single event
event = client.fleet.events.create(
    recording="rec_abc123",
    device="dev_xyz789",
    label="Obstacle Detected",
    type="detection",
    timestamp="2024-01-15T14:32:15Z",
    severity="medium",
    metadata={
        "object_type": "pedestrian",
        "distance_meters": 15.5,
        "confidence": 0.95
    }
)

print(f"Event created: {event.uid}")

# Batch create multiple events
events_to_create = [
    {
        "recording": "rec_abc123",
        "device": "dev_xyz789",
        "label": "Traffic Light - Red",
        "type": "traffic_signal",
        "timestamp": "2024-01-15T14:30:00Z"
    },
    {
        "recording": "rec_abc123",
        "device": "dev_xyz789",
        "label": "Turn Signal Activated",
        "type": "vehicle_action",
        "timestamp": "2024-01-15T14:30:45Z"
    }
]

created = client.fleet.events.create_batch(events=events_to_create)
print(f"Batch created {created} events")

# Query events by severity
high_severity = client.fleet.events.list(
    recording_id="rec_abc123",
    severity="high"
)

for evt in high_severity.data:
    print(f"{evt.label} at {evt.timestamp}: {evt.description}")

# Get specific event
event_detail = client.fleet.events.get(event.uid)
print(f"Event type: {event_detail.type}")

Async Usage

from avala import AsyncAvala
import asyncio

async def process_events():
    client = AsyncAvala(api_key="your-api-key")
    
    # Create event
    event = await client.fleet.events.create(
        recording="rec_abc123",
        device="dev_xyz789",
        label="Async Event",
        type="test",
        timestamp="2024-01-15T14:32:15Z"
    )
    
    # List events
    events = await client.fleet.events.list(type="safety")
    
    # Batch create
    count = await client.fleet.events.create_batch(
        events=[...]
    )
    
asyncio.run(process_events())

Build docs developers (and LLMs) love