Connection API

Connection class.

Represents a connection from an event to a slot.

class routilux.connection.Connection(source_event=None, target_slot=None, param_mapping=None)

Bases: Serializable

Connection object representing a link from an event to a slot.

A Connection establishes a unidirectional data flow path from a source Event to a target Slot. When the source event is emitted, the connection automatically transmits the data to the target slot, optionally applying parameter mapping to transform parameter names during transmission.

Key Concepts:

• One-to-One Relationship: Each Connection links exactly one Event to one Slot

• Parameter Mapping: Can rename parameters when transmitting data

• Automatic Transmission: Data flows automatically when event is emitted

• Bidirectional Link: Connection also establishes bidirectional link between Event and Slot (Event.connected_slots and Slot.connected_events)

Parameter Mapping:

Parameter mapping allows you to rename parameters when data flows from event to slot. This is useful when: - Event and slot use different parameter names for the same data - You want to standardize parameter names across different routines - You need to map multiple events to the same slot with different names

Format: {event_param_name: slot_param_name} - Mapped parameters: Renamed according to mapping - Unmapped parameters: Passed with original names (if not conflicting) - If a parameter is in mapping but not in event data: Not included - If a parameter is in mapping values but not in event data: Not included

Data Flow:

1. Event emits data via emit(**kwargs)

2. Connection receives data dictionary

3. Connection applies parameter mapping (if defined)

4. Connection calls slot.receive(mapped_data)

5. Slot merges data and calls handler

Examples

Basic connection (no parameter mapping):

>>> connection = Connection(source_event, target_slot)
>>> # Event emits: {"data": "value", "count": 5}
>>> # Slot receives: {"data": "value", "count": 5}

Connection with parameter mapping:

>>> connection = Connection(
...     source_event, target_slot,
...     param_mapping={"source_data": "target_input", "count": "total"}
... )
>>> # Event emits: {"source_data": "x", "count": 5, "extra": "ignored"}
>>> # Slot receives: {"target_input": "x", "total": 5, "extra": "ignored"}

__init__(source_event=None, target_slot=None, param_mapping=None)

Initialize a Connection between an event and a slot.

This constructor creates a connection and automatically establishes the bidirectional link between the event and slot. The connection is ready to transmit data immediately after creation.

Parameters:

• source_event (Event | None) – Source Event object that will emit data. Must be an Event instance created via Routine.define_event(). If None, connection is created but not active until both are set.

• target_slot (Slot | None) – Target Slot object that will receive data. Must be a Slot instance created via Routine.define_slot(). If None, connection is created but not active until both are set.

• param_mapping (dict[str, str] | None) – Optional dictionary mapping event parameter names to slot parameter names. Format: {event_param: slot_param} If None or empty, parameters are passed unchanged. Example: {“event_data”: “slot_input”, “event_count”: “slot_total”}

Side Effects:

• Automatically calls source_event.connect(target_slot) if both are provided

• Establishes bidirectional link between event and slot

• Connection is immediately active and ready to transmit data

Examples

Basic connection:

>>> event = routine1.define_event("output")
>>> slot = routine2.define_slot("input")
>>> connection = Connection(event, slot)
>>> # Connection is active, data will flow when event emits

Connection with parameter mapping:

>>> connection = Connection(
...     event, slot,
...     param_mapping={"result": "data", "status": "state"}
... )
>>> # When event emits {"result": "x", "status": "ok"},
>>> # slot receives {"data": "x", "state": "ok"}

serialize()

Serialize the Connection.

Returns:

Serialized dictionary containing connection data.

Return type:

dict[str, Any]

deserialize(data)

Deserialize the Connection.

Parameters:

data (dict[str, Any]) – Serialized data dictionary.

__repr__()

Return string representation of the Connection.

activate(data)

Activate the connection and transmit data to the target slot.

This method is called automatically when the source event is emitted. It applies parameter mapping (if defined) and transmits the data to the target slot. You typically don’t call this directly.

Processing Steps:

1. Apply parameter mapping to transform parameter names

2. Call target_slot.receive() with the mapped data

3. Slot merges data and calls its handler

Parameters:

data (dict[str, Any]) – Data dictionary from the source event. This is the kwargs passed to Event.emit(**kwargs). The dictionary will be transformed according to param_mapping before being sent to the slot. Example: {“result”: “success”, “count”: 42}

Examples

Manual activation (for testing):

>>> connection = Connection(event, slot, {"old": "new"})
>>> connection.activate({"old": "value", "other": "data"})
>>> # Slot receives {"new": "value", "other": "data"}

Automatic activation (normal usage):

>>> # When event emits, activate() is called automatically
>>> event.emit(flow=my_flow, result="data", count=5)
>>> # Connection.activate() is called internally

disconnect()

Disconnect the connection.