roboto.domain.events

Clear a single custom-field value on this event to None.

Clear multiple custom-field values on this event to None.

Create a new event associated with at least one dataset, file, topic, or message path.

Creates a time-anchored event that can be associated with various Roboto entities. For instantaneous events (a point in time), only start_time is required. Otherwise, both start_time and end_time should be provided. These fields accept nanoseconds since the UNIX epoch, or any other compatible representations supported by to_epoch_nanoseconds().

Events must be associated with at least one entity. While associations, file_ids, topic_ids, dataset_ids and message_path_ids are all optional, at least one of them must contain a valid association for the event.

Usage

Create an event for a sensor anomaly on a specific topic:

from roboto.domain.events import Event
event = Event.create(
    name="Temperature Spike",
    start_time=1722870127699468923,
    end_time=1722870127799468923,
    description="Unusual temperature readings detected",
    topic_ids=["tp_abc123"],
    tags=["anomaly", "temperature"],
    metadata={"severity": "high", "sensor_id": "temp_01"},
)

Create an instantaneous event on a file:

event = Event.create(
    name="System Boot",
    start_time="1722870127.699468923",  # String format also supported
    file_ids=["fl_xyz789"],
    tags=["system", "boot"],
)

Create an event with display options:

from roboto.domain.events import EventDisplayOptions
event = Event.create(
    name="Critical Alert",
    start_time=1722870127699468923,
    end_time=1722870127799468923,
    dataset_ids=["ds_abc123"],
    display_options=EventDisplayOptions(color="red"),
    metadata={"alert_type": "critical", "component": "engine"},
)

Get dataset IDs associated with this event.

Usage

Get all associated dataset IDs:

event = Event.from_id("ev_abc123")
dataset_ids = event.dataset_ids()
print(f"Associated with {len(dataset_ids)} datasets")

Get only directly associated datasets:

strict_dataset_ids = event.dataset_ids(strict_associations=True)
print(f"Directly associated with {len(strict_dataset_ids)} datasets")

Delete this event permanently.

This operation cannot be undone. The event and all its associations will be permanently removed from the platform.

Usage

Delete an event:

event = Event.from_id("ev_abc123")
event.delete()
# Event is now permanently deleted

Conditional deletion:

event = Event.from_id("ev_abc123")
if "temporary" in event.tags:
    event.delete()
    print("Temporary event deleted")

Delete multiple events.

Authorization works just like delete(): you can delete any event you’re able to manage. If the list includes an event you’re not allowed to delete, the whole request is rejected and nothing is deleted. This operation cannot be undone. Event IDs that don’t exist are ignored, so the call is idempotent and safe to retry.

The bulk delete API caps each request at MAX_EVENTS_PER_DELETE_BATCH event IDs. This method removes that limit for callers by splitting event_ids into chunks of that size and sending one request per chunk. Because each chunk is its own request, deleting a very large number of events is not atomic: if a request fails partway through, earlier chunks stay deleted. The operation is idempotent, so retrying with the same IDs safely finishes the job.

Usage

Delete several events at once:

from roboto.domain.events import Event
Event.delete_many(["ev_abc123", "ev_def456", "ev_ghi789"])

Delete the events surfaced by a query:

events = list(Event.get_by_dataset("ds_abc123"))
Event.delete_many([event.event_id for event in events])

Get file IDs associated with this event.

Usage

Get all associated file IDs:

event = Event.from_id("ev_abc123")
file_ids = event.file_ids()
print(f"Associated with {len(file_ids)} files")

Get only directly associated files:

strict_file_ids = event.file_ids(strict_associations=True)
for file_id in strict_file_ids:
    print(f"Directly associated file: {file_id}")

Load an existing event by its ID.

Usage

Load an event by ID:

event = Event.from_id("ev_abc123")
print(f"Event: {event.name}")
print(f"Created: {event.created}")

Load and update an event:

event = Event.from_id("ev_abc123")
updated_event = event.set_description("Updated description")
print(f"New description: {updated_event.description}")

Retrieve all events associated with the provided associations.

Returns events that match any of the provided associations. Events that you don’t have access to will be filtered out of the response rather than raising an exception.

Usage

Query events for multiple associations:

from roboto import Association
associations = [Association.topic("tp_abc123"), Association.file("fl_xyz789")]
events = list(Event.get_by_associations(associations))
for event in events:
    print(f"Event: {event.name}")

