Skip to content

generator

frequenz.client.dispatch.test.generator ¤

A sampler for dispatch messages.

Attributes¤

Classes¤

frequenz.client.dispatch.test.generator.DispatchGenerator ¤

Generates random dispatch messages.

Source code in frequenz/client/dispatch/test/generator.py 
 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
 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
108
109
110
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
class DispatchGenerator:
    """Generates random dispatch messages."""

    def __init__(self, seed: int = 0) -> None:
        """Initialize the sampler.

        Args:
            seed: seed to initialize the rng with
        """
        self._rng = random.Random(seed)
        self._last_id: int = 0

    def generate_recurrence_rule(self) -> RecurrenceRule:
        """Generate a random recurrence rule.

        Returns:
            a random recurrence rule
        """
        return RecurrenceRule(
            frequency=self._rng.choice(list(Frequency)[1:]),
            interval=self._rng.randint(1, 100),
            end_criteria=self._rng.choice(
                [
                    None,
                    self._rng.choice(
                        [
                            EndCriteria(count=self._rng.randint(1, 1000)),
                            EndCriteria(
                                until=datetime.fromtimestamp(
                                    self._rng.randint(0, 1000000),
                                    tz=timezone.utc,
                                )
                            ),
                        ]
                    ),
                ]
            ),
            byminutes=[
                self._rng.randint(0, 59) for _ in range(self._rng.randint(0, 10))
            ],
            byhours=[self._rng.randint(0, 23) for _ in range(self._rng.randint(0, 10))],
            byweekdays=[
                self._rng.choice(list(Weekday)[1:])
                for _ in range(self._rng.randint(0, 7))
            ],
            bymonthdays=[
                self._rng.randint(1, 31) for _ in range(self._rng.randint(0, 10))
            ],
            bymonths=[
                self._rng.randint(1, 12) for _ in range(self._rng.randint(0, 12))
            ],
        )

    def generate_target_category_and_type(self) -> TargetCategory:
        """Generate a random category and type.

        Returns:
            a random category and type
        """
        category = self._rng.choice(list(ComponentCategory)[1:])
        category_type: BatteryType | InverterType | EvChargerType | None = None

        match category:
            case ComponentCategory.BATTERY:
                category_type = self._rng.choice(list(BatteryType)[1:])
            case ComponentCategory.INVERTER:
                category_type = self._rng.choice(list(InverterType)[1:])
            case ComponentCategory.EV_CHARGER:
                category_type = self._rng.choice(list(EvChargerType)[1:])
            case _:
                category_type = None

        return TargetCategory(category_type or category)

    def generate_dispatch(self) -> Dispatch:
        """Generate a random dispatch instance.

        Returns:
            a random dispatch instance
        """
        self._last_id += 1
        create_time = datetime.fromtimestamp(
            self._rng.randint(0, 1000000), tz=timezone.utc
        )

        target_choices: list[TargetComponents] = [
            TargetIds(
                *[self._rng.randint(1, 100) for _ in range(self._rng.randint(1, 10))]
            ),
            TargetCategories(
                *[
                    # Not yet used
                    # self.generate_target_category_and_type()
                    self._rng.choice(list(ComponentCategory)[1:])
                    for _ in range(self._rng.randint(1, 10))
                ]
            ),
        ]

        return Dispatch(
            id=DispatchId(self._last_id),
            create_time=create_time,
            update_time=create_time + timedelta(seconds=self._rng.randint(0, 1000000)),
            type=str(self._rng.randint(0, 100_000)),
            start_time=rounded_start_time(
                datetime.now(tz=timezone.utc)
                + timedelta(seconds=self._rng.randint(0, 1000000))
            ),
            duration=self._rng.choice(
                [
                    None,
                    timedelta(seconds=self._rng.randint(0, 1000000)),
                ]
            ),
            target=self._rng.choice(target_choices),
            active=self._rng.choice([True, False]),
            dry_run=self._rng.choice([True, False]),
            payload={
                f"key_{i}": self._rng.choice(
                    [
                        self._rng.randint(0, 100),
                        self._rng.uniform(0, 100),
                        self._rng.choice([True, False]),
                        self._rng.choice(["a", "b", "c"]),
                    ]
                )
                for i in range(self._rng.randint(0, 10))
            },
            recurrence=self.generate_recurrence_rule(),
        )
Functions¤
__init__ ¤
__init__(seed: int = 0) -> None

Initialize the sampler.

PARAMETER DESCRIPTION
seed

seed to initialize the rng with

TYPE: int DEFAULT: 0

