"""Module containing convenience event decorators.
An event is a function that is triggered by other events or by the simulation itself.
"""
from __future__ import annotations
import inspect
import re
from datetime import timedelta
from typing import (
TYPE_CHECKING,
)
from eclypse.utils.constants import MAX_FLOAT
from eclypse.workflow.trigger import (
PeriodicTrigger,
ScheduledTrigger,
)
from .event import (
EclypseEvent,
)
from .role import EventRole
from .wrapper import EventWrapper
if TYPE_CHECKING:
from collections.abc import (
Callable,
)
from eclypse.utils.types import (
ActivatesOnType,
EventType,
TriggerCondition,
)
from eclypse.workflow.trigger.trigger import Trigger
def _event(
fn_or_class: Callable | None = None,
*,
name: str | None = None,
event_type: EventType | None = None,
activates_on: ActivatesOnType | None = None,
schedule_trigger: Trigger | None = None,
trigger_every_ms: float | None = None,
max_triggers: int | None = int(MAX_FLOAT),
triggers: Trigger | list[Trigger] | None = None,
trigger_condition: TriggerCondition | None = "any",
role: EventRole = EventRole.EVENT,
report: str | list[str] | None = None,
remote: bool = False,
verbose: bool = False,
) -> Callable:
"""Build an event wrapper from a callable.
Args:
fn_or_class (Callable | None, optional): The function or class to decorate
as an event. Defaults to None.
name (str | None, optional): The name of the event. If not provided,
the name will be derived from the function or class name. Defaults to None.
event_type (EventType | None, optional):
The type of the event. Defaults to None.
activates_on (ActivatesOnType | None, optional): The conditions that will
trigger the event. Defaults to None.
schedule_trigger: Trigger added by the public scheduling decorator.
trigger_every_ms (float | None, optional): The time in milliseconds between
each trigger of the event. Defaults to None.
max_triggers (int | None, optional): The maximum number of times the event
can be triggered. Defaults to no limit.
triggers (Trigger | list[Trigger] | None, optional): The triggers that will
trigger the event. If not provided, the event will not be
triggered by any triggers.
Defaults to None.
trigger_condition (str | None): The condition for the triggers to fire the
event. If "any", the event fires if any trigger is active. If "all",
the event fires only if all triggers are active. Defaults to "any".
role (EventRole, optional): The workflow role assigned to the event.
Defaults to EventRole.EVENT.
report (str | list[str] | None, optional): The type of report to generate
for the event. If not provided, the default report type will be used.
Defaults to DEFAULT_REPORT_TYPE.
remote (bool, optional): Whether the event is remote. Defaults to False.
verbose (bool, optional): Whether to print verbose output. Defaults to False.
Returns:
Callable: The decorated function.
"""
def decorator(decoratee: Callable) -> Callable:
if not callable(decoratee):
raise ValueError(
"The decorator must be applied to a function or a class"
+ "that implements the __call__ method.",
)
_name = _camel_to_snake(name if name else decoratee.__name__)
_triggers = (
triggers if isinstance(triggers, list) else [triggers] if triggers else []
)
if schedule_trigger:
_triggers.insert(0, schedule_trigger)
curr_opt = {
"name": _name,
"event_type": event_type,
"activates_on": activates_on,
"trigger_every_ms": trigger_every_ms,
"max_triggers": max_triggers,
"triggers": _triggers,
"trigger_condition": trigger_condition,
"role": role,
"report": report,
"remote": remote,
"verbose": verbose,
}
if inspect.isclass(decoratee):
class EventClassWrapper(decoratee): # type: ignore[misc, valid-type]
def __new__(cls, *args, **kwargs):
instance = (
decoratee(_name, *args, **kwargs)
if issubclass(decoratee, EclypseEvent)
else decoratee(*args, **kwargs)
)
event_obj = EventWrapper(instance, **curr_opt)
return event_obj
return EventClassWrapper
return EventWrapper(decoratee, **curr_opt) # type: ignore[arg-type]
if fn_or_class:
return decorator(fn_or_class)
return decorator
[docs]
def every(
fn_or_class: Callable | None = None,
*,
ms: float,
name: str | None = None,
event_type: EventType | None = None,
activates_on: ActivatesOnType | None = None,
max_triggers: int | None = int(MAX_FLOAT),
triggers: Trigger | list[Trigger] | None = None,
trigger_condition: TriggerCondition | None = "any",
role: EventRole = EventRole.EVENT,
report: str | list[str] | None = None,
remote: bool = False,
verbose: bool = False,
) -> Callable:
"""Define an event that fires periodically.
Args:
fn_or_class: The function or class to decorate.
ms: The period between triggers in milliseconds.
name: Optional event name. Defaults to the decorated object name.
event_type: Optional report event type.
activates_on: Cascade activation rules.
max_triggers: Maximum number of firings.
triggers: Additional triggers to combine with the periodic trigger.
trigger_condition: Whether any or all triggers must fire.
role: Workflow role assigned to the event.
report: Report formats generated by the event.
remote: Whether the event runs remotely.
verbose: Whether verbose event logging is enabled.
Returns:
The decorated event wrapper.
"""
return _event(
fn_or_class,
name=name,
event_type=event_type,
activates_on=activates_on,
schedule_trigger=PeriodicTrigger(ms),
max_triggers=max_triggers,
triggers=triggers,
trigger_condition=trigger_condition,
role=role,
report=report,
remote=remote,
verbose=verbose,
)
[docs]
def after(
fn_or_class: Callable | None = None,
*,
sim_seconds: float,
name: str | None = None,
event_type: EventType | None = None,
activates_on: ActivatesOnType | None = None,
max_triggers: int | None = 1,
triggers: Trigger | list[Trigger] | None = None,
trigger_condition: TriggerCondition | None = "any",
role: EventRole = EventRole.EVENT,
report: str | list[str] | None = None,
remote: bool = False,
verbose: bool = False,
) -> Callable:
"""Define an event that fires after a simulation-time delay.
Args:
fn_or_class: The function or class to decorate.
sim_seconds: Delay in simulation seconds before the event can fire.
name: Optional event name. Defaults to the decorated object name.
event_type: Optional report event type.
activates_on: Cascade activation rules.
max_triggers: Maximum number of firings. Defaults to once.
triggers: Additional triggers to combine with the scheduled trigger.
trigger_condition: Whether any or all triggers must fire.
role: Workflow role assigned to the event.
report: Report formats generated by the event.
remote: Whether the event runs remotely.
verbose: Whether verbose event logging is enabled.
Returns:
The decorated event wrapper.
"""
return _event(
fn_or_class,
name=name,
event_type=event_type,
activates_on=activates_on,
schedule_trigger=ScheduledTrigger(timedelta(seconds=sim_seconds)),
max_triggers=max_triggers,
triggers=triggers,
trigger_condition=trigger_condition,
role=role,
report=report,
remote=remote,
verbose=verbose,
)
[docs]
def once_at(
fn_or_class: Callable | None = None,
*,
sim_seconds: float,
name: str | None = None,
event_type: EventType | None = None,
activates_on: ActivatesOnType | None = None,
triggers: Trigger | list[Trigger] | None = None,
trigger_condition: TriggerCondition | None = "any",
role: EventRole = EventRole.EVENT,
report: str | list[str] | None = None,
remote: bool = False,
verbose: bool = False,
) -> Callable:
"""Define an event that fires once at a simulation-time offset.
Args:
fn_or_class: The function or class to decorate.
sim_seconds: Simulation-time offset in seconds.
name: Optional event name. Defaults to the decorated object name.
event_type: Optional report event type.
activates_on: Cascade activation rules.
triggers: Additional triggers to combine with the scheduled trigger.
trigger_condition: Whether any or all triggers must fire.
role: Workflow role assigned to the event.
report: Report formats generated by the event.
remote: Whether the event runs remotely.
verbose: Whether verbose event logging is enabled.
Returns:
The decorated event wrapper.
"""
return after(
fn_or_class,
sim_seconds=sim_seconds,
name=name,
event_type=event_type,
activates_on=activates_on,
max_triggers=1,
triggers=triggers,
trigger_condition=trigger_condition,
role=role,
report=report,
remote=remote,
verbose=verbose,
)
def _camel_to_snake(name: str) -> str:
"""Convert a CamelCase string to a snake_case string.
.. code-block:: python
name = "MyCamelCaseSentence"
print(_camel_to_snake(name)) # my_camel_case_sentence
Args:
name (str): The CamelCase string to convert.
Returns:
str: The snake_case string.
"""
return re.sub(r"(?<!^)(?=[A-Z])", "_", name).lower()
