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
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
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
|
"""Support for communicating with myStrom plugs/switches."""
from typing import Optional, Union
import aiohttp
from yarl import URL
from . import _request as request
from .device_types import DEVICE_MAPPING_LITERAL, DEVICE_MAPPING_NUMERIC
class MyStromSwitch:
"""A class for a myStrom switch/plug."""
def __init__(
self,
host: str,
session: aiohttp.client.ClientSession = None,
token: Optional[str] = None,
) -> None:
"""Initialize the switch."""
self._close_session = False
self._host = host
self._token = token
self._session = session
self._consumption = 0
self._consumedWs = 0
self._boot_id = None
self._energy_since_boot = None
self._time_since_boot = None
self._state = None
self._temperature = None
self._firmware = None
self._mac = None
self._device_type: Optional[Union[str, int]] = None
self.uri = URL.build(scheme="http", host=self._host)
async def turn_on(self) -> None:
"""Turn the relay on."""
parameters = {"state": "1"}
url = URL(self.uri).join(URL("relay"))
await request(self, uri=url, params=parameters, token=self._token)
await self.get_state()
async def turn_off(self) -> None:
"""Turn the relay off."""
parameters = {"state": "0"}
url = URL(self.uri).join(URL("relay"))
await request(self, uri=url, params=parameters, token=self._token)
await self.get_state()
async def toggle(self) -> None:
"""Toggle the relay."""
url = URL(self.uri).join(URL("toggle"))
await request(self, uri=url, token=self._token)
await self.get_state()
async def get_state(self) -> None:
"""Get the details from the switch/plug."""
url = URL(self.uri).join(URL("report"))
response = await request(self, uri=url, token=self._token)
try:
self._consumption = response["power"]
except KeyError:
self._consumption = None
try:
self._consumedWs = response["Ws"]
except KeyError:
self._consumedWs = None
try:
self._boot_id = response["boot_id"]
except KeyError:
self._boot_id = None
try:
self._energy_since_boot = response["energy_since_boot"]
except KeyError:
self._energy_since_boot = None
try:
self._time_since_boot = response["time_since_boot"]
except KeyError:
self._time_since_boot = None
self._state = response["relay"]
try:
self._temperature = response["temperature"]
except KeyError:
self._temperature = None
# Try the new API (Devices with newer firmware)
url = URL(self.uri).join(URL("api/v1/info"))
response = await request(self, uri=url, token=self._token)
if not isinstance(response, dict):
# Fall back to the old API version if the device runs with old firmware
url = URL(self.uri).join(URL("info.json"))
response = await request(self, uri=url, token=self._token)
# Tolerate missing keys on legacy firmware (e.g., v1 devices)
self._firmware = response.get("version")
self._mac = response.get("mac")
self._device_type = response.get("type")
@property
def device_type(self) -> Optional[str]:
"""Return the device type as string (e.g. "Switch CH v1" or "Button+")."""
if isinstance(self._device_type, int):
return DEVICE_MAPPING_NUMERIC.get(self._device_type)
elif isinstance(self._device_type, str):
return DEVICE_MAPPING_LITERAL.get(self._device_type)
return None
@property
def relay(self) -> bool:
"""Return the relay state."""
return bool(self._state)
@property
def consumption(self) -> Optional[float]:
"""Return the current power consumption in mWh."""
if self._consumption is not None:
return round(self._consumption, 1)
return self._consumption
@property
def consumedWs(self) -> Optional[float]:
"""The average of energy consumed per second since last report call."""
if self._consumedWs is not None:
return round(self._consumedWs, 1)
return self._consumedWs
@property
def boot_id(self) -> Optional[str]:
"""A unique identifier to distinguish whether the energy counter has been reset."""
return self._boot_id
@property
def energy_since_boot(self) -> Optional[float]:
"""The total energy in watt seconds (Ws) that has been measured since the last power-up or restart of the device."""
if self._energy_since_boot is not None:
return round(self._energy_since_boot, 2)
return self._energy_since_boot
@property
def time_since_boot(self) -> Optional[int]:
"""The time in seconds that has elapsed since the last start or restart of the device."""
return self._time_since_boot
@property
def firmware(self) -> Optional[str]:
"""Return the current firmware."""
return self._firmware
@property
def mac(self) -> Optional[str]:
"""Return the MAC address."""
return self._mac
@property
def temperature(self) -> Optional[float]:
"""Return the current temperature in Celsius."""
if self._temperature is not None:
return round(self._temperature, 1)
return self._temperature
async def get_temperature_full(self) -> str:
"""Get current temperature in celsius."""
url = URL(self.uri).join(URL("temp"))
response = await request(self, uri=url, token=self._token)
return response
async def close(self) -> None:
"""Close an open client session."""
if self._session and self._close_session:
await self._session.close()
async def __aenter__(self) -> "MyStromSwitch":
"""Async enter."""
return self
async def __aexit__(self, *exc_info) -> None:
"""Async exit."""
await self.close()
|
