Mirrored_Repos/python-kasa
1
0
Fork 0
mirror of https://github.com/python-kasa/python-kasa.git synced 2025-01-08 22:07:06 +00:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

Simplify device class detection, fix hardcoded timeout for discovery (#112)

Browse Source
This commit is contained in:
Teemu R 2020-11-08 14:32:27 +01:00 committed by GitHub
parent 70061cbe67
commit 56eb2cdcb5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 14 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
kasa/discover.py
View File

@ -33,13 +33,11 @@ class _DiscoverProtocol(asyncio.DatagramProtocol):
*, *,
on_discovered: OnDiscoveredCallable = None, on_discovered: OnDiscoveredCallable = None,
target: str = "255.255.255.255", target: str = "255.255.255.255",
timeout: int = 5,
discovery_packets: int = 3, discovery_packets: int = 3,
interface: Optional[str] = None, interface: Optional[str] = None,
): ):
self.transport = None self.transport = None
self.tries = discovery_packets self.discovery_packets = discovery_packets
self.timeout = timeout
self.interface = interface self.interface = interface
self.on_discovered = on_discovered self.on_discovered = on_discovered
self.protocol = TPLinkSmartHomeProtocol() self.protocol = TPLinkSmartHomeProtocol()
@ -65,7 +63,7 @@ class _DiscoverProtocol(asyncio.DatagramProtocol):
req = json.dumps(Discover.DISCOVERY_QUERY) req = json.dumps(Discover.DISCOVERY_QUERY)
_LOGGER.debug("[DISCOVERY] %s >> %s", self.target, Discover.DISCOVERY_QUERY) _LOGGER.debug("[DISCOVERY] %s >> %s", self.target, Discover.DISCOVERY_QUERY)
encrypted_req = self.protocol.encrypt(req) encrypted_req = self.protocol.encrypt(req)
for i in range(self.tries): for i in range(self.discovery_packets):
self.transport.sendto(encrypted_req[4:], self.target) # type: ignore self.transport.sendto(encrypted_req[4:], self.target) # type: ignore
def datagram_received(self, data, addr) -> None: def datagram_received(self, data, addr) -> None:
@ -176,7 +174,6 @@ class Discover:
lambda: _DiscoverProtocol( lambda: _DiscoverProtocol(
target=target, target=target,
on_discovered=on_discovered, on_discovered=on_discovered,
timeout=timeout,
discovery_packets=discovery_packets, discovery_packets=discovery_packets,
interface=interface, interface=interface,
), ),
@ -186,7 +183,7 @@ class Discover:
try: try:
_LOGGER.debug("Waiting %s seconds for responses...", timeout) _LOGGER.debug("Waiting %s seconds for responses...", timeout)
await asyncio.sleep(5) await asyncio.sleep(timeout)
finally: finally:
transport.close() transport.close()
@ -220,32 +217,24 @@ class Discover:
@staticmethod @staticmethod
def _get_device_class(info: dict) -> Type[SmartDevice]: def _get_device_class(info: dict) -> Type[SmartDevice]:
"""Find SmartDevice subclass for device described by passed data.""" """Find SmartDevice subclass for device described by passed data."""
if "system" in info and "get_sysinfo" in info["system"]: if "system" not in info or "get_sysinfo" not in info["system"]:
sysinfo = info["system"]["get_sysinfo"] raise SmartDeviceException("No 'system' or 'get_sysinfo' in response")
if "type" in sysinfo:
type_ = sysinfo["type"]
elif "mic_type" in sysinfo:
type_ = sysinfo["mic_type"]
else:
raise SmartDeviceException("Unable to find the device type field!")
else:
raise SmartDeviceException("No 'system' nor 'get_sysinfo' in response")
if ( sysinfo = info["system"]["get_sysinfo"]
"smartlife.iot.dimmer" in info type_ = sysinfo.get("type", sysinfo.get("mic_type"))
and "get_dimmer_parameters" in info["smartlife.iot.dimmer"] if type_ is None:
): raise SmartDeviceException("Unable to find the device type field!")
if "dev_name" in sysinfo and "Dimmer" in sysinfo["dev_name"]:
return SmartDimmer return SmartDimmer
elif "smartplug" in type_.lower() and "children" in sysinfo: if "smartplug" in type_.lower():
return SmartStrip
elif "smartplug" in type_.lower():
if "children" in sysinfo: if "children" in sysinfo:
return SmartStrip return SmartStrip
return SmartPlug return SmartPlug
elif "smartbulb" in type_.lower():
if "smartbulb" in type_.lower():
if "length" in sysinfo: # strips have length if "length" in sysinfo: # strips have length
return SmartLightStrip return SmartLightStrip