Skip to content

Time Utilities

hades.time

Time steps in hades can represent anything. This package contains some helper functions for common use cases particularly with time steps as days

QuarterStartScheduler

Bases: Process

adds quarter start events to be used for things occurring with a quarterly cadence. Depends on the YearStarted event being broadcast to it

Source code in hades/time/process.py 
73
74
75
76
77
78
79
80
81
82
83
class QuarterStartScheduler(Process):
    """adds quarter start events to be used for things occurring with a quarterly cadence. Depends on the YearStarted event
    being broadcast to it"""

    async def notify(self, event: Event) -> NotificationResponse:
        match event:
            case YearStarted(t=t):
                for i in range(4):
                    self.add_event(QuarterStarted(t=datetime_to_step(date(step_to_date(t).year, (i * 3) + 1, 1))))
                return NotificationResponse.ACK
        return NotificationResponse.NO_ACK

QuarterStarted

Bases: Event

the quarter started on this day

Source code in hades/time/event.py 
37
38
39
40
41
42
43
44
45
46
class QuarterStarted(Event):
    """the quarter started on this day"""

    @property
    def year(self) -> int:
        return step_to_date(self.t).year

    @property
    def quarter_number(self) -> int:
        return quarter_from_datetime(step_to_date(self.t))

YearStartScheduler

Bases: Process

adds year start events to be used by other processes for scheduling things which come with an annual cadence, has no dependencies in terms of other events apart from the built-in SimulationStarted

Source code in hades/time/process.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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
class YearStartScheduler(Process):
    """adds year start events to be used by other processes for scheduling things which come with an annual cadence,
    has no dependencies in terms of other events apart from the built-in SimulationStarted"""

    def __init__(self, start_year: int, look_ahead_years: int = 100) -> None:
        self._look_ahead_years = look_ahead_years
        self._start_year = start_year
        self._latest_year_added: int | None = None
        super().__init__()

    async def notify(self, event: Event) -> NotificationResponse:
        match event:
            case SimulationStarted(t=t) as e:
                _logger.debug(
                    "adding look ahead YearStarted events between %d and %d due to %s",
                    self._start_year,
                    self._look_ahead_years + self._start_year,
                    repr(e),
                )
                for year in range(self._start_year, self._look_ahead_years + self._start_year):
                    self.add_event(YearStarted(t=datetime_to_step(date(year, 1, 1))))
                self._latest_year_added = self._look_ahead_years + self._start_year - 1
                return NotificationResponse.ACK
            case YearStarted():
                return NotificationResponse.ACK_BUT_IGNORED
            case QuarterStarted():
                return NotificationResponse.ACK_BUT_IGNORED
            case Event(t=t) as e:
                # ensure that we keep the look ahead window maintained given other events
                if (
                    self._latest_year_added is not None
                    and (current_year := step_to_date(t).year) > self._latest_year_added - self._look_ahead_years + 1
                ):
                    _logger.debug(
                        "adding look ahead YearStarted events between %d and %d due to %s",
                        self._latest_year_added + 1,
                        current_year + self._look_ahead_years,
                        repr(e),
                    )
                    for year in range(self._latest_year_added + 1, current_year + self._look_ahead_years):
                        self.add_event(YearStarted(t=datetime_to_step(date(year, 1, 1))))
                        self._latest_year_added = year

                    return NotificationResponse.ACK
                return NotificationResponse.ACK_BUT_IGNORED
        return NotificationResponse.NO_ACK

YearStarted

Bases: Event

the year started on this day (e.g. 1/1)

Source code in hades/time/event.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
class YearStarted(Event):
    """the year started on this day (e.g. 1/1)"""

    @property
    def year(self) -> int:
        return step_to_date(self.t).year

    @property
    def is_leap(self) -> bool:
        return calendar.isleap(self.year)

    @property
    def number_of_days(self) -> int:
        return 365 + int(self.is_leap)

datetime_to_step(dt, epoch=EPOCH)

datetime or date as days since epoch

