/
switch.py
105 lines (86 loc) · 3.18 KB
/
switch.py
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
"""Support for communicating with myStrom plugs/switches."""
import aiohttp
from yarl import URL
from typing import Any, Dict, Iterable, List, Optional, Union
from . import _request as request
class MyStromSwitch:
"""A class for a myStrom switch/plug."""
def __init__(self, host: str, session: aiohttp.client.ClientSession = None) -> None:
"""Initialize the switch."""
self._close_session = False
self._host = host
self._session = session
self._consumption = 0
self._state = None
self._temperature = None
self._firmware = None
self._mac = 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)
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)
await self.get_state()
async def toggle(self) -> None:
"""Toggle the relay."""
url = URL(self.uri).join(URL("toggle"))
await request(self, uri=url)
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)
self._consumption = response["power"]
self._state = response["relay"]
try:
self._temperature = response["temperature"]
except KeyError:
self._temperature = None
url = URL(self.uri).join(URL("info.json"))
response = await request(self, uri=url)
self._firmware = response["version"]
self._mac = response["mac"]
@property
def relay(self) -> bool:
"""Return the relay state."""
return bool(self._state)
@property
def consumption(self) -> float:
"""Return the current power consumption in mWh."""
return round(self._consumption, 1)
@property
def firmware(self) -> float:
"""Return the current firmware."""
return self._firmware
@property
def mac(self) -> float:
"""Return the MAC address."""
return self._mac
@property
def temperature(self) -> 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)
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()