forked from Danielhiversen/pyMillLocal
-
Notifications
You must be signed in to change notification settings - Fork 0
/
__init__.py
98 lines (86 loc) · 3.22 KB
/
__init__.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
"""Local support for Mill wifi-enabled home heaters."""
import asyncio
import json
import logging
import async_timeout
_LOGGER = logging.getLogger(__name__)
class Mill:
"""Mill data handler."""
def __init__(self, device_ip, websession, timeout=15):
"""Init Mill data handler."""
self.device_ip = device_ip
self.websession = websession
self.url = "http:https://" + device_ip
self._timeout = timeout
self._status = {}
@property
def version(self):
"""Return version."""
return self._status.get("version", "")
@property
def name(self):
"""Return name."""
return self._status.get("name", "")
async def set_target_temperature(self, target_temperature):
"""Set target temperature."""
payload = {
"type": "Normal",
"value": target_temperature,
}
with async_timeout.timeout(self._timeout):
async with self.websession.post(
f"{self.url}/set-temperature",
data=json.dumps(payload),
) as response:
_LOGGER.debug("Heater response %s", response.status)
if response.status != 200:
_LOGGER.error(
"Failed to set target temperature %s %s",
response.status,
response.reason,
)
return response.status
async def set_normal_operation_mode(self):
"""Set target temperature."""
payload = {"mode": "Control individually"}
with async_timeout.timeout(self._timeout):
async with self.websession.post(
f"{self.url}/operation-mode",
payload=payload,
) as response:
_LOGGER.debug("Heater response %s", response.status)
if response.status != 200:
_LOGGER.error(
"Failed to set target temperature %s %s",
response.status,
response.reason,
)
return response.status
async def get_status(self):
"""Get heater control status."""
self._status = await self._request("status")
return self._status
async def get_control_status(self):
"""Get heater status."""
return await self._request("control-status")
async def _request(self, command):
try:
with async_timeout.timeout(self._timeout):
async with self.websession.get(
f"{self.url}/{command}",
) as response:
if response.status != 200:
_LOGGER.error(
"Failed to get %s %s %s",
command,
response.status,
response.reason,
)
return None
res = await response.json()
if res["status"] != "ok":
_LOGGER.error("Request %s failed: %s", command, res)
return None
return res
except asyncio.TimeoutError:
return None