monitoring

oxytcmri.domain.ports.monitoring

Classes:

Name	Description
Event	Base class for all events emitted by the system.
ProgressEvent	Event indicating progress in a long-running process.
LogEvent	Event representing a log message.
Listener	Abstract interface for any object that listens to events.
EventDispatcher	Central hub for dispatching events to registered listeners.

Event

Bases: ABC

Base class for all events emitted by the system.

ProgressEvent(step, total)

Bases: Event

Event indicating progress in a long-running process.

Parameters:

Name	Type	Description	Default
step	int	The current step.	required
total	int	The total number of steps.	required

Attributes:

Name	Type	Description
step	int
total	int

Source code in oxytcmri/domain/ports/monitoring.py

def __init__(self, step: int, total: int) -> None:
    self.step: int = step
    self.total: int = total

step = step instance-attribute

total = total instance-attribute

LogEvent(message, level='INFO')

Bases: Event

Event representing a log message.

Parameters:

Name	Type	Description	Default
message	str	The message to log.	required
level	str	The log level ('INFO', 'DEBUG', 'WARNING', 'ERROR').	'INFO'

Attributes:

Name	Type	Description
message	str
level	str

Source code in oxytcmri/domain/ports/monitoring.py

def __init__(self, message: str, level: str = "INFO") -> None:
    self.message: str = message
    self.level: str = level.upper()

message = message instance-attribute

level = level.upper() instance-attribute

Listener

Bases: ABC

Abstract interface for any object that listens to events.

Methods:

Name	Description
on_event	React to an emitted event.

on_event(event) abstractmethod

React to an emitted event.

Parameters:

Name	Type	Description	Default
event	Event	The emitted event instance.	required

Source code in oxytcmri/domain/ports/monitoring.py

@abstractmethod
def on_event(self, event: Event) -> None:
    """
    React to an emitted event.

    Parameters
    ----------
    event : Event
        The emitted event instance.
    """

EventDispatcher()

Central hub for dispatching events to registered listeners.

Attributes:

Name	Type	Description
_listeners	list[Listener]	The registered listeners that will receive all events.

Methods:

Name	Description
register	Add a listener to receive future events.
dispatch	Dispatch an event to all registered listeners.

Source code in oxytcmri/domain/ports/monitoring.py

def __init__(self) -> None:
    self._listeners: list[Listener] = []

register(listener)

Add a listener to receive future events.

Parameters:

Name	Type	Description	Default
listener	Listener	A listener instance implementing on_event().	required

Source code in oxytcmri/domain/ports/monitoring.py

def register(self, listener: Listener) -> None:
    """
    Add a listener to receive future events.

    Parameters
    ----------
    listener : Listener
        A listener instance implementing on_event().
    """
    self._listeners.append(listener)

dispatch(event)

Dispatch an event to all registered listeners.

Parameters:

Name	Type	Description	Default
event	Event	An event instance to send to all listeners.	required

Source code in oxytcmri/domain/ports/monitoring.py

def dispatch(self, event: Event) -> None:
    """
    Dispatch an event to all registered listeners.

    Parameters
    ----------
    event : Event
        An event instance to send to all listeners.
    """
    for listener in self._listeners:
        listener.on_event(event)