Query events for a specific dataset and file combination:

associations = [Association.dataset("ds_abc123"), Association.file("fl_xyz789")]
events = list(Event.get_by_associations(associations))

Retrieve all events associated with a specific dataset.

Returns events that are associated with the given dataset. By default, this includes events associated with the dataset itself, as well as events associated with any files or topics within that dataset. Use strict_associations=True to only return events with direct dataset associations.

Usage

Get all events for a dataset (including file and topic events):

events = list(Event.get_by_dataset("ds_abc123"))
for event in events:
    print(f"Event: {event.name} at {event.start_time}")

Get only events directly associated with the dataset:

strict_events = list(Event.get_by_dataset("ds_abc123", strict_associations=True))
print(f"Found {len(strict_events)} dataset-level events")

Process events in batches:

for event in Event.get_by_dataset("ds_abc123"):
    if "anomaly" in event.tags:
        print(f"Anomaly event: {event.name}")

Retrieve all events with a direct association to a specific file.

Usage

Get all events for a specific file:

events = list(Event.get_by_file("fl_xyz789"))
for event in events:
    print(f"File event: {event.name}")

Check if a file has any events:

file_events = list(Event.get_by_file("fl_xyz789"))
if file_events:
    print(f"File has {len(file_events)} events")
else:
    print("No events found for this file")

Retrieve all events with a direct association to a specific message path.

Usage

Get all events for a specific message path:

events = list(Event.get_by_message_path("mp_abc123"))
for event in events:
    print(f"Message path event: {event.name}")

Find events within a time range for a message path:

events = Event.get_by_message_path("mp_abc123")
filtered_events = [event for event in events if event.start_time >= 1722870127699468923]

Retrieve all events with a direct association to a specific topic.

Usage

Get all events for a specific topic:

events = list(Event.get_by_topic("tp_abc123"))
for event in events:
    print(f"Topic event: {event.name}")

Analyze event patterns for a topic:

events = list(Event.get_by_topic("tp_abc123"))
anomaly_events = [e for e in events if "anomaly" in e.tags]
print(f"Found {len(anomaly_events)} anomaly events")

Iteratively yield records of the underlying topic data this event annotates.

An event can be associated with data at multiple resolutions:

• as an event on its containing dataset, file, and/or topic,
• but also directly with the message path (“signal data”)

A single event can also span signals that share a timeline, so it may annotate multiple topics in a file, or even multiple files in a dataset.

For now, getting the underlying signal data associated with an event only works for events that can be sourced to a single topic (extracted from one file, uploaded to one dataset). This means that the event must have been made on either a single file, a single topic, or one or many message paths within that topic.

If the event was made on a file, topic_name must be provided, and either or both of message_paths_include or message_paths_exclude may be provided, but are optional.

If the event was made on a topic, either or both of message_paths_include or message_paths_exclude may be provided, but are optional. topic_name, if provided in this instance, is ignored.

If the event was made on one or many message paths, each of those message paths must be found in the same topic. topic_name, message_paths_include, and message_paths_exclude, if provided in this instance, are ignored.

If the event is associated with data at multiple resolutions (e.g., two message paths, one topic, one file), this method will consider the lowest resolution associations first (message path), then topic, then file.

If message_paths_include or message_paths_exclude are defined, they should be dot notation paths that match attributes of individual data records. If a partial path is provided, it is treated as a wildcard, matching all subpaths.

For example, given topic data with the following interface:

{
    "velocity": {
        "x": <uint32>,
        "y": <uint32>,
        "z": <uint32>
    }
}

Calling get_data on an Event associated with that topic like:

event.get_data(message_paths_include=["velocity.x", "velocity.y"])

is expected to give the same output as:

event.get_data(message_paths_include=["velocity"], message_paths_exclude=["velocity.z"])

Return the underlying topic data this event annotates as a pandas DataFrame.

Collects all data from get_data() and returns it as a pandas DataFrame with the log time as the index. Requires installing this package using the roboto[analytics] extra.

Usage

Get event data as a DataFrame:

event = Event.from_id("ev_abc123")
df = event.get_data_as_df()
print(f"Data shape: {df.shape}")
print(df.head())

Get specific message paths as DataFrame:

df = event.get_data_as_df(message_paths_include=["velocity.x", "velocity.y"])
print(df.columns.tolist())

Analyze event data:

