diff --git a/custom_components/elektronny_gorod/__init__.py b/custom_components/elektronny_gorod/__init__.py index e06be05..a6f6be9 100644 --- a/custom_components/elektronny_gorod/__init__.py +++ b/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 diff --git a/custom_components/elektronny_gorod/api.py b/custom_components/elektronny_gorod/api.py index 56a4fff..485e5f5 100644 --- a/custom_components/elektronny_gorod/api.py +++ b/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() diff --git a/custom_components/elektronny_gorod/camera.py b/custom_components/elektronny_gorod/camera.py index 9a22133..0c56bca 100644 --- a/custom_components/elektronny_gorod/camera.py +++ b/custom_components/elektronny_gorod/camera.py @@ -4,7 +4,7 @@ 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, @@ -12,7 +12,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 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"] diff --git a/custom_components/elektronny_gorod/const.py b/custom_components/elektronny_gorod/const.py index 73d3d86..c47b31c 100644 --- a/custom_components/elektronny_gorod/const.py +++ b/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" diff --git a/custom_components/elektronny_gorod/coordinator.py b/custom_components/elektronny_gorod/coordinator.py index 4699b61..cb26d47 100644 --- a/custom_components/elektronny_gorod/coordinator.py +++ b/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: diff --git a/custom_components/elektronny_gorod/helpers.py b/custom_components/elektronny_gorod/helpers.py index 91b2bdd..f97774c 100644 --- a/custom_components/elektronny_gorod/helpers.py +++ b/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()}" \ No newline at end of file diff --git a/custom_components/elektronny_gorod/lock.py b/custom_components/elektronny_gorod/lock.py index 348443e..a417384 100644 --- a/custom_components/elektronny_gorod/lock.py +++ b/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"]