Time Control

abses.core.time_driver.time_condition

time_condition(condition, when_run=True)

A decorator to run a method based on a time condition.

Parameters:

Name	Type	Description	Default
condition	dict	A dictionary containing conditions to check against the time attribute. The keys can be ['year', 'month', 'weekday', 'freqstr'].	required
when_run	bool	If True, the decorated method will run when the condition is met. If False, the decorated method will not run when the condition is met.	True

Example

class TestActor(Actor):
    @time_condition(condition={"month": 1, "day": 1}, when_run=True)
    def happy_new_year(self):
        print("Today is 1th, January, Happy new year!")


parameters = {"time": {"start": "1996-12-24", "days": 1}}


model = MainModel(parameters=parameters)
agent = model.agents.new(TestActor, 1, singleton=True)

for _ in range(10):
    print(f"Time now is {model.time}")
    model.time.go()
    agent.happy_new_year()

It should be called again in the next year beginning (i.e., 1998-01-01) if we run this model longer... It means, the function will be called when the condition is fully satisfied.

Source code in abses/core/time_driver.py

def time_condition(condition: dict, when_run: bool = True) -> Callable:
    """
    A decorator to run a method based on a time condition.

    Parameters:
        condition:
            A dictionary containing conditions to check against the `time` attribute.
            The keys can be ['year', 'month', 'weekday', 'freqstr'].
        when_run:
            If True, the decorated method will run when the condition is met.
            If False, the decorated method will not run when the condition is met.

    Example:
        ```
        class TestActor(Actor):
            @time_condition(condition={"month": 1, "day": 1}, when_run=True)
            def happy_new_year(self):
                print("Today is 1th, January, Happy new year!")


        parameters = {"time": {"start": "1996-12-24", "days": 1}}


        model = MainModel(parameters=parameters)
        agent = model.agents.new(TestActor, 1, singleton=True)

        for _ in range(10):
            print(f"Time now is {model.time}")
            model.time.go()
            agent.happy_new_year()
        ```
        It should be called again in the next year beginning (i.e., `1998-01-01`) if we run this model longer... It means, the function will be called when the condition is fully satisfied.
    """

    def decorator(func):
        @wraps(func)
        def wrapper(self, *args, **kwargs):
            if not hasattr(self, "time"):
                raise AttributeError(
                    "The object doesn't have a TimeDriver object as `time` attribute."
                )
            time = self.time
            if not isinstance(time, TimeDriver):
                raise TypeError(
                    "Decorated function is not belonged to an object with `TimeDriver`."
                )

            ok = all(
                getattr(time.dt, unit, None) == value
                for unit, value in condition.items()
            )

            if (ok and when_run) or (not ok and not when_run):
                return func(self, *args, **kwargs)

        return wrapper

    return decorator

abses.core.time_driver.TimeDriver

TimeDriver(model)

Bases: BaseModelElement, TimeDriverProtocol

TimeDriver provides the functionality to manage time.

A wrapper around datetime that adds simulation-specific functionality while providing access to all datetime attributes and methods.

Source code in abses/core/time_driver.py

def __init__(self, model: MainModelProtocol):
    super().__init__(model=model, name="time")
    self._history: Deque[DateTime] = deque()
    self._history_ticks: Deque[int] = deque()
    # End time can only be DateTime | int | None at runtime
    self._end_dt: DateTime | int | None = None
    self._parse_ticking_mode(set(self.params.keys()))
    self._parse_time_settings(self.params)
    self._dt = self.start_dt
    self._logging_setup()

fmt cached property

fmt

String format of datetime. If the datetime is a date object, return the date format. Otherwise, return the datetime format.

expected_ticks property

expected_ticks

Returns the expected ticks.

If the end_at is an integer or None, return the end_at. Otherwise, calculate the expected ticks.

should_end property

should_end

Should the model end or not. If the end_dt is a datetime object, return True if the current time is greater than or equal to the end_dt. If the end_dt is an integer, return True if the current tick is greater than or equal to the end_dt.

history property

history

Returns the history of the time driver. The history is a pandas Series object with the datetime as the index.

is_tick_mode property

is_tick_mode

Returns the tick mode.

duration property

duration

Returns the duration of the time driver.

start_dt property writable

start_dt

Returns the starting time for the model.

end_at property writable

end_at

The real-world time or the ticks when the model should be end.

If the end time is a datetime object, it will be converted to a DateTime object. If the end time is an integer, it will be interpreted as a number of ticks.

dt property writable

dt

Current simulation time.

If assigned a start/duration/end time, it will be a datetime object representing the current simulation time. Otherwise, it will be a datetime object representing the tick-updated time.

to

to(dt)

Specific the current time.

If the time is a string, it will be converted to a datetime object. If the time is an integer, it will be interpreted as a number of ticks.

Source code in abses/core/time_driver.py

def to(self, dt: DateTimeOrStr | int) -> None:
    """Specific the current time.

    If the time is a string, it will be converted to a datetime object.
    If the time is an integer, it will be interpreted as a number of ticks.
    """
    if isinstance(dt, str):
        dt = parse_datetime(dt)
    self.dt = dt

go

go(ticks=1)

Advance simulation time by a given number of ticks.

Source code in abses/core/time_driver.py

def go(self, ticks: int = 1) -> None:
    """Advance simulation time by a given number of ticks."""
    is_positive_int(ticks, raise_error=True)
    for _ in range(ticks):
        self._go_one_tick()