Skip to content

Event

risansym.Event dataclass

Encapsulates the information exchanged between active processes in the simulation.

Parameters:

Name Type Description Default
time float

Simulation time at which the event is scheduled.

required
name str

Human-readable name identifying the event type.

required
source int

Node ID of the sender.

required
target int

Node ID of the receiver.

required
payload JsonPayload

JSON-serializable data attached to the event. The mapping is deep-copied on construction, so later mutations of the input do not alter the scheduled event. Treat event.payload as read-only.

dict()
Note on event ordering

Events scheduled for the exact same time are processed in FIFO order (the order in which they were inserted into the simulator).

Source code in core/src/risansym/event.py
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
@dataclass(order=True, slots=True, frozen=True)
class Event:
    """Encapsulates the information exchanged between active processes in the simulation.

    Args:
        time: Simulation time at which the event is scheduled.
        name: Human-readable name identifying the event type.
        source: Node ID of the sender.
        target: Node ID of the receiver.
        payload: JSON-serializable data attached to the event. The mapping is
            deep-copied on construction, so later mutations of the input do not
            alter the scheduled event. Treat ``event.payload`` as read-only.

    Note on event ordering:
        Events scheduled for the exact same time are processed in FIFO order
        (the order in which they were inserted into the simulator).
    """

    time: float
    # field(compare=False) prevents tie-breaking by name/IDs when times are equal
    name: str = field(compare=False)
    source: int = field(compare=False)
    target: int = field(compare=False)
    payload: JsonPayload = field(default_factory=dict, compare=False)

    def __post_init__(self) -> None:
        if not isinstance(self.time, (int, float)) or isinstance(self.time, bool):
            raise InvalidEventError("Event time must be a number.")
        if not math.isfinite(self.time) or self.time < 0:
            raise InvalidEventError(
                f"Invalid event time: {self.time}. Time must be finite and non-negative."
            )
        if not isinstance(self.source, int) or isinstance(self.source, bool) or self.source < 1:
            raise InvalidEventError("Event source must be a positive integer.")
        if not isinstance(self.target, int) or isinstance(self.target, bool) or self.target < 1:
            raise InvalidEventError("Event target must be a positive integer.")
        if not isinstance(self.name, str) or not self.name:
            raise InvalidEventError("Event name must be a non-empty string.")
        if not isinstance(self.payload, dict):
            raise InvalidEventError("Event payload must be a dictionary.")
        _validate_json_value(self.payload)
        object.__setattr__(self, "payload", copy.deepcopy(self.payload))

    def __repr__(self) -> str:
        return f"Event(t={self.time}, '{self.name}' {self.source}{self.target})"