Merge pull request #8 from gentslava/fix/code_style

Fix - Code style

custom_components/elektronny_gorod/__init__.py

@@ -6,7 +6,7 @@
 from homeassistant.core import HomeAssistant
 
 from .const import DOMAIN
-from .coordinator import ElektronnyGorogDataUpdateCoordinator
+from .coordinator import ElektronnyGorodUpdateCoordinator
 
 PLATFORMS: list[Platform] = [
     Platform.CAMERA,
@@ -18,7 +18,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up Elektronny Gorod from a config entry."""
     hass.data.setdefault(DOMAIN, {})
 
-    coordinator = ElektronnyGorogDataUpdateCoordinator(hass, entry=entry)
+    coordinator = ElektronnyGorodUpdateCoordinator(hass, entry = entry)
     await coordinator.async_config_entry_first_refresh()
     hass.data[DOMAIN][entry.entry_id] = coordinator
 

custom_components/elektronny_gorod/api.py

@@ -16,16 +16,19 @@ def __init__(
         access_token: str | None = None,
         refresh_token: str | None = None,
         headers: dict = {}
-    ):
-        self.base_url: str = BASE_API_URL
-        self.headers: object = {**{
-            "Host": "api-mh.ertelecom.ru",
-            "User-Agent": generate_user_agent(),
-            "Content-Type": "application/json; charset=UTF-8",
-            "Authorization": "",
-            "Accept": "*/*",
-            "Accept-Language": "ru"
-        }, **headers}
+    ) -> None:
+        self.base_url: str = f"https://{BASE_API_URL}"
+        self.headers: dict = {
+            **{
+                "Host": BASE_API_URL,
+                "User-Agent": generate_user_agent(),
+                "Content-Type": "application/json; charset=UTF-8",
+                "Authorization": "",
+                "Accept": "*/*",
+                "Accept-Language": "ru"
+            },
+            **headers
+        }
         self.phone: str | None = None
         self.access_token: str | None = access_token
         self.refresh_token: str | None = refresh_token
@@ -38,7 +41,7 @@ async def query_contracts(self, phone: str):
         contracts = await self.request(api_url)
         return contracts if contracts else []
 
-    async def request_sms_code(self, contract: object):
+    async def request_sms_code(self, contract: dict):
         """Request SMS code for the selected contract."""
         api_url = f"{self.base_url}/auth/v2/confirmation/{self.phone}"
         data = json.dumps(
@@ -51,7 +54,7 @@ async def request_sms_code(self, contract: object):
         )
         return await self.request(api_url, data, method="POST")
 
-    async def verify_sms_code(self, contract:object, code:str) -> dict:
+    async def verify_sms_code(self, contract: dict, code: str) -> dict:
         """Verify the SMS code."""
         api_url = f"{self.base_url}/auth/v2/auth/{self.phone}/confirmation"
         data = json.dumps(
@@ -100,7 +103,7 @@ async def query_camera_stream(self, id) -> str | None:
     async def query_camera_snapshot(self, id, width, height) -> bytes:
         """Query the camera snapshot for the id."""
         api_url = f"{self.base_url}/rest/v1/forpost/cameras/{id}/snapshots?width={width}&height={height}"
-        return await self.request(api_url, binary=True)
+        return await self.request(api_url, binary = True)
 
     async def open_lock(self, place_id, access_control_id, entrance_id) -> list:
         """Query the list of places for subscriber."""
@@ -110,7 +113,7 @@ async def open_lock(self, place_id, access_control_id, entrance_id) -> list:
                 "name": "accessControlOpen"
             }
         )
-        return await self.request(api_url, data, method="POST")
+        return await self.request(api_url, data, method = "POST")
 
     async def request(
         self,
@@ -125,9 +128,9 @@ async def request(
         async with ClientSession() as session:
             LOGGER.info("Sending API request to %s with headers=%s and data=%s", url, self.headers, data)
             if method == "GET":
-                response = await session.get(url, headers=self.headers)
+                response = await session.get(url, headers = self.headers)
             elif method == "POST":
-                response = await session.post(url, data=data, headers=self.headers)
+                response = await session.post(url, data = data, headers = self.headers)
 
             if binary: return await response.read()
 

custom_components/elektronny_gorod/camera.py

@@ -4,15 +4,15 @@
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
 from .const import CONF_HEIGHT, CONF_WIDTH, DOMAIN, LOGGER
-from .coordinator import ElektronnyGorogDataUpdateCoordinator
+from .coordinator import ElektronnyGorodUpdateCoordinator
 
 async def async_setup_entry(
     hass: HomeAssistant,
     entry: ConfigEntry,
     async_add_entities: AddEntitiesCallback,
 ) -> None:
     """Set up Elektronny Gorog Camera based on a config entry."""
-    coordinator: ElektronnyGorogDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
+    coordinator: ElektronnyGorodUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
 
     # Get cameras info
     cameras_info = await coordinator.get_cameras_info()
@@ -28,12 +28,12 @@ async def async_setup_entry(
 class ElektronnyGorogCamera(Camera):
     def __init__(
         self,
-        coordinator: ElektronnyGorogDataUpdateCoordinator,
+        coordinator: ElektronnyGorodUpdateCoordinator,
         camera_info: dict
     ) -> None:
         LOGGER.info("ElektronnyGorogCamera init %s", camera_info)
         super().__init__()
-        self._coordinator: ElektronnyGorogDataUpdateCoordinator = coordinator
+        self._coordinator: ElektronnyGorodUpdateCoordinator = coordinator
         self._camera_info: dict = camera_info
         self._id = self._camera_info["ID"]
         self._name = self._camera_info["Name"]

custom_components/elektronny_gorod/const.py

@@ -3,7 +3,7 @@
 from typing import Final
 
 DOMAIN = "elektronny_gorod"
-BASE_API_URL: Final = "https://api-mh.ertelecom.ru"
+BASE_API_URL: Final = "api-mh.ertelecom.ru"
 LOGGER = logging.getLogger(__name__)
 
 CONF_ACCESS_TOKEN: Final = "access_token"

custom_components/elektronny_gorod/coordinator.py

@@ -15,7 +15,7 @@
 from .api import ElektronnyGorodAPI
 from .helpers import find
 
-class ElektronnyGorogDataUpdateCoordinator(DataUpdateCoordinator):
+class ElektronnyGorodUpdateCoordinator(DataUpdateCoordinator):
     """Class to manage fetching Elektronny Gorod data from single endpoint."""
     def __init__(
         self,
@@ -28,9 +28,9 @@ def __init__(
         self.refresh_token = entry.data[CONF_REFRESH_TOKEN]
         self.operatorId = entry.data[CONF_OPERATOR_ID]
         self.api = ElektronnyGorodAPI(
-            access_token=self.access_token,
-            refresh_token=self.refresh_token,
-            headers={
+            access_token = self.access_token,
+            refresh_token = self.refresh_token,
+            headers = {
                 "Operator": str(self.operatorId),
                 "Content-Type": "application/json"
             }
@@ -41,7 +41,7 @@ def __init__(
         super().__init__(
             hass,
             LOGGER,
-            name=DOMAIN,
+            name = DOMAIN,
         )
 
         async_dispatcher_connect(
@@ -56,7 +56,7 @@ def _notification_dismiss_listener(self, type, data) -> None:
     async def _async_update_data(self) -> None:
         """Handle device update. This function is only called once when the integration is added to Home Assistant."""
         try:
-            LOGGER.info("Integration starting...")
+            LOGGER.info("Integration starting")
             pass
         except Exception as ex:
             LOGGER.error("Integration start failed: %s", traceback.format_exc())
@@ -66,21 +66,21 @@ async def get_cameras_info(self) -> list:
         LOGGER.info("Get cameras info")
         return await self.api.query_cameras()
 
-    async def get_camera_stream(self, id) -> str | None:
+    async def get_camera_stream(self, camera_id) -> str | None:
         LOGGER.info("Get camera stream")
-        return await self.api.query_camera_stream(id)
+        return await self.api.query_camera_stream(camera_id)
 
-    async def get_camera_snapshot(self, id, width, height) -> bytes:
-        LOGGER.info(f"Get camera {id} snapshot with size {width}x{height}")
-        return await self.api.query_camera_snapshot(id, width, height)
+    async def get_camera_snapshot(self, camera_id, width, height) -> bytes:
+        LOGGER.info(f"Get camera {camera_id} snapshot with size {width}x{height}")
+        return await self.api.query_camera_snapshot(camera_id, width, height)
 
-    async def update_camera_state(self, id) -> dict:
-        LOGGER.info(f"Update camera {id} state")
+    async def update_camera_state(self, camera_id) -> dict:
+        LOGGER.info(f"Update camera {camera_id} state")
 
         cameras = await self.api.query_cameras()
         return find(
             cameras,
-            lambda camera: camera["ID"] == id
+            lambda camera: camera["ID"] == camera_id
         )
 
     async def get_locks_info(self) -> list:

custom_components/elektronny_gorod/helpers.py

@@ -1,18 +1,19 @@
 import json
 import uuid
+from collections.abc import Callable
 
-def is_json(value: str):
+def is_json(value: str) -> bool:
     try:
         json.loads(value)
     except ValueError as e:
         return False
     return True
 
-def contains(list, condition):
-    return any(condition(item) for item in list)
+def contains(items: list, condition: Callable) -> bool:
+    return any(condition(item) for item in items)
 
-def find(list, condition):
-    for item in list:
+def find(items: list, condition: Callable) -> object:
+    for item in items:
         if condition(item):
             return item
     return None
@@ -23,5 +24,5 @@ def generate_user_agent(
     app_ver: str = "6.16.5 (build 1)",
     account_id: str = "_",
     operator: str = "1"
-):
+) -> str:
     return f"{iphone} | {ios} | ntk | {app_ver} | {account_id} | {operator} | {str(uuid.uuid4()).upper()}"

custom_components/elektronny_gorod/lock.py

@@ -15,7 +15,7 @@
 )
 
 from .const import DOMAIN, LOGGER
-from .coordinator import ElektronnyGorogDataUpdateCoordinator
+from .coordinator import ElektronnyGorodUpdateCoordinator
 
 LOCK_UNLOCK_DELAY = 5  # Used to give a realistic lock/unlock experience in frontend
 LOCK_JAMMED_DELAY = 2
@@ -26,7 +26,7 @@ async def async_setup_entry(
     async_add_entities: AddEntitiesCallback,
 ) -> None:
     """Set up Elektronny Gorog Camera based on a config entry."""
-    coordinator: ElektronnyGorogDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
+    coordinator: ElektronnyGorodUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
 
     # Get cameras info
     locks_info = await coordinator.get_locks_info()
@@ -42,12 +42,12 @@ async def async_setup_entry(
 class ElektronnyGorogLock(LockEntity):
     def __init__(
         self,
-        coordinator: ElektronnyGorogDataUpdateCoordinator,
+        coordinator: ElektronnyGorodUpdateCoordinator,
         lock_info: dict
     ) -> None:
         LOGGER.info("ElektronnyGorogLock init %s", lock_info)
         super().__init__()
-        self._coordinator: ElektronnyGorogDataUpdateCoordinator = coordinator
+        self._coordinator: ElektronnyGorodUpdateCoordinator = coordinator
         self._lock_info: dict = lock_info
         self._place_id = self._lock_info["place_id"]
         self._access_control_id = self._lock_info["access_control_id"]