Fix lint errors

Author: lafriks

karcher/cli.py

@@ -25,7 +25,11 @@ def default(self, o):
         return super().default(o)
 
 class GlobalContextObject:
-    def __init__(self, debug: int = 0, output: str = "json", region: Region = Region.EU):
+    def __init__(self,
+        debug: int = 0,
+        output: str = "json",
+        region: Region = Region.EU
+    ):
         self.debug = debug
         self.output = output
         self.region = region
@@ -104,17 +108,17 @@ def devices(ctx: click.Context, username: str, password: str, token: str):
 
     kh = KarcherHome(region=ctx.obj.region)
     sess = None
-    if token != None:
+    if token is not None:
         sess = Session.from_token(token, '')
-    elif username != None and password != None:
+    elif username is not None and password is not None:
         sess = kh.login(username, password)
     else:
         raise click.BadParameter('Must provide either token or username and password.')
 
     devices = kh.get_devices(sess)
 
     # Logout if we used a username and password
-    if token == None:
+    if token is None:
         kh.logout(sess)
 
     ctx.obj.print(devices)

karcher/consts.py

@@ -5,6 +5,7 @@
 
 from enum import Enum
 
+
 class Region(str, Enum):
     """Region enum.
 
@@ -14,6 +15,7 @@ class Region(str, Enum):
     US = 'us'
     CN = 'cn'
 
+
 class Language(int, Enum):
     """Language enum.
 
@@ -57,13 +59,15 @@ def __str__(self):
         else:
             return 'en'
 
+
 class Product(str, Enum):
     """Product model enum."""
 
     RCV3 = '1528986273083777024'
     RCV5 = '1540149850806333440'
     RCF5 = '1599715149861306368'
 
