roboto.domain.events.event#

Module Contents#

class roboto.domain.events.event.Event(record, roboto_client=None)#

Represents an event within the Roboto platform.

An event is a time-anchored annotation that relates Roboto entities (datasets, files, topics, and message paths) to specific time periods. Events enable temporal analysis, data correlation, and annotation of activities across different data sources.

Events serve as temporal markers that can:

  • Annotate specific time periods in your data

  • Associate multiple entities (datasets, files, topics, message paths) with time ranges

  • Enable time-based data retrieval and analysis

  • Support metadata and tagging for organization and search

  • Provide visual markers in timeline views and analysis tools

Events can represent instantaneous moments (point in time) or time ranges. They are particularly useful for marking significant occurrences like sensor anomalies, system events, behavioral patterns, or any other time-based phenomena in your data.

Events cannot be instantiated directly through the constructor. Use the class methods Event.create() to create new events or Event.from_id() to load existing events.

Parameters:
property color: str | None#

Display color for the event, if set.

Return type:

Optional[str]

classmethod create(name, start_time, end_time=None, associations=None, dataset_ids=None, file_ids=None, topic_ids=None, message_path_ids=None, description=None, metadata=None, tags=None, display_options=None, caller_org_id=None, roboto_client=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.

Parameters:

  • name (str) – Human-readable name for the event. Required.

  • start_time (roboto.time.Time) – Start timestamp of the event as nanoseconds since UNIX epoch, or any value convertible by to_epoch_nanoseconds().

  • end_time (Optional[roboto.time.Time]) – End timestamp of the event. If not provided, defaults to start_time for instantaneous events.

  • associations (Optional[collections.abc.Collection[roboto.association.Association]]) – Collection of Association objects linking the event to specific entities.

  • dataset_ids (Optional[collections.abc.Collection[str]]) – Dataset IDs to associate the event with.

  • file_ids (Optional[collections.abc.Collection[str]]) – File IDs to associate the event with.

  • topic_ids (Optional[collections.abc.Collection[str]]) – Topic IDs to associate the event with.

  • message_path_ids (Optional[collections.abc.Collection[str]]) – Message path IDs to associate the event with.

  • description (Optional[str]) – Optional human-readable description of the event.

  • metadata (Optional[dict[str, Any]]) – Key-value metadata for discovery and search.

  • tags (Optional[list[str]]) – Tags for categorizing and searching the event.

  • display_options (Optional[roboto.domain.events.operations.EventDisplayOptions]) – Visual display options such as color.

  • caller_org_id (Optional[str]) – Organization ID of the SDK caller. If not provided, uses the caller’s organization.

  • roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Returns:

Event instance with the provided attributes and associations.

Raises:
Return type:

Event

Examples

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"}
... )
property created: datetime.datetime#

Date and time when this event was created.

Return type:

datetime.datetime

property created_by: str#

User who created this event.

Return type:

str

dataset_ids(strict_associations=False)#

Get dataset IDs associated with this event.

Parameters:

strict_associations (bool) – If True, only return datasets with direct associations. If False (default), also return datasets inferred from file and topic associations.

Returns:

List of unique dataset IDs associated with this event.

Return type:

list[str]

Examples

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()#

Delete this event permanently.

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

Raises:
Return type:

None

Examples

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")
property description: str | None#

Optional human-readable description of the event.

Return type:

Optional[str]

property display_options: roboto.domain.events.operations.EventDisplayOptions | None#

Display options for the event, such as color.

Return type:

Optional[roboto.domain.events.operations.EventDisplayOptions]

property end_time: int#

End time of the event in nanoseconds since UNIX epoch.

Return type:

int

property event_id: str#

Unique identifier for this event.

Return type:

str

file_ids(strict_associations=False)#

Get file IDs associated with this event.

Parameters:

strict_associations (bool) – If True, only return files with direct associations. If False (default), also return files inferred from topic and message path associations.

Returns:

List of unique file IDs associated with this event.

Return type:

list[str]

Examples

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}")
classmethod from_id(event_id, roboto_client=None)#

Load an existing event by its ID.

Parameters:

  • event_id (str) – Unique identifier of the event to retrieve.

  • roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Returns:

Event instance for the specified ID.

Raises:
Return type:

Event

Examples

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}")
classmethod get_by_associations(associations, roboto_client=None)#

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.

Parameters:
Yields:

Event instances associated with any of the specified associations.

Return type:

collections.abc.Generator[Event, None, None]

Examples

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))
classmethod get_by_dataset(dataset_id, roboto_client=None, strict_associations=False)#

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.

Parameters:

  • dataset_id (str) – ID of the dataset to query events for.

  • roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

  • strict_associations (bool) – If True, only return events with direct dataset associations. If False (default), also return events associated with files or topics within the dataset.

Yields:

Event instances associated with the specified dataset.

Return type:

collections.abc.Generator[Event, None, None]

Examples

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}")
classmethod get_by_file(file_id, roboto_client=None)#

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

Parameters:

  • file_id (str) – ID of the file to query events for.

  • roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Yields:

Event instances directly associated with the specified file.

Return type:

collections.abc.Generator[Event, None, None]

Examples

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")
classmethod get_by_message_path(message_path_id, roboto_client=None)#

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

Parameters:

  • message_path_id (str) – ID of the message path to query events for.

  • roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Yields:

Event instances directly associated with the specified message path.

Return type:

collections.abc.Generator[Event, None, None]

Examples

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
... ]
classmethod get_by_topic(topic_id, roboto_client=None)#

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

Parameters:

  • topic_id (str) – ID of the topic to query events for.

  • roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Yields:

Event instances directly associated with the specified topic.

Return type:

