Skip to main content

impulse_reporting.events.sequence_of_events

SequenceOfEvents​

class SequenceOfEvents(Event)

Class representing a sequence-of-events event in a report.

Visual timeline (two intervals where the second starts before the first ends):

time --->
event_1: | ------------------- |
event_2:             | ------------- |
sequence:| ------------------------- |

Rule:

  • If the next interval starts before the current one ends, the resulting sequence interval starts at the first event's start and ends at the next event's end.

__init__​

def __init__(name: str,
             expressions: list[TimeSeriesExpression],
             desc: str | None = None,
             required_channels: list[str] | None = None,
             max_overlap: float | None = None,
             attributes: Mapping[str, str] = None)

Initialize a SequenceOfEvents object.

Arguments:

  • name (str): Name of the event.
  • expressions (list of TimeSeriesExpression): Ordered list of expressions. Each expression must yield Intervals.
  • desc (str): Description of the event.
  • required_channels (list of str): List of required channels for the event.
  • max_overlap (float): Maximum allowed overlap time between consecutive events. If not None, intervals whose overlap exceeds this value are skipped. Expressed in the same time units as the underlying data (e.g. milliseconds-since-epoch timestamps), not seconds or any other derived unit.

get_id​

def get_id() -> int

Returns a unique identifier for the event.

Returns:

int: Unique positive 32-bit integer identifier for the event.

get_expression​

def get_expression() -> TimeSeriesExpression | None

Get the time series expression associated with the event.

Returns:

TimeSeriesExpression or None: The time series expression for the event.

get_event_type_str​

def get_event_type_str() -> str

Get the event type string for SequenceOfEvents.

Returns:

str: Event type string.

determine_definition_hash​

def determine_definition_hash() -> int

Calculate definition hash for sequence-of-events event.

Only includes the expression (computation logic), which is the only attribute that affects the event results.

Excludes: name, description, required_channels, report_id

Returns:

int: Hash value representing the computation definition.

as_dict​

def as_dict() -> dict

Get a dictionary representation of the event.

Returns:

dict: Dictionary containing event metadata.

as_spark_row​

def as_spark_row() -> Row

Get a Spark Row representation of the event.

Returns:

Row: Spark Row containing event metadata.

determine_events​

def determine_events(cls,
                     spark: SparkSession,
                     events: list[SequenceOfEvents],
                     *,
                     solved_df: "DataFrame" = None,
                     query: QueryBuilder = None,
                     solver: QuerySolver = None,
                     pre_filtered_containers_df=None)

Extract event fact table for the given list of SequenceOfEvents objects.

Arguments:

  • spark (SparkSession): Spark session for data processing.
  • events (list of SequenceOfEvents): List of SequenceOfEvents objects to process.
  • solved_df (DataFrame): Pre-solved wide DataFrame from centralized batch solve. Required.
  • query (QueryBuilder): Query builder (unused, kept for interface compatibility).
  • solver (QuerySolver): Query solver (unused, kept for interface compatibility).
  • pre_filtered_containers_df (DataFrame): Pre-filtered containers for incremental processing.

Returns:

DataFrame: Spark DataFrame containing event instance facts.

determine_metadata_df​

def determine_metadata_df(cls, spark: SparkSession,
                          events: list[SequenceOfEvents])

Create a Spark DataFrame containing event metadata.

Arguments:

  • spark (SparkSession): Spark session for data processing.
  • events (list of SequenceOfEvents): List of SequenceOfEvents objects.

Returns:

DataFrame: Spark DataFrame containing event metadata.