williamp/localtuya-modded
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Run pylint from tox (#243)

Browse Source
This commit is contained in:
Pierre Ståhl
2020-12-17 09:26:55 +01:00
committed by GitHub
parent 3cf15f216e
commit f72cbc0b5a
9 changed files with 132 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
custom_components/localtuya/__init__.py
View File

@@ -209,7 +209,7 @@ async def async_setup(hass: HomeAssistant, config: dict):
await discovery.start()
hass.data[DOMAIN][DATA_DISCOVERY] = discovery
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _shutdown)
except Exception:
except Exception: # pylint: disable=broad-except
_LOGGER.exception("failed to set up discovery")
hass.helpers.service.async_register_admin_service(

20
custom_components/localtuya/common.py
View File

@@ -65,7 +65,7 @@ async def async_setup_entry(
# Add DPS used by this platform to the request list
for dp_conf in dps_config_fields:
if dp_conf in device_config:
tuyainterface._dps_to_request[device_config[dp_conf]] = None
tuyainterface.add_dps_to_request(device_config[dp_conf])
entities.append(
entity_class(
@@ -93,11 +93,22 @@ def get_entity_config(config_entry, dp_id):
raise Exception(f"missing entity config for id {dp_id}")
@callback
def async_config_entry_by_device_id(hass, device_id):
"""Look up config entry by device id."""
current_entries = hass.config_entries.async_entries(DOMAIN)
for entry in current_entries:
if entry.data[CONF_DEVICE_ID] == device_id:
return entry
return None
class TuyaDevice(pytuya.TuyaListener, pytuya.ContextualLogger):
"""Cache wrapper for pytuya.TuyaInterface."""
def __init__(self, hass, config_entry):
"""Initialize the cache."""
super().__init__()
self._hass = hass
self._config_entry = config_entry
self._interface = None
@@ -140,7 +151,7 @@ class TuyaDevice(pytuya.TuyaListener, pytuya.ContextualLogger):
raise Exception("Failed to retrieve status")
self.status_updated(status)
except Exception:
except Exception: # pylint: disable=broad-except
self.exception(f"Connect to {self._config_entry[CONF_HOST]} failed")
if self._interface is not None:
await self._interface.close()
@@ -161,7 +172,7 @@ class TuyaDevice(pytuya.TuyaListener, pytuya.ContextualLogger):
if self._interface is not None:
try:
await self._interface.set_dp(state, dp_index)
except Exception:
except Exception: # pylint: disable=broad-except
self.exception("Failed to set DP %d to %d", dp_index, state)
else:
self.error(
@@ -173,7 +184,7 @@ class TuyaDevice(pytuya.TuyaListener, pytuya.ContextualLogger):
if self._interface is not None:
try:
await self._interface.set_dps(states)
except Exception:
except Exception: # pylint: disable=broad-except
self.exception("Failed to set DPs %r", states)
else:
self.error(
@@ -203,6 +214,7 @@ class LocalTuyaEntity(Entity, pytuya.ContextualLogger):
def __init__(self, device, config_entry, dp_id, logger, **kwargs):
"""Initialize the Tuya entity."""
super().__init__()
self._device = device
self._config_entry = config_entry
self._config = get_entity_config(config_entry, dp_id)

10
custom_components/localtuya/config_flow.py
View File

@@ -171,10 +171,10 @@ async def validate_input(hass: core.HomeAssistant, data):
)
detected_dps = await interface.detect_available_dps()
except (ConnectionRefusedError, ConnectionResetError):
raise CannotConnect
except ValueError:
raise InvalidAuth
except (ConnectionRefusedError, ConnectionResetError) as ex:
raise CannotConnect from ex
except ValueError as ex:
raise InvalidAuth from ex
finally:
if interface:
await interface.close()
@@ -230,7 +230,7 @@ class LocaltuyaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors["base"] = "address_in_use"
else:
errors["base"] = "discovery_failed"
except Exception:
except Exception: # pylint: disable= broad-except
_LOGGER.exception("discovery failed")
errors["base"] = "discovery_failed"

12
custom_components/localtuya/discovery.py
View File

@@ -54,7 +54,7 @@ class TuyaDiscovery(asyncio.DatagramProtocol):
def close(self):
"""Stop discovery."""
self.callback = None
self._callback = None
for transport, _ in self._listeners:
transport.close()
@@ -63,7 +63,7 @@ class TuyaDiscovery(asyncio.DatagramProtocol):
data = data[20:-8]
try:
data = decrypt_udp(data)
except Exception:
except Exception: # pylint: disable=broad-except
data = data.decode()
decoded = json.loads(data)
@@ -81,10 +81,10 @@ class TuyaDiscovery(asyncio.DatagramProtocol):
async def discover():
"""Discover and return devices on local network."""
discover = TuyaDiscovery()
discovery = TuyaDiscovery()
try:
await discover.start()
await discovery.start()
await asyncio.sleep(DEFAULT_TIMEOUT)
finally:
discover.close()
return discover.devices
discovery.close()
return discovery.devices

2
custom_components/localtuya/light.py
View File

@@ -188,7 +188,7 @@ class LocaltuyaLight(LocalTuyaEntity, LightEntity):
"""Return the hs color value."""
if self.is_color_mode:
return self._hs
elif self.supported_features & SUPPORT_COLOR:
if self.supported_features & SUPPORT_COLOR:
return [0, 0]
return None

36
custom_components/localtuya/pytuya/__init__.py
View File

@@ -178,7 +178,7 @@ class AESCipher:
def __init__(self, key):
"""Initialize a new AESCipher."""
self.bs = 16
self.block_size = 16
self.cipher = Cipher(algorithms.AES(key), modes.ECB(), default_backend())
def encrypt(self, raw, use_base64=True):
@@ -195,13 +195,13 @@ class AESCipher:
decryptor = self.cipher.decryptor()
return self._unpad(decryptor.update(enc) + decryptor.finalize()).decode()
def _pad(self, s):
padnum = self.bs - len(s) % self.bs
return s + padnum * chr(padnum).encode()
def _pad(self, data):
padnum = self.block_size - len(data) % self.block_size
return data + padnum * chr(padnum).encode()
@staticmethod
def _unpad(s):
return s[: -ord(s[len(s) - 1 :])]
def _unpad(data):
return data[: -ord(data[len(data) - 1 :])]
class MessageDispatcher(ContextualLogger):
@@ -213,6 +213,7 @@ class MessageDispatcher(ContextualLogger):
def __init__(self, dev_id, listener):
"""Initialize a new MessageBuffer."""
super().__init__()
self.buffer = b""
self.listeners = {}
self.listener = listener
@@ -311,7 +312,7 @@ class TuyaListener(ABC):
"""Device updated status."""
@abstractmethod
def disconnected(self, exc):
def disconnected(self):
"""Device disconnected."""
@@ -321,7 +322,7 @@ class EmptyListener(TuyaListener):
def status_updated(self, status):
"""Device updated status."""
def disconnected(self, exc):
def disconnected(self):
"""Device disconnected."""
@@ -340,6 +341,7 @@ class TuyaProtocol(asyncio.Protocol, ContextualLogger):
Attributes:
port (int): The port to connect to.
"""
super().__init__()
self.loop = asyncio.get_running_loop()
self.set_logger(_LOGGER, dev_id)
self.id = dev_id
@@ -381,7 +383,7 @@ class TuyaProtocol(asyncio.Protocol, ContextualLogger):
except asyncio.CancelledError:
self.debug("Stopped heartbeat loop")
raise
except Exception as ex:
except Exception as ex: # pylint: disable=broad-except
self.exception("Heartbeat failed (%s), disconnecting", ex)
break
@@ -404,7 +406,7 @@ class TuyaProtocol(asyncio.Protocol, ContextualLogger):
listener = self.listener and self.listener()
if listener is not None:
listener.disconnected()
except Exception:
except Exception: # pylint: disable=broad-except
self.exception("Failed to call disconnected callback")
async def close(self):
@@ -503,8 +505,8 @@ class TuyaProtocol(asyncio.Protocol, ContextualLogger):
self.add_dps_to_request(range(*dps_range))
try:
data = await self.status()
except Exception as e:
self.exception("Failed to get status: %s", e)
except Exception as ex:
self.exception("Failed to get status: %s", ex)
raise
if "dps" in data:
self.dps_cache.update(data["dps"])
@@ -525,7 +527,7 @@ class TuyaProtocol(asyncio.Protocol, ContextualLogger):
if not payload:
payload = "{}"
elif payload.startswith(b"{"):
payload = payload
pass
elif payload.startswith(PROTOCOL_VERSION_BYTES_31):
payload = payload[len(PROTOCOL_VERSION_BYTES_31) :] # remove version header
# remove (what I'm guessing, but not confirmed is) 16-bytes of MD5
@@ -591,7 +593,7 @@ class TuyaProtocol(asyncio.Protocol, ContextualLogger):
payload = PROTOCOL_33_HEADER + payload
elif command == SET:
payload = self.cipher.encrypt(payload)
preMd5String = (
to_hash = (
b"data="
+ payload
+ b"||lpv="
@@ -599,9 +601,9 @@ class TuyaProtocol(asyncio.Protocol, ContextualLogger):
+ b"||"
+ self.local_key
)
m = md5()
m.update(preMd5String)
hexdigest = m.hexdigest()
hasher = md5()
hasher.update(to_hash)
hexdigest = hasher.hexdigest()
payload = (
PROTOCOL_VERSION_BYTES_31
+ hexdigest[8:][:16].encode("latin1")