Source code for noob.event
import sys
from enum import StrEnum
from typing import Any, Literal
from noob.types import Picklable, SerializableDatetime
if sys.version_info < (3, 12):
from typing_extensions import TypedDict
else:
from typing import TypedDict
[docs]
class Event(TypedDict):
"""
Container for a single value returned from a single :meth:`.Node.process` call
"""
id: int
"""Unique ID for each event"""
timestamp: SerializableDatetime
"""Timestamp of when the event was received by the :class:`.TubeRunner`"""
node_id: str
"""ID of node that emitted the event"""
signal: str
"""name of the signal that emitted the event"""
epoch: int
"""Epoch number the event was emitted in"""
value: Picklable[Any]
"""Value emitted by the processing node"""
[docs]
class MetaEventType(StrEnum):
"""
Types of meta events emitted by tubes, schedulers, stores, and runners.
"""
NodeReady = "NodeReady"
"""
A node was made ready in the toplogical sorting graph.
The value of the event is the node_id of the node that is ready.
"""
EpochEnded = "EpochEnded"
"""
An epoch has ended, the value of the event contains which epoch has ended
"""
[docs]
class MetaEvent(Event):
"""
All events generated by internal processes rather than nodes.
Used to coordinate the tube as well as allow code to hook into tube execution.
These are not stored in the :class:`.EventStore`,
but emitted by callbacks and consumed internally.
See :class:`.MetaEventType` for descriptions of the types of MetaEvents.
"""
# mypy doesn't allow narrowing types in typed dicts?
node_id: Literal["meta"] # type: ignore
signal: MetaEventType # type: ignore