From 88f889757acb52db084221c368d9c137b7313607 Mon Sep 17 00:00:00 2001 From: Jules Dejaeghere Date: Thu, 28 Dec 2023 21:37:51 +0100 Subject: [PATCH] Add docstring and reformat files --- custom_components/irm_kmi/api.py | 28 ++++++++---------------- custom_components/irm_kmi/camera.py | 27 +++++++++++------------ custom_components/irm_kmi/config_flow.py | 14 +++++------- custom_components/irm_kmi/const.py | 2 +- custom_components/irm_kmi/coordinator.py | 23 +++++++++++++------ custom_components/irm_kmi/data.py | 4 ++++ 6 files changed, 48 insertions(+), 50 deletions(-) diff --git a/custom_components/irm_kmi/api.py b/custom_components/irm_kmi/api.py index 988d4a6..c1a360e 100644 --- a/custom_components/irm_kmi/api.py +++ b/custom_components/irm_kmi/api.py @@ -18,33 +18,28 @@ class IrmKmiApiError(Exception): """Exception to indicate a general API error.""" -class IrmKmiApiCommunicationError( - IrmKmiApiError -): +class IrmKmiApiCommunicationError(IrmKmiApiError): """Exception to indicate a communication error.""" -class IrmKmiApiParametersError( - IrmKmiApiError -): +class IrmKmiApiParametersError(IrmKmiApiError): """Exception to indicate a parameter error.""" -def _api_key(method_name: str): +def _api_key(method_name: str) -> str: """Get API key.""" return hashlib.md5(f"r9EnW374jkJ9acc;{method_name};{datetime.now().strftime('%d/%m/%Y')}".encode()).hexdigest() class IrmKmiApiClient: - """Sample API Client.""" + """API client for IRM KMI weather data""" COORD_DECIMALS = 6 def __init__(self, session: aiohttp.ClientSession) -> None: - """Sample API Client.""" self._session = session self._base_url = "https://app.meteo.be/services/appv4/" - async def get_forecasts_coord(self, coord: dict) -> any: + async def get_forecasts_coord(self, coord: dict) -> dict: """Get forecasts for given city.""" assert 'lat' in coord assert 'long' in coord @@ -55,6 +50,7 @@ async def get_forecasts_coord(self, coord: dict) -> any: return await response.json() async def get_image(self, url, params: dict | None = None) -> bytes: + """Get the image at the specified url with the parameters""" # TODO support etag and head request before requesting content r: ClientResponse = await self._api_wrapper(base_url=url, params={} if params is None else params) return await r.read() @@ -83,14 +79,8 @@ async def _api_wrapper( return response except asyncio.TimeoutError as exception: - raise IrmKmiApiCommunicationError( - "Timeout error fetching information", - ) from exception + raise IrmKmiApiCommunicationError("Timeout error fetching information") from exception except (aiohttp.ClientError, socket.gaierror) as exception: - raise IrmKmiApiCommunicationError( - "Error fetching information", - ) from exception + raise IrmKmiApiCommunicationError("Error fetching information") from exception except Exception as exception: # pylint: disable=broad-except - raise IrmKmiApiError( - f"Something really wrong happened! {exception}" - ) from exception + raise IrmKmiApiError(f"Something really wrong happened! {exception}") from exception diff --git a/custom_components/irm_kmi/camera.py b/custom_components/irm_kmi/camera.py index 26f81ff..d2c3b5e 100644 --- a/custom_components/irm_kmi/camera.py +++ b/custom_components/irm_kmi/camera.py @@ -2,6 +2,7 @@ import logging +from aiohttp import web from homeassistant.components.camera import Camera, async_get_still_stream from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant @@ -20,19 +21,17 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_e _LOGGER.debug(f'async_setup_entry entry is: {entry}') coordinator = hass.data[DOMAIN][entry.entry_id] - async_add_entities( - [IrmKmiRadar(coordinator, entry)] - ) + async_add_entities([IrmKmiRadar(coordinator, entry)]) class IrmKmiRadar(CoordinatorEntity, Camera): - """Representation of a local file camera.""" + """Representation of a radar view camera.""" def __init__(self, coordinator: IrmKmiCoordinator, entry: ConfigEntry, ) -> None: - """Initialize Local File Camera component.""" + """Initialize IrmKmiRadar component.""" super().__init__(coordinator) Camera.__init__(self) self._name = f"Radar {entry.title}" @@ -48,12 +47,13 @@ def __init__(self, @property def frame_interval(self) -> float: - """Return the interval between frames of the mjpeg stream""" + """Return the interval between frames of the mjpeg stream.""" return 0.3 def camera_image(self, width: int | None = None, height: int | None = None) -> bytes | None: + """Return still image to be used as thumbnail.""" return self.coordinator.data.get('animation', {}).get('most_recent_image') async def async_camera_image( @@ -61,23 +61,22 @@ async def async_camera_image( width: int | None = None, height: int | None = None ) -> bytes | None: - """Return bytes of camera image.""" + """Return still image to be used as thumbnail.""" return self.camera_image() - async def handle_async_still_stream(self, request, interval): + async def handle_async_still_stream(self, request: web.Request, interval: float) -> web.StreamResponse: """Generate an HTTP MJPEG stream from camera images.""" _LOGGER.info("handle_async_still_stream") self._image_index = 0 - return await async_get_still_stream( - request, self.iterate, self.content_type, interval - ) + return await async_get_still_stream(request, self.iterate, self.content_type, interval) - async def handle_async_mjpeg_stream(self, request): + async def handle_async_mjpeg_stream(self, request: web.Request) -> web.StreamResponse: """Serve an HTTP MJPEG stream from the camera.""" _LOGGER.info("handle_async_mjpeg_stream") return await self.handle_async_still_stream(request, self.frame_interval) async def iterate(self) -> bytes | None: + """Loop over all the frames when called multiple times.""" sequence = self.coordinator.data.get('animation', {}).get('sequence') if isinstance(sequence, list) and len(sequence) > 0: r = sequence[self._image_index].get('image', None) @@ -86,12 +85,12 @@ async def iterate(self) -> bytes | None: return None @property - def name(self): + def name(self) -> str: """Return the name of this camera.""" return self._name @property - def extra_state_attributes(self): + def extra_state_attributes(self) -> dict: """Return the camera state attributes.""" attrs = {"hint": self.coordinator.data.get('animation', {}).get('hint')} return attrs diff --git a/custom_components/irm_kmi/config_flow.py b/custom_components/irm_kmi/config_flow.py index 0f44dc6..9320ab8 100644 --- a/custom_components/irm_kmi/config_flow.py +++ b/custom_components/irm_kmi/config_flow.py @@ -1,4 +1,4 @@ -"""Config flow to set up IRM KMI integration via the UI""" +"""Config flow to set up IRM KMI integration via the UI.""" import logging import voluptuous as vol @@ -17,7 +17,7 @@ class IrmKmiConfigFlow(ConfigFlow, domain=DOMAIN): VERSION = 1 async def async_step_user(self, user_input: dict | None = None) -> FlowResult: - + """Define the user step of the configuration flow.""" if user_input is not None: _LOGGER.debug(f"Provided config user is: {user_input}") @@ -32,11 +32,7 @@ async def async_step_user(self, user_input: dict | None = None) -> FlowResult: return self.async_show_form( step_id="user", - data_schema=vol.Schema( - { - vol.Required(CONF_ZONE): EntitySelector( - EntitySelectorConfig(domain=ZONE_DOMAIN), - ), - } - ), + data_schema=vol.Schema({ + vol.Required(CONF_ZONE): EntitySelector(EntitySelectorConfig(domain=ZONE_DOMAIN)), + }) ) diff --git a/custom_components/irm_kmi/const.py b/custom_components/irm_kmi/const.py index 3107c59..f5d8771 100644 --- a/custom_components/irm_kmi/const.py +++ b/custom_components/irm_kmi/const.py @@ -1,4 +1,4 @@ -"""Constants for the IRM KMI integration""" +"""Constants for the IRM KMI integration.""" from homeassistant.components.weather import (ATTR_CONDITION_CLEAR_NIGHT, ATTR_CONDITION_CLOUDY, diff --git a/custom_components/irm_kmi/coordinator.py b/custom_components/irm_kmi/coordinator.py index d6a2a96..3dad16d 100644 --- a/custom_components/irm_kmi/coordinator.py +++ b/custom_components/irm_kmi/coordinator.py @@ -3,7 +3,7 @@ import logging from datetime import datetime, timedelta from io import BytesIO -from typing import List, Tuple +from typing import Any, List, Tuple import async_timeout import pytz @@ -29,7 +29,7 @@ class IrmKmiCoordinator(DataUpdateCoordinator): """Coordinator to update data from IRM KMI""" def __init__(self, hass: HomeAssistant, zone: Zone): - """Initialize my coordinator.""" + """Initialize the coordinator.""" super().__init__( hass, _LOGGER, @@ -69,7 +69,8 @@ async def _async_update_data(self) -> ProcessedCoordinatorData: return await self.process_api_data(api_data) async def _async_animation_data(self, api_data: dict) -> RadarAnimationData: - + """From the API data passed in, call the API to get all the images and create the radar animation data object. + Frames from the API are merged with the background map and the location marker to create each frame.""" animation_data = api_data.get('animation', {}).get('sequence') localisation_layer_url = api_data.get('animation', {}).get('localisationLayer') country = api_data.get('country', '') @@ -92,7 +93,7 @@ async def _async_animation_data(self, api_data: dict) -> RadarAnimationData: return radar_animation async def process_api_data(self, api_data: dict) -> ProcessedCoordinatorData: - + """From the API data, create the object that will be used in the entities""" return ProcessedCoordinatorData( current_weather=IrmKmiCoordinator.current_weather_from_data(api_data), daily_forecast=IrmKmiCoordinator.daily_list_to_forecast(api_data.get('for', {}).get('daily')), @@ -100,7 +101,11 @@ async def process_api_data(self, api_data: dict) -> ProcessedCoordinatorData: animation=await self._async_animation_data(api_data=api_data) ) - async def download_images_from_api(self, animation_data, country, localisation_layer_url): + async def download_images_from_api(self, + animation_data: dict, + country: str, + localisation_layer_url: str) -> tuple[Any]: + """Download a batch of images to create the radar frames.""" coroutines = list() coroutines.append( self._api_client.get_image(localisation_layer_url, @@ -110,7 +115,7 @@ async def download_images_from_api(self, animation_data, country, localisation_l if frame.get('uri', None) is not None: coroutines.append(self._api_client.get_image(frame.get('uri'))) async with async_timeout.timeout(20): - images_from_api = await asyncio.gather(*coroutines, return_exceptions=True) + images_from_api = await asyncio.gather(*coroutines) _LOGGER.debug(f"Just downloaded {len(images_from_api)} images") return images_from_api @@ -121,7 +126,8 @@ async def merge_frames_from_api(self, images_from_api: Tuple[bytes], localisation_layer: Image ) -> RadarAnimationData: - + """Merge three layers to create one frame of the radar: the basemap, the clouds and the location marker. + Adds text in the top right to specify the timestamp of each image.""" background: Image fill_color: tuple @@ -177,6 +183,7 @@ async def merge_frames_from_api(self, @staticmethod def current_weather_from_data(api_data: dict) -> CurrentWeatherData: + """Parse the API data to build a CurrentWeatherData.""" # Process data to get current hour forecast now_hourly = None hourly_forecast_data = api_data.get('for', {}).get('hourly') @@ -234,6 +241,7 @@ def current_weather_from_data(api_data: dict) -> CurrentWeatherData: @staticmethod def hourly_list_to_forecast(data: List[dict] | None) -> List[Forecast] | None: + """Parse data from the API to create a list of hourly forecasts""" if data is None or not isinstance(data, list) or len(data) == 0: return None @@ -276,6 +284,7 @@ def hourly_list_to_forecast(data: List[dict] | None) -> List[Forecast] | None: @staticmethod def daily_list_to_forecast(data: List[dict] | None) -> List[Forecast] | None: + """Parse data from the API to create a list of daily forecasts""" if data is None or not isinstance(data, list) or len(data) == 0: return None diff --git a/custom_components/irm_kmi/data.py b/custom_components/irm_kmi/data.py index 4868787..539124d 100644 --- a/custom_components/irm_kmi/data.py +++ b/custom_components/irm_kmi/data.py @@ -14,6 +14,7 @@ class IrmKmiForecast(Forecast): class CurrentWeatherData(TypedDict, total=False): + """Class to hold the currently observable weather at a given location""" condition: str | None temperature: float | None wind_speed: float | None @@ -24,17 +25,20 @@ class CurrentWeatherData(TypedDict, total=False): class AnimationFrameData(TypedDict, total=False): + """Holds one single frame of the radar camera, along with the timestamp of the frame""" time: datetime | None image: bytes | None class RadarAnimationData(TypedDict, total=False): + """Holds frames and additional data for the animation to be rendered""" sequence: List[AnimationFrameData] | None most_recent_image: bytes | None hint: str | None class ProcessedCoordinatorData(TypedDict, total=False): + """Data class that will be exposed to the entities consuming data from an IrmKmiCoordinator""" current_weather: CurrentWeatherData hourly_forecast: List[Forecast] | None daily_forecast: List[IrmKmiForecast] | None