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
|
import logging
import typing
from .device_service import SHCDeviceService
from .exceptions import SHCException
from .services_impl import SUPPORTED_DEVICE_SERVICE_IDS, build
logger = logging.getLogger("boschshcpy")
class SHCDevice:
def __init__(self, api, raw_device, raw_device_services):
self._api = api
self._raw_device = raw_device
self._callbacks = {}
self._device_services_by_id = {}
if not raw_device_services:
raw_device_services = self._enumerate_services()
self._init_services(raw_device_services)
def _enumerate_services(self):
raw_device_services = []
for device_service_id in self._raw_device["deviceServiceIds"]:
if device_service_id not in SUPPORTED_DEVICE_SERVICE_IDS:
continue
raw_device_service_data = self._api.get_device_service(
self.id, device_service_id
)
raw_device_services.append(raw_device_service_data)
return raw_device_services
def _init_services(self, raw_device_services):
for raw_device_service_data in raw_device_services:
device_service = build(self._api, raw_device_service_data)
self._device_services_by_id[raw_device_service_data["id"]] = device_service
@property
def root_device_id(self):
return self._raw_device["rootDeviceId"]
@property
def id(self):
return self._raw_device["id"]
@property
def manufacturer(self):
return self._raw_device["manufacturer"]
@property
def room_id(self):
return self._raw_device["roomId"] if "roomId" in self._raw_device else None
@property
def device_model(self):
return self._raw_device["deviceModel"]
@property
def serial(self):
return self._raw_device["serial"] if "serial" in self._raw_device else None
@property
def profile(self):
return self._raw_device["profile"] if "profile" in self._raw_device else None
@property
def name(self):
return self._raw_device["name"]
@property
def status(self):
return self._raw_device["status"]
@property
def deleted(self):
return (
True
if "deleted" in self._raw_device and self._raw_device["deleted"] == True
else False
)
@property
def child_device_ids(self):
return (
self._raw_device["childDeviceIds"]
if "childDeviceIds" in self._raw_device
else None
)
@property
def parent_device_id(self):
return (
self._raw_device["parentDeviceId"]
if "parentDeviceId" in self._raw_device
else None
)
@property
def device_services(self) -> typing.Sequence[SHCDeviceService]:
return list(self._device_services_by_id.values())
@property
def device_service_ids(self) -> typing.Set[str]:
return set(self._device_services_by_id.keys())
def subscribe_callback(self, entity, callback):
self._callbacks[entity] = callback
def unsubscribe_callback(self, entity):
self._callbacks.pop(entity, None)
def update_raw_information(self, raw_device):
if self._raw_device["id"] != raw_device["id"]:
raise SHCException("Error due to mismatching device ids!")
self._raw_device = raw_device
for callback in self._callbacks:
self._callbacks[callback]()
def device_service(self, device_service_id):
return (
self._device_services_by_id[device_service_id]
if device_service_id in self._device_services_by_id
else None
)
def update(self):
for service in self.device_services:
service.short_poll()
def summary(self):
print(f"Device: {self.id}")
print(f" Name : {self.name}")
print(f" Manufacturer : {self.manufacturer}")
print(f" Model : {self.device_model}")
print(f" Room : {self.room_id}")
print(f" Serial : {self.serial}")
print(f" Profile : {self.profile}")
print(f" Status : {self.status}")
print(f" ParentDevice : {self.parent_device_id}")
print(f" ChildDevices : {self.child_device_ids}")
print(f" DeviceServices: {self._raw_device['deviceServiceIds']}")
for device_service in self.device_services:
device_service.summary()
def process_long_polling_poll_result(self, raw_result):
assert raw_result["@type"] == "DeviceServiceData"
device_service_id = raw_result["id"]
if device_service_id in self._device_services_by_id.keys():
device_service = self._device_services_by_id[device_service_id]
device_service.process_long_polling_poll_result(raw_result)
else:
logger.debug(
f"Skipping polling result with unknown device service id {device_service_id}."
)
|
