diff --git a/msmart/cloud.py b/msmart/cloud.py index 662401d7..7f89d5db 100644 --- a/msmart/cloud.py +++ b/msmart/cloud.py @@ -118,7 +118,7 @@ async def _api_request(self, endpoint: str, body: Dict[str, Any]) -> Optional[di # Sign the contents and add it to the header sign = self._security.sign(contents, random) headers = { - 'Content-Type': 'application/json', + "Content-Type": "application/json", "secretVersion": "1", "sign": sign, "random": random, @@ -212,7 +212,7 @@ async def get_token(self, udpid: str) -> Tuple[str, str]: """Get token and key for the provided udpid.""" response = await self._api_request( - '/v1/iot/secure/getToken', + "/v1/iot/secure/getToken", self._build_request_body({"udpid": udpid}) ) diff --git a/msmart/discover.py b/msmart/discover.py index 180a43ce..8b39412b 100644 --- a/msmart/discover.py +++ b/msmart/discover.py @@ -248,9 +248,9 @@ def _get_device_version(cls, data: bytes) -> int: # Use start of packet data to differentiate between V2 and V3 start_of_packet = data_mv[:2] - if start_of_packet == b'\x5a\x5a': + if start_of_packet == b"\x5a\x5a": return 2 - elif start_of_packet == b'\x83\x70': + elif start_of_packet == b"\x83\x70": return 3 raise DiscoverError() @@ -307,7 +307,7 @@ async def _get_device_info(cls, ip: str, version: int, data: bytes) -> Dict[str, encrypted_data = data_mv[40:-16] # Extract ID - device_id = int.from_bytes(data_mv[20:26], 'little') + device_id = int.from_bytes(data_mv[20:26], "little") # Attempt to decrypt the packet try: @@ -323,7 +323,7 @@ async def _get_device_info(cls, ip: str, version: int, data: bytes) -> Dict[str, # Extract IP and port ip_address = str(ipaddress.IPv4Address( decrypted_mv[3::-1].tobytes())) - port = int.from_bytes(decrypted_mv[4:6], 'little') + port = int.from_bytes(decrypted_mv[4:6], "little") if ip_address != ip: _LOGGER.warning( @@ -336,7 +336,7 @@ async def _get_device_info(cls, ip: str, version: int, data: bytes) -> Dict[str, name_length = decrypted_mv[40] name = decrypted_mv[41:41+name_length].tobytes().decode() - device_type = int(name.split('_')[1], 16) + device_type = int(name.split("_")[1], 16) # Return dictionary of device info return {"ip": ip_address, "port": port, "device_id": device_id, "name": name, "sn": sn, "device_type": device_type, "version": version} diff --git a/msmart/lan.py b/msmart/lan.py index dd34536f..92679d3b 100644 --- a/msmart/lan.py +++ b/msmart/lan.py @@ -77,7 +77,7 @@ def connection_made(self, transport) -> None: self._transport = transport # Save peer name for logging - peername = transport.get_extra_info('peername') + peername = transport.get_extra_info("peername") self._peer = self._format_socket_name(peername) _LOGGER.debug("Connected to %s.", self._peer) @@ -660,11 +660,11 @@ def encode(cls, device_id: int, command: bytes) -> bytes: header = b"\x5A\x5A" # Start of packet header += b"\x01\x11" # Message type - header += length.to_bytes(2, 'little') # Packet size + header += length.to_bytes(2, "little") # Packet size header += b"\x20\x00" # Magic bytes header += bytes(4) # Message ID header += cls._timestamp() # Timestamp - header += device_id.to_bytes(8, 'little') # Device ID + header += device_id.to_bytes(8, "little") # Device ID header += bytes(12) # ??? packet = header + encrypted_payload