forked from frequenz-floss/frequenz-client-dispatch-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathclient.py
More file actions
97 lines (74 loc) · 2.53 KB
/
client.py
File metadata and controls
97 lines (74 loc) · 2.53 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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
# License: MIT
# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
"""Fake client for testing."""
from typing import Any
from .. import Client
from ..types import Dispatch
from ._service import ALL_KEY, NONE_KEY, FakeService
__all__ = ["FakeClient", "to_create_params", "ALL_KEY", "NONE_KEY"]
class FakeClient(Client):
"""Fake client for testing.
This client uses a fake service to simulate the dispatch api.
"""
def __init__(
self,
) -> None:
"""Initialize the mock client."""
super().__init__(server_url="mock", key=ALL_KEY, connect=False)
self._stuba: FakeService = FakeService()
@property
def stub(self) -> FakeService: # type: ignore
"""The fake service.
Returns:
FakeService: The fake service.
"""
return self._stuba
def dispatches(self, microgrid_id: int) -> list[Dispatch]:
"""List of dispatches.
Args:
microgrid_id: The microgrid id.
Returns:
list[Dispatch]: The list of dispatches
"""
return self._service.dispatches.get(microgrid_id, [])
def set_dispatches(self, microgrid_id: int, value: list[Dispatch]) -> None:
"""Set the list of dispatches.
Args:
microgrid_id: The microgrid id.
value: The list of dispatches to set.
"""
self._service.dispatches[microgrid_id] = value
if len(value) == 0:
return
# Max between last id and the max id in the list
# pylint: disable=protected-access
self._service._last_id = max(
self._service._last_id, max(dispatch.id for dispatch in value)
)
# pylint: enable=protected-access
@property
def _service(self) -> FakeService:
"""The fake service.
Returns:
FakeService: The fake service.
"""
return self._stuba
def to_create_params(microgrid_id: int, dispatch: Dispatch) -> dict[str, Any]:
"""Convert a dispatch to client.create parameters.
Args:
microgrid_id: The microgrid id.
dispatch: The dispatch to convert.
Returns:
dict[str, Any]: The create parameters.
"""
return {
"microgrid_id": microgrid_id,
"type": dispatch.type,
"start_time": dispatch.start_time,
"duration": dispatch.duration,
"target": dispatch.target,
"active": dispatch.active,
"dry_run": dispatch.dry_run,
"payload": dispatch.payload,
"recurrence": dispatch.recurrence,
}