platypush/platypush
platypush/platypush
2
2
Fork 4
Code Issues 30 Pull requests Projects 4 Releases Packages Wiki Activity

[switch.tplink] Extended exception handling to all SmartDevice methods.
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse source
This commit is contained in:
Fabio Manganiello 2024-11-04 00:28:03 +01:00
parent 43e88c71c6
commit 3e54d5d7b3
Signed by: blacklight
GPG key ID: D90FBA7F76362774
1 changed files with 32 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

45
platypush/plugins/switch/tplink/__init__.py
View file

@ -196,16 +196,21 @@ class SwitchTplinkPlugin(RunnablePlugin, SwitchEntityManager):
) )
return None return None
def _serialize(self, device: SmartDevice) -> dict: def _serialize(self, device: SmartDevice) -> Optional[dict]:
return { try:
'current_consumption': self._current_consumption(device), return {
'id': device.host, 'current_consumption': self._current_consumption(device),
'ip': device.host, 'id': device.host,
'host': device.host, 'ip': device.host,
'hw_info': device.hw_info, 'host': device.host,
'name': device.alias, 'hw_info': device.hw_info,
'on': device.is_on, 'name': device.alias,
} 'on': device.is_on,
}
except SmartDeviceException as e:
self.logger.warning(
'Could not communicate with device %s: %s', device.host, e
)
@action @action
def status(self, *_, **__) -> List[dict]: def status(self, *_, **__) -> List[dict]:
@ -227,15 +232,29 @@ class SwitchTplinkPlugin(RunnablePlugin, SwitchEntityManager):
] ]
""" """
return [self._serialize(dev) for dev in self._scan().values()] return [
ser_dev
for ser_dev in [self._serialize(dev) for dev in self._scan().values()]
if ser_dev
]
def main(self): def main(self):
devices = {ip: self._serialize(dev) for ip, dev in self._ip_to_dev.items()} devices = {
ip_: dev_
for ip_, dev_ in {
ip: self._serialize(dev) for ip, dev in self._ip_to_dev.items()
}.items()
if dev_
}
while not self.should_stop(): while not self.should_stop():
new_devices = self._scan(publish_entities=False) new_devices = self._scan(publish_entities=False)
new_serialized_devices = { new_serialized_devices = {
ip: self._serialize(dev) for ip, dev in new_devices.items() ip_: dev_
for ip_, dev_ in {
ip: self._serialize(dev) for ip, dev in new_devices.items()
}.items()
if dev_
} }
updated_devices = { updated_devices = {