+
 REGION_URLS = {
     Region.EU: "https://eu-appaiot.3irobotix.net",
     Region.US: "https://us-appaiot.3irobotix.net",

karcher/device.py

@@ -11,11 +11,13 @@
 from .consts import Product
 from .utils import snake_case
 
+
 class DeviceStatus(int, Enum):
     """Device status enum."""
     Offline = 0
     Online = 1
 
+
 @dataclass(init=False)
 class DeviceVersion:
     """Device version class.
@@ -34,6 +36,7 @@ def __init__(self, **kwargs):
             if k in names:
                 setattr(self, k, v)
 
+
 @dataclass(init=False)
 class Device:
     """Device class.

karcher/exception.py

@@ -28,14 +28,17 @@ def __init__(self, message):
         self.message = message
         super().__init__(608, self.message)
 
+
 def handle_error_code(code, message):
     if code == 608:
         raise KarcherHomeAccessDenied('Forbidden')
     elif code == 609:
-        raise KarcherHomeAccessDenied('Unauthorized or authorization expired, please log in again')
+        raise KarcherHomeAccessDenied(
+            'Unauthorized or authorization expired, please log in again')
     elif code == 613:
         raise KarcherHomeException(613, 'Invalid token')
     elif code == 620:
-        raise KarcherHomeException(620, 'The username or password is incorrect')
+        raise KarcherHomeException(
+            620, 'The username or password is incorrect')
     else:
         raise KarcherHomeException(code, message)

karcher/identifiers.py

@@ -6,6 +6,7 @@
 """Compat layer for homeassistant."""
 from enum import Enum, auto
 
+
 class VacuumState(Enum):
     """Vacuum state enum.
 

karcher/karcher.py

@@ -9,11 +9,15 @@
 import urllib.parse
 
 from .auth import Domains, Session
-from .consts import APP_VERSION_CODE, APP_VERSION_NAME, PROJECT_TYPE, PROTOCOL_VERSION, REGION_URLS, TENANT_ID, Region, Language
+from .consts import APP_VERSION_CODE, APP_VERSION_NAME, PROJECT_TYPE, \
+    PROTOCOL_VERSION, REGION_URLS, TENANT_ID, \
+    Region, Language
 from .device import Device
 from .exception import KarcherHomeAccessDenied, KarcherHomeException, handle_error_code
 from .map import Map
-from .utils import decrypt, decrypt_map, encrypt, get_nonce, get_random_string, get_timestamp, is_email, md5
+from .utils import decrypt, decrypt_map, encrypt, get_nonce, get_random_string, \
+    get_timestamp, is_email, md5
+
 
 class KarcherHome:
     """Main class to access Karcher Home Robots API"""
@@ -94,13 +98,15 @@ def _download(self, url):
 
         resp = session.get(url, headers=headers)
         if resp.status_code != 200:
-            raise KarcherHomeException(-1, 'HTTP error: ' + str(resp.status_code))
+            raise KarcherHomeException(-1,
+                                       'HTTP error: ' + str(resp.status_code))
 
         return resp.content
 
-    def _process_response(self, resp, prop = None):
+    def _process_response(self, resp, prop=None):
         if resp.status_code != 200:
-            raise KarcherHomeException(-1, 'HTTP error: ' + str(resp.status_code))
+            raise KarcherHomeException(-1,
+                                       'HTTP error: ' + str(resp.status_code))
         data = resp.json()
         # Check for error response
         if data['code'] != 0:
@@ -112,7 +118,7 @@ def _process_response(self, resp, prop = None):
         result = data['result']
         if type(result) == str:
             raise KarcherHomeException(-2, 'Invalid response: ' + result)
-        if prop != None:
+        if prop is not None:
             return json.loads(decrypt(result[prop]))
         return result
 
@@ -128,7 +134,7 @@ def get_urls(self):
         d = self._process_response(resp, 'domain')
         return Domains(**d)
 
-    def login(self, username, password, register_id = None):
+    def login(self, username, password, register_id=None):
         """Login using provided credentials."""
 
         if register_id is None or register_id == '':
@@ -164,14 +170,15 @@ def login(self, username, password, register_id = None):
 
     def logout(self, sess: Session):
         """End current session.
-        
+
         This will also reset the session object.
         """
         if sess.auth_token == '' or sess.user_id == '':
             sess.reset()
             return
-        
-        self._process_response(self._request(sess, 'POST', '/user-center/auth/logout'))
+
+        self._process_response(self._request(
+            sess, 'POST', '/user-center/auth/logout'))
         sess.reset()
 
     def get_devices(self, sess: Session):
@@ -180,21 +187,26 @@ def get_devices(self, sess: Session):
         if sess is None or sess.auth_token == '' or sess.user_id == '':
             raise KarcherHomeAccessDenied('Not authorized')
 
-        resp = self._request(sess, 'GET', '/smart-home-service/smartHome/user/getDeviceInfoByUserId/' + sess.user_id)
+        resp = self._request(
+            sess, 'GET',
+            '/smart-home-service/smartHome/user/getDeviceInfoByUserId/' + sess.user_id)
 
         return [Device(**d) for d in self._process_response(resp)]
 
     def get_map_data(self, sess: Session, dev: Device, map: int = 1):
         # <tenantId>/<modeType>/<deviceSn>/01-01-2022/map/temp/0046690461_<deviceSn>_1
         mapDir = TENANT_ID + '/' + dev.product_mode_code + '/' +\
-            dev.sn + '/01-01-2022/map/temp/0046690461_' + dev.sn + '_' + str(map)
-
-        resp = self._request(sess, 'POST', '/storage-management/storage/aws/getAccessUrl', json={
-            'dir': mapDir,
-            'countryCode': sess.get_country_code(),
-            'serviceType': 2,
-            'tenantId': TENANT_ID,
-        })
+            dev.sn + '/01-01-2022/map/temp/0046690461_' + \
+            dev.sn + '_' + str(map)
+
+        resp = self._request(sess, 'POST',
+                             '/storage-management/storage/aws/getAccessUrl',
+                             json={
+                                 'dir': mapDir,
+                                 'countryCode': sess.get_country_code(),
+                                 'serviceType': 2,
+                                 'tenantId': TENANT_ID,
+                             })
 
         d = self._process_response(resp)
         downloadUrl = d['url']
@@ -207,21 +219,3 @@ def get_map_data(self, sess: Session, dev: Device, map: int = 1):
             return Map.parse(data)
         else:
             return json.loads(data)
-
-    def get_families(self, sess: Session):
-
-        if sess is None or sess.auth_token == '' or sess.user_id == '':
-            raise KarcherHomeAccessDenied('Not authorized')
-        
-        resp = self._request(sess, 'GET', '/smart-home-service/smartHome/familyInfo/list/' + sess.user_id)
-
-        return self._process_response(resp)
-
-    def get_consumables(self, sess: Session, familyID: str):
-
-        if sess is None or sess.auth_token == '' or sess.user_id == '':
-            raise KarcherHomeAccessDenied('Not authorized')
-        
-        resp = self._request(sess, 'GET', '/smart-home-service/smartHome/consumablesInfo/getConsumablesInfoByFamilyId/' + familyID)
-
-        return self._process_response(resp)

karcher/map.py

@@ -1,8 +1,14 @@
+# -----------------------------------------------------------
+# Copyright (c) 2023 Lauris BH
+# SPDX-License-Identifier: MIT
+# -----------------------------------------------------------
+
 from google.protobuf.json_format import MessageToDict
 
 from . import mapdata_pb2
 from .utils import snake_case, snake_case_fields
 
+
 class Map:
     @staticmethod
     def parse(data: bytes):

karcher/utils.py

@@ -14,35 +14,44 @@
 
 from .consts import TENANT_ID, Product
 
+
+EMAIL_REGEX = "^\\w+([-+.]\\w+)*@\\w+([-.]\\w+)*\\.\\w+([-.]\\w+)*$"
+
 def get_random_string(length):
     letters = string.digits + string.ascii_lowercase + string.ascii_uppercase
     result_str = ''.join(random.choice(letters) for i in range(length))
     return result_str
 
+
 def get_nonce():
     return get_random_string(32)
 
+
 def get_timestamp():
     return int(time.time())
 
+
 def get_enc_key():
     m = hashlib.md5()
     m.update(bytes(TENANT_ID, 'utf-8'))
     h = m.hexdigest()
     return bytes(h[8:24], 'utf-8')
 
+
 def decrypt(data):
     cipher = AES.new(get_enc_key(), AES.MODE_ECB)
     buf = cipher.decrypt(base64.b64decode(data))
     return str(buf[:-ord(buf[-1:])], 'utf-8')
 
+
 def encrypt(data):
     cipher = AES.new(get_enc_key(), AES.MODE_ECB)
     buf = bytes(data, 'utf-8')
     pad_len = cipher.block_size - (len(buf) % cipher.block_size)
     buf = buf + bytes([pad_len]) * pad_len
     return base64.b64encode(cipher.encrypt(buf)).decode()
 
+
 def get_map_enc_key(sn: str, mac: str, product_id: Product):
     sub_key = mac.replace(':', '').lower() + str(product_id.value)
     cipher = AES.new(bytes(sub_key[0:16], 'utf-8'), AES.MODE_ECB)
@@ -57,6 +66,7 @@ def get_map_enc_key(sn: str, mac: str, product_id: Product):
     h = m.hexdigest()
     return bytes(h[8:24], 'utf-8')
 
+
 def decrypt_map(sn: str, mac: str, product_id: Product, data: bytes):
     cipher = AES.new(get_map_enc_key(sn, mac, product_id), AES.MODE_ECB)
     buf = cipher.decrypt(base64.b64decode(data))
@@ -65,18 +75,22 @@ def decrypt_map(sn: str, mac: str, product_id: Product, data: bytes):
     except Exception:
         return bytes.fromhex(str(buf[:-ord(buf[-1:])], 'utf-8'))
 
+
 def md5(data):
     m = hashlib.md5()
     m.update(bytes(data, 'utf-8'))
     return m.hexdigest()
 
+
 def is_email(email):
-    return re.search("^\\w+([-+.]\\w+)*@\\w+([-.]\\w+)*\\.\\w+([-.]\\w+)*$", email) is not None
+    return re.search(EMAIL_REGEX, email) is not None
+
 
 def snake_case(value):
     first_underscore = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', value)
     return re.sub('([a-z0-9])([A-Z])', r'\1_\2', first_underscore).lower()
 
+
 def snake_case_fields(data):
     if type(data) is dict:
         n = {}

ruff.toml

@@ -1,3 +1,6 @@
 extend-exclude = [
     "*_pb2.py*"
 ]
+
+[per-file-ignores]
+"tests/test_*.py" = ["E501"]