Skip to content

recurrence

frequenz.client.dispatch.recurrence ¤

Types for recurrence rules.

Classes¤

frequenz.client.dispatch.recurrence.EndCriteria dataclass ¤

Controls when a recurring dispatch should end.

Source code in frequenz/client/dispatch/recurrence.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@dataclass(kw_only=True)
class EndCriteria:
    """Controls when a recurring dispatch should end."""

    count: int | None = None
    """The number of times this dispatch should recur."""
    until: datetime | None = None
    """The end time of this dispatch in UTC."""

    @classmethod
    def from_protobuf(cls, pb_criteria: PBRecurrenceRule.EndCriteria) -> "EndCriteria":
        """Convert a protobuf end criteria to an end criteria.

        Args:
            pb_criteria: The protobuf end criteria to convert.

        Returns:
            The converted end criteria.
        """
        instance = cls()

        match pb_criteria.WhichOneof("count_or_until"):
            case "count":
                instance.count = pb_criteria.count
            case "until":
                instance.until = to_datetime(pb_criteria.until)
        return instance

    def to_protobuf(self) -> PBRecurrenceRule.EndCriteria:
        """Convert an end criteria to a protobuf end criteria.

        Returns:
            The converted protobuf end criteria.
        """
        pb_criteria = PBRecurrenceRule.EndCriteria()

        if self.count is not None:
            pb_criteria.count = self.count
        elif self.until is not None:
            pb_criteria.until.CopyFrom(to_timestamp(self.until))

        return pb_criteria
Attributes¤
count class-attribute instance-attribute ¤
count: int | None = None

The number of times this dispatch should recur.

until class-attribute instance-attribute ¤
until: datetime | None = None

The end time of this dispatch in UTC.

Functions¤
from_protobuf classmethod ¤
from_protobuf(pb_criteria: EndCriteria) -> EndCriteria

Convert a protobuf end criteria to an end criteria.

PARAMETER DESCRIPTION
pb_criteria

The protobuf end criteria to convert.

TYPE: EndCriteria

RETURNS DESCRIPTION
EndCriteria

The converted end criteria.
Source code in frequenz/client/dispatch/recurrence.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@classmethod
def from_protobuf(cls, pb_criteria: PBRecurrenceRule.EndCriteria) -> "EndCriteria":
    """Convert a protobuf end criteria to an end criteria.

    Args:
        pb_criteria: The protobuf end criteria to convert.

    Returns:
        The converted end criteria.
    """
    instance = cls()

    match pb_criteria.WhichOneof("count_or_until"):
        case "count":
            instance.count = pb_criteria.count
        case "until":
            instance.until = to_datetime(pb_criteria.until)
    return instance
to_protobuf ¤
to_protobuf() -> EndCriteria

Convert an end criteria to a protobuf end criteria.

RETURNS DESCRIPTION
EndCriteria

The converted protobuf end criteria.
Source code in frequenz/client/dispatch/recurrence.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def to_protobuf(self) -> PBRecurrenceRule.EndCriteria:
    """Convert an end criteria to a protobuf end criteria.

    Returns:
        The converted protobuf end criteria.
    """
    pb_criteria = PBRecurrenceRule.EndCriteria()

    if self.count is not None:
        pb_criteria.count = self.count
    elif self.until is not None:
        pb_criteria.until.CopyFrom(to_timestamp(self.until))

    return pb_criteria

frequenz.client.dispatch.recurrence.Frequency ¤

Bases: IntEnum

Enum representing the frequency of the recurrence.

Source code in frequenz/client/dispatch/recurrence.py 
33
34
35
36
37
38
39
40
41
42
class Frequency(IntEnum):
    """Enum representing the frequency of the recurrence."""

    UNSPECIFIED = PBRecurrenceRule.FREQUENCY_UNSPECIFIED
    MINUTELY = PBRecurrenceRule.FREQUENCY_MINUTELY
    HOURLY = PBRecurrenceRule.FREQUENCY_HOURLY
    DAILY = PBRecurrenceRule.FREQUENCY_DAILY
    WEEKLY = PBRecurrenceRule.FREQUENCY_WEEKLY
    MONTHLY = PBRecurrenceRule.FREQUENCY_MONTHLY
    YEARLY = PBRecurrenceRule.FREQUENCY_YEARLY