Source code in hades/time/day_steps.py 
22
23
24
25
26
def datetime_to_step(dt: Union[datetime, date], epoch: datetime = EPOCH) -> int:
    """datetime or date as days since epoch"""
    if not isinstance(dt, datetime):
        return datetime_to_step(datetime.fromisoformat(dt.isoformat()))
    return int((dt - epoch).total_seconds() // timedelta(days=1).total_seconds())

step_to_date(step)

days since epoch as date

Source code in hades/time/day_steps.py 
34
35
36
def step_to_date(step: int) -> date:
    """days since epoch as date"""
    return step_to_datetime(step).date()

step_to_datetime(step, epoch=EPOCH)

days since epoch as datetime

Source code in hades/time/day_steps.py 
29
30
31
def step_to_datetime(step: int, epoch: datetime = EPOCH) -> datetime:
    """days since epoch as datetime"""
    return epoch + timedelta(days=step)

hades.time.event

QuarterStarted

Bases: Event

the quarter started on this day

Source code in hades/time/event.py 
37
38
39
40
41
42
43
44
45
46
class QuarterStarted(Event):
    """the quarter started on this day"""

    @property
    def year(self) -> int:
        return step_to_date(self.t).year

    @property
    def quarter_number(self) -> int:
        return quarter_from_datetime(step_to_date(self.t))

YearStarted

Bases: Event

the year started on this day (e.g. 1/1)

Source code in hades/time/event.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
class YearStarted(Event):
    """the year started on this day (e.g. 1/1)"""

    @property
    def year(self) -> int:
        return step_to_date(self.t).year

    @property
    def is_leap(self) -> bool:
        return calendar.isleap(self.year)

    @property
    def number_of_days(self) -> int:
        return 365 + int(self.is_leap)

hades.time.day_steps

datetime_to_step(dt, epoch=EPOCH)

datetime or date as days since epoch

Source code in hades/time/day_steps.py 
22
23
24
25
26
def datetime_to_step(dt: Union[datetime, date], epoch: datetime = EPOCH) -> int:
    """datetime or date as days since epoch"""
    if not isinstance(dt, datetime):
        return datetime_to_step(datetime.fromisoformat(dt.isoformat()))
    return int((dt - epoch).total_seconds() // timedelta(days=1).total_seconds())

step_to_date(step)

days since epoch as date

Source code in hades/time/day_steps.py 
34
35
36
def step_to_date(step: int) -> date:
    """days since epoch as date"""
    return step_to_datetime(step).date()

step_to_datetime(step, epoch=EPOCH)

days since epoch as datetime

Source code in hades/time/day_steps.py 
29
30
31
def step_to_datetime(step: int, epoch: datetime = EPOCH) -> datetime:
    """days since epoch as datetime"""
    return epoch + timedelta(days=step)

hades.time.process

QuarterStartScheduler

Bases: Process

adds quarter start events to be used for things occurring with a quarterly cadence. Depends on the YearStarted event being broadcast to it

Source code in hades/time/process.py 
73
74
75
76
77
78
79
80
81
82
83
class QuarterStartScheduler(Process):
    """adds quarter start events to be used for things occurring with a quarterly cadence. Depends on the YearStarted event
    being broadcast to it"""

    async def notify(self, event: Event) -> NotificationResponse:
        match event:
            case YearStarted(t=t):
                for i in range(4):
                    self.add_event(QuarterStarted(t=datetime_to_step(date(step_to_date(t).year, (i * 3) + 1, 1))))
                return NotificationResponse.ACK
        return NotificationResponse.NO_ACK

YearStartScheduler

Bases: Process

adds year start events to be used by other processes for scheduling things which come with an annual cadence, has no dependencies in terms of other events apart from the built-in SimulationStarted

Source code in hades/time/process.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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
class YearStartScheduler(Process):
    """adds year start events to be used by other processes for scheduling things which come with an annual cadence,
    has no dependencies in terms of other events apart from the built-in SimulationStarted"""

    def __init__(self, start_year: int, look_ahead_years: int = 100) -> None:
        self._look_ahead_years = look_ahead_years
        self._start_year = start_year
        self._latest_year_added: int | None = None
        super().__init__()

    async def notify(self, event: Event) -> NotificationResponse:
        match event:
            case SimulationStarted(t=t) as e:
                _logger.debug(
                    "adding look ahead YearStarted events between %d and %d due to %s",
                    self._start_year,
                    self._look_ahead_years + self._start_year,
                    repr(e),
                )
                for year in range(self._start_year, self._look_ahead_years + self._start_year):
                    self.add_event(YearStarted(t=datetime_to_step(date(year, 1, 1))))
                self._latest_year_added = self._look_ahead_years + self._start_year - 1
                return NotificationResponse.ACK
            case YearStarted():
                return NotificationResponse.ACK_BUT_IGNORED
            case QuarterStarted():
                return NotificationResponse.ACK_BUT_IGNORED
            case Event(t=t) as e:
                # ensure that we keep the look ahead window maintained given other events
                if (
                    self._latest_year_added is not None
                    and (current_year := step_to_date(t).year) > self._latest_year_added - self._look_ahead_years + 1
                ):
                    _logger.debug(
                        "adding look ahead YearStarted events between %d and %d due to %s",
                        self._latest_year_added + 1,
                        current_year + self._look_ahead_years,
                        repr(e),
                    )
                    for year in range(self._latest_year_added + 1, current_year + self._look_ahead_years):
                        self.add_event(YearStarted(t=datetime_to_step(date(year, 1, 1))))
                        self._latest_year_added = year

                    return NotificationResponse.ACK
                return NotificationResponse.ACK_BUT_IGNORED
        return NotificationResponse.NO_ACK

hades.time.logging