Source code in frequenz/client/dispatch/test/generator.py 
29
30
31
32
33
34
35
36
def __init__(self, seed: int = 0) -> None:
    """Initialize the sampler.

    Args:
        seed: seed to initialize the rng with
    """
    self._rng = random.Random(seed)
    self._last_id: int = 0
generate_dispatch ¤
generate_dispatch() -> Dispatch

Generate a random dispatch instance.

RETURNS DESCRIPTION
Dispatch

a random dispatch instance
Source code in frequenz/client/dispatch/test/generator.py 
100
101
102
103
104
105
106
107
108
109
110
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
def generate_dispatch(self) -> Dispatch:
    """Generate a random dispatch instance.

    Returns:
        a random dispatch instance
    """
    self._last_id += 1
    create_time = datetime.fromtimestamp(
        self._rng.randint(0, 1000000), tz=timezone.utc
    )

    target_choices: list[TargetComponents] = [
        TargetIds(
            *[self._rng.randint(1, 100) for _ in range(self._rng.randint(1, 10))]
        ),
        TargetCategories(
            *[
                # Not yet used
                # self.generate_target_category_and_type()
                self._rng.choice(list(ComponentCategory)[1:])
                for _ in range(self._rng.randint(1, 10))
            ]
        ),
    ]

    return Dispatch(
        id=DispatchId(self._last_id),
        create_time=create_time,
        update_time=create_time + timedelta(seconds=self._rng.randint(0, 1000000)),
        type=str(self._rng.randint(0, 100_000)),
        start_time=rounded_start_time(
            datetime.now(tz=timezone.utc)
            + timedelta(seconds=self._rng.randint(0, 1000000))
        ),
        duration=self._rng.choice(
            [
                None,
                timedelta(seconds=self._rng.randint(0, 1000000)),
            ]
        ),
        target=self._rng.choice(target_choices),
        active=self._rng.choice([True, False]),
        dry_run=self._rng.choice([True, False]),
        payload={
            f"key_{i}": self._rng.choice(
                [
                    self._rng.randint(0, 100),
                    self._rng.uniform(0, 100),
                    self._rng.choice([True, False]),
                    self._rng.choice(["a", "b", "c"]),
                ]
            )
            for i in range(self._rng.randint(0, 10))
        },
        recurrence=self.generate_recurrence_rule(),
    )
generate_recurrence_rule ¤
generate_recurrence_rule() -> RecurrenceRule

Generate a random recurrence rule.

RETURNS DESCRIPTION
RecurrenceRule

a random recurrence rule
Source code in frequenz/client/dispatch/test/generator.py 
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
71
72
73
74
75
76
77
def generate_recurrence_rule(self) -> RecurrenceRule:
    """Generate a random recurrence rule.

    Returns:
        a random recurrence rule
    """
    return RecurrenceRule(
        frequency=self._rng.choice(list(Frequency)[1:]),
        interval=self._rng.randint(1, 100),
        end_criteria=self._rng.choice(
            [
                None,
                self._rng.choice(
                    [
                        EndCriteria(count=self._rng.randint(1, 1000)),
                        EndCriteria(
                            until=datetime.fromtimestamp(
                                self._rng.randint(0, 1000000),
                                tz=timezone.utc,
                            )
                        ),
                    ]
                ),
            ]
        ),
        byminutes=[
            self._rng.randint(0, 59) for _ in range(self._rng.randint(0, 10))
        ],
        byhours=[self._rng.randint(0, 23) for _ in range(self._rng.randint(0, 10))],
        byweekdays=[
            self._rng.choice(list(Weekday)[1:])
            for _ in range(self._rng.randint(0, 7))
        ],
        bymonthdays=[
            self._rng.randint(1, 31) for _ in range(self._rng.randint(0, 10))
        ],
        bymonths=[
            self._rng.randint(1, 12) for _ in range(self._rng.randint(0, 12))
        ],
    )
generate_target_category_and_type ¤
generate_target_category_and_type() -> TargetCategory

Generate a random category and type.

RETURNS DESCRIPTION
TargetCategory

a random category and type
Source code in frequenz/client/dispatch/test/generator.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def generate_target_category_and_type(self) -> TargetCategory:
    """Generate a random category and type.

    Returns:
        a random category and type
    """
    category = self._rng.choice(list(ComponentCategory)[1:])
    category_type: BatteryType | InverterType | EvChargerType | None = None

    match category:
        case ComponentCategory.BATTERY:
            category_type = self._rng.choice(list(BatteryType)[1:])
        case ComponentCategory.INVERTER:
            category_type = self._rng.choice(list(InverterType)[1:])
        case ComponentCategory.EV_CHARGER:
            category_type = self._rng.choice(list(EvChargerType)[1:])
        case _:
            category_type = None

    return TargetCategory(category_type or category)

Functions¤