frequenz.client.dispatch.recurrence.RecurrenceRule dataclass ¤

Ruleset governing when and how a dispatch should re-occur.

Attributes follow the iCalendar specification (RFC5545) for recurrence rules.

Source code in frequenz/client/dispatch/recurrence.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
@dataclass(kw_only=True)
class RecurrenceRule:
    """Ruleset governing when and how a dispatch should re-occur.

    Attributes follow the iCalendar specification (RFC5545) for recurrence rules.
    """

    frequency: Frequency = Frequency.UNSPECIFIED
    """The frequency specifier of this recurring dispatch."""

    interval: int = 0
    """How often this dispatch should recur, based on the frequency."""

    end_criteria: EndCriteria | None = None
    """When this dispatch should end.

    Can recur a fixed number of times or until a given timestamp."""

    byminutes: list[int] = field(default_factory=list)
    """On which minute(s) of the hour the event occurs."""

    byhours: list[int] = field(default_factory=list)
    """On which hour(s) of the day the event occurs."""

    byweekdays: list[Weekday] = field(default_factory=list)
    """On which day(s) of the week the event occurs."""

    bymonthdays: list[int] = field(default_factory=list)
    """On which day(s) of the month the event occurs."""

    bymonths: list[int] = field(default_factory=list)
    """On which month(s) of the year the event occurs."""

    @classmethod
    def from_protobuf(cls, pb_rule: PBRecurrenceRule) -> "RecurrenceRule":
        """Convert a protobuf recurrence rule to a recurrence rule.

        Args:
            pb_rule: The protobuf recurrence rule to convert.

        Returns:
            The converted recurrence rule.
        """
        return RecurrenceRule(
            frequency=Frequency(pb_rule.freq),
            interval=pb_rule.interval,
            end_criteria=(
                EndCriteria.from_protobuf(pb_rule.end_criteria)
                if pb_rule.HasField("end_criteria")
                else None
            ),
            byminutes=list(pb_rule.byminutes),
            byhours=list(pb_rule.byhours),
            byweekdays=[Weekday(day) for day in pb_rule.byweekdays],
            bymonthdays=list(pb_rule.bymonthdays),
            bymonths=list(pb_rule.bymonths),
        )

    def to_protobuf(self) -> PBRecurrenceRule:
        """Convert a recurrence rule to a protobuf recurrence rule.

        Returns:
            The converted protobuf recurrence rule.
        """
        pb_rule = PBRecurrenceRule()

        pb_rule.freq = self.frequency.value
        pb_rule.interval = self.interval
        if self.end_criteria is not None:
            pb_rule.end_criteria.CopyFrom(self.end_criteria.to_protobuf())
        pb_rule.byminutes.extend(self.byminutes)
        pb_rule.byhours.extend(self.byhours)
        pb_rule.byweekdays.extend([day.value for day in self.byweekdays])
        pb_rule.bymonthdays.extend(self.bymonthdays)
        pb_rule.bymonths.extend(self.bymonths)

        return pb_rule

    def _as_rrule(self, start_time: datetime) -> rrule.rrule:
        """Prepare the rrule object.

        Args:
            start_time: The start time of the dispatch.

        Returns:
            The rrule object.

        Raises:
            ValueError: If the interval is 0.
        """
        if self.interval == 0:
            raise ValueError("Interval must be greater than 0")

        count, until = (None, None)
        if end := self.end_criteria:
            count = end.count
            until = end.until

        rrule_obj = rrule.rrule(
            freq=_RRULE_FREQ_MAP[self.frequency],
            dtstart=start_time,
            count=count,
            until=until,
            byminute=self.byminutes or None,
            byhour=self.byhours or None,
            byweekday=[_RRULE_WEEKDAY_MAP[weekday] for weekday in self.byweekdays]
            or None,
            bymonthday=self.bymonthdays or None,
            bymonth=self.bymonths or None,
            interval=self.interval,
        )

        return rrule_obj
Attributes¤
byhours class-attribute instance-attribute ¤
byhours: list[int] = field(default_factory=list)

On which hour(s) of the day the event occurs.

byminutes class-attribute instance-attribute ¤
byminutes: list[int] = field(default_factory=list)

On which minute(s) of the hour the event occurs.

