Skip to content

generator

frequenz.client.dispatch.test.generator ¤

A sampler for dispatch messages.

Classes¤

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

Generates random dispatch messages.

Source code in frequenz/client/dispatch/test/generator.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 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
 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
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_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
        )

        return Dispatch(
            id=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(  # type: ignore
                [
                    [
                        self._rng.choice(list(ComponentCategory)[1:])
                        for _ in range(self._rng.randint(1, 10))
                    ],
                    [
                        self._rng.randint(1, 100)
                        for _ in range(self._rng.randint(1, 10))
                    ],
                ]
            ),
            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 
19
20
21
22
23
24
25
26
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 
 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
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
    )

    return Dispatch(
        id=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(  # type: ignore
            [
                [
                    self._rng.choice(list(ComponentCategory)[1:])
                    for _ in range(self._rng.randint(1, 10))
                ],
                [
                    self._rng.randint(1, 100)
                    for _ in range(self._rng.randint(1, 10))
                ],
            ]
        ),
        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 
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
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))
        ],
    )

Functions¤