collections.abc.Generator[Event, None, None]

Examples

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")
get_data(message_paths_include=None, message_paths_exclude=None, topic_name=None, topic_data_service=None, cache_dir=None, strict_associations=False)#

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"])
Parameters:

  • message_paths_include (Optional[collections.abc.Sequence[str]])

  • message_paths_exclude (Optional[collections.abc.Sequence[str]])

  • topic_name (Optional[str])

  • topic_data_service (Optional[roboto.domain.topics.TopicDataService])

  • cache_dir (Union[str, pathlib.Path, None])

  • strict_associations (bool)

Return type:

collections.abc.Generator[dict[str, Any], None, None]

get_data_as_df(message_paths_include=None, message_paths_exclude=None, topic_name=None, topic_data_service=None, cache_dir=None, strict_associations=False)#

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.

Parameters:

  • message_paths_include (Optional[collections.abc.Sequence[str]]) – Dot notation paths to include in the data.

  • message_paths_exclude (Optional[collections.abc.Sequence[str]]) – Dot notation paths to exclude from the data.

  • topic_name (Optional[str]) – Required when event is associated with a file.

  • topic_data_service (Optional[roboto.domain.topics.TopicDataService]) – Service for accessing topic data.

  • cache_dir (Union[str, pathlib.Path, None]) – Directory for caching downloaded data.

  • strict_associations (bool)

Returns:

DataFrame containing the event’s underlying topic data, indexed by log time.

Raises:

  • ImportError – If pandas is not installed (install with roboto[analytics]).

  • RobotoInvalidRequestException – Invalid parameters or event associations.

Return type:

pandas.DataFrame

Examples

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)}")
message_path_ids()#

Get message path IDs directly associated with this event.

Returns:

List of unique message path IDs directly associated with this event.

Return type:

list[str]

Examples

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")
property metadata: dict[str, Any]#

Key-value metadata associated with this event.

Return type:

dict[str, Any]

property modified: datetime.datetime#

Date and time when this event was last modified.

Return type:

datetime.datetime

property modified_by: str#

User who last modified this event.

Return type:

str

property name: str#

Human-readable name of the event.

Return type:

str

put_metadata(metadata)#

Add or update metadata fields for this event.

Parameters:

metadata (dict[str, Any]) – Dictionary of key-value pairs to add or update.

Returns:

Updated Event instance.

Return type:

Event

Examples

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'
put_tags(tags)#

Replace all tags for this event.

Parameters:

tags (list[str]) – List of tags to set for this event.

Returns:

Updated Event instance.

Return type:

Event

Examples

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']
property record: roboto.domain.events.record.EventRecord#

Underlying event record data.

Return type:

roboto.domain.events.record.EventRecord

refresh()#

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.

Returns:

This Event instance with refreshed data.

Return type:

Event

Examples

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(metadata)#

Remove metadata fields from this event.

Parameters:

metadata (roboto.updates.StrSequence) – Sequence of metadata field names to remove. Supports dot notation for nested fields.

Returns:

Updated Event instance.

Return type:

Event

Examples

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_tags(tags)#

Remove specific tags from this event.

Parameters:

tags (roboto.updates.StrSequence) – Sequence of tag names to remove from this event.

Returns:

Updated Event instance.

Return type:

Event

Examples

Remove specific tags:

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

Set the display color for this event.

Parameters:

color (Optional[str]) – CSS-compatible color value (e.g., “red”, “#ff0000”, “rgb(255,0,0)”). Use None to clear the color and use automatic coloring.

Returns:

Updated Event instance.

Return type:

Event

Examples

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_description(description)#

Set the description for this event.

Parameters:

description (Optional[str]) – New description for the event. Use None to clear the description.

Returns:

Updated Event instance.

Return type:

Event

Examples

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_name(name)#

Set the name for this event.

Parameters:

name (str) – New name for the event.

Returns:

Updated Event instance.

Return type:

Event

Examples

Update event name:

>>> event = Event.from_id("ev_abc123")
>>> updated_event = event.set_name("Critical System Alert")
>>> print(updated_event.name)
'Critical System Alert'
property start_time: int#

Start time of the event in nanoseconds since UNIX epoch.

Return type:

int

property tags: list[str]#

Tags associated with this event for categorization and search.

Return type:

list[str]

to_dict()#

Convert this event to a dictionary representation.

Returns:

Dictionary containing all event data in JSON-serializable format.

Return type:

dict[str, Any]

Examples

Convert event to dictionary:

>>> event = Event.from_id("ev_abc123")
>>> event_dict = event.to_dict()
>>> print(event_dict["name"])
>>> print(event_dict["start_time"])
topic_ids(strict_associations=False)#

Get topic IDs associated with this event.

Parameters:

strict_associations (bool) – If True, only return topics with direct associations. If False (default), also return topics inferred from message path associations.

Returns:

List of unique topic IDs associated with this event.

Return type:

list[str]

Examples

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(name=NotSet, start_time=NotSet, end_time=NotSet, description=NotSet, metadata_changeset=NotSet, display_options_changeset=NotSet)#

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().

Parameters:
Returns:

This Event instance with attributes updated accordingly.

Raises:
Return type:

Event

Examples

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")
... )
class roboto.domain.events.event.GetDataArgs#

Internal interface used to collect arguments passed to get_data and get_data_as_df.

cache_dir: str | pathlib.Path | None = None#
end_time: roboto.time.Time | None = None#
message_paths_exclude: collections.abc.Sequence[str] | None = None#
message_paths_include: collections.abc.Sequence[str] | None = None#
start_time: roboto.time.Time | None = None#
topic: roboto.domain.topics.Topic#
roboto.domain.events.event.logger#
On this page