bymonthdays class-attribute instance-attribute ¤
bymonthdays: list[int] = field(default_factory=list)

On which day(s) of the month the event occurs.

bymonths class-attribute instance-attribute ¤
bymonths: list[int] = field(default_factory=list)

On which month(s) of the year the event occurs.

byweekdays class-attribute instance-attribute ¤
byweekdays: list[Weekday] = field(default_factory=list)

On which day(s) of the week the event occurs.

end_criteria class-attribute instance-attribute ¤
end_criteria: EndCriteria | None = None

When this dispatch should end.

Can recur a fixed number of times or until a given timestamp.

frequency class-attribute instance-attribute ¤
frequency: Frequency = UNSPECIFIED

The frequency specifier of this recurring dispatch.

interval class-attribute instance-attribute ¤
interval: int = 0

How often this dispatch should recur, based on the frequency.

Functions¤
from_protobuf classmethod ¤
from_protobuf(pb_rule: RecurrenceRule) -> RecurrenceRule

Convert a protobuf recurrence rule to a recurrence rule.

PARAMETER DESCRIPTION
pb_rule

The protobuf recurrence rule to convert.

TYPE: RecurrenceRule

RETURNS DESCRIPTION
RecurrenceRule

The converted recurrence rule.
Source code in frequenz/client/dispatch/recurrence.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
@classmethod
def from_protobuf(cls, pb_rule: PBRecurrenceRule) -> "RecurrenceRule":
    """Convert a protobuf recurrence rule to a recurrence rule.

    Args:
        pb_rule: The protobuf recurrence rule to convert.

    Returns:
        The converted recurrence rule.
    """
    return RecurrenceRule(
        frequency=Frequency(pb_rule.freq),
        interval=pb_rule.interval,
        end_criteria=(
            EndCriteria.from_protobuf(pb_rule.end_criteria)
            if pb_rule.HasField("end_criteria")
            else None
        ),
        byminutes=list(pb_rule.byminutes),
        byhours=list(pb_rule.byhours),
        byweekdays=[Weekday(day) for day in pb_rule.byweekdays],
        bymonthdays=list(pb_rule.bymonthdays),
        bymonths=list(pb_rule.bymonths),
    )
to_protobuf ¤
to_protobuf() -> RecurrenceRule

Convert a recurrence rule to a protobuf recurrence rule.

RETURNS DESCRIPTION
RecurrenceRule

The converted protobuf recurrence rule.
Source code in frequenz/client/dispatch/recurrence.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def to_protobuf(self) -> PBRecurrenceRule:
    """Convert a recurrence rule to a protobuf recurrence rule.

    Returns:
        The converted protobuf recurrence rule.
    """
    pb_rule = PBRecurrenceRule()

    pb_rule.freq = self.frequency.value
    pb_rule.interval = self.interval
    if self.end_criteria is not None:
        pb_rule.end_criteria.CopyFrom(self.end_criteria.to_protobuf())
    pb_rule.byminutes.extend(self.byminutes)
    pb_rule.byhours.extend(self.byhours)
    pb_rule.byweekdays.extend([day.value for day in self.byweekdays])
    pb_rule.bymonthdays.extend(self.bymonthdays)
    pb_rule.bymonths.extend(self.bymonths)

    return pb_rule

frequenz.client.dispatch.recurrence.Weekday ¤

Bases: IntEnum

Enum representing the day of the week.

Source code in frequenz/client/dispatch/recurrence.py 
20
21
22
23
24
25
26
27
28
29
30
class Weekday(IntEnum):
    """Enum representing the day of the week."""

    UNSPECIFIED = PBRecurrenceRule.WEEKDAY_UNSPECIFIED
    MONDAY = PBRecurrenceRule.WEEKDAY_MONDAY
    TUESDAY = PBRecurrenceRule.WEEKDAY_TUESDAY
    WEDNESDAY = PBRecurrenceRule.WEEKDAY_WEDNESDAY
    THURSDAY = PBRecurrenceRule.WEEKDAY_THURSDAY
    FRIDAY = PBRecurrenceRule.WEEKDAY_FRIDAY
    SATURDAY = PBRecurrenceRule.WEEKDAY_SATURDAY
    SUNDAY = PBRecurrenceRule.WEEKDAY_SUNDAY

Functions¤