df = event.get_data_as_df()
print(f"Event duration: {df.index.max() - df.index.min()} ns")
print(f"Data points: {len(df)}")

Get message path IDs directly associated with this event.

Usage

Get message path IDs:

event = Event.from_id("ev_abc123")
msgpath_ids = event.message_path_ids()
print(f"Associated with {len(msgpath_ids)} message paths")

Add or update metadata fields for this event.

Usage

Add metadata to an event:

event = Event.from_id("ev_abc123")
updated_event = event.put_metadata({"severity": "high", "component": "engine", "alert_id": "alert_001"})
print(updated_event.metadata["severity"])
# 'high'

Replace all tags for this event.

Usage

Set tags for an event:

event = Event.from_id("ev_abc123")
updated_event = event.put_tags(["anomaly", "critical", "engine"])
print(updated_event.tags)
# ['anomaly', 'critical', 'engine']

Refresh this event’s data from the server.

Fetches the latest version of this event from the server, updating all properties to reflect any changes made by other processes.

Usage

Refresh an event to get latest changes:

event = Event.from_id("ev_abc123")
# Event may have been updated by another process
refreshed_event = event.refresh()
print(f"Current description: {refreshed_event.description}")

Remove metadata fields from this event.

Usage

Remove specific metadata fields:

event = Event.from_id("ev_abc123")
updated_event = event.remove_metadata(["severity", "temp_data.max"])
# Fields 'severity' and nested 'temp_data.max' are now removed

Remove specific tags from this event.

Usage

Remove specific tags:

event = Event.from_id("ev_abc123")
updated_event = event.remove_tags(["temporary", "draft"])
# Tags 'temporary' and 'draft' are now removed

Set the display color for this event.

Usage

Set event color to red:

event = Event.from_id("ev_abc123")
updated_event = event.set_color("red")
print(updated_event.color)
# 'red'

Clear event color:

updated_event = event.set_color(None)
print(updated_event.color)
# None

Set a single custom-field value on this event.

name must be the name of a Ready custom field for this event’s org and the Event entity type; value must satisfy the field’s declared type.

Set or overwrite multiple custom-field values on this event.

Each key must name a Ready custom field for this event’s org and the Event entity type; each value must satisfy the field’s declared type.

Set the description for this event.

Usage

Set event description:

event = Event.from_id("ev_abc123")
updated_event = event.set_description("Updated event description")
print(updated_event.description)
# 'Updated event description'

Clear event description:

updated_event = event.set_description(None)
print(updated_event.description)
# None

Set the name for this event.

Usage

Update event name:

event = Event.from_id("ev_abc123")
updated_event = event.set_name("Critical System Alert")
print(updated_event.name)
# 'Critical System Alert'

Convert this event to a dictionary representation.

Usage

Convert event to dictionary:

event = Event.from_id("ev_abc123")
event_dict = event.to_dict()
print(event_dict["name"])
print(event_dict["start_time"])

Get topic IDs associated with this event.

Usage

Get all associated topic IDs:

event = Event.from_id("ev_abc123")
topic_ids = event.topic_ids()
print(f"Associated with {len(topic_ids)} topics")

Get only directly associated topics:

strict_topic_ids = event.topic_ids(strict_associations=True)
for topic_id in strict_topic_ids:
    print(f"Directly associated topic: {topic_id}")

Update this event’s attributes.

Updates various properties of the event including name, time range, description, metadata, and display options. Only specified parameters are updated; others remain unchanged.

When provided, start_time and end_time should be integers representing nanoseconds since the UNIX epoch, or convertible to such integers by to_epoch_nanoseconds().

Usage

Update event name and description:

event = Event.from_id("ev_abc123")
updated_event = event.update(
    name="Critical System Alert", description="Updated description with more details"
)

Update event time range:

updated_event = event.update(start_time=1722870127699468923, end_time=1722870127799468923)

Update metadata and display options:

from roboto.updates import MetadataChangeset
from roboto.domain.events import EventDisplayOptionsChangeset
updated_event = event.update(
    metadata_changeset=MetadataChangeset(
        put_fields={"severity": "high"}, put_tags=["critical", "urgent"]
    ),
    display_options_changeset=EventDisplayOptionsChangeset(color="red"),
)

Checks whether any display options have been specified.

Applies this changeset to some existing display options.

Checks whether this changeset contains any changes.