Add docstring and reformat files

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

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,36 +47,36 @@ 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(
             self,
             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

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)),
+            })
         )

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,

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,15 +93,19 @@ 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')),
             hourly_forecast=IrmKmiCoordinator.hourly_list_to_forecast(api_data.get('for', {}).get('hourly')),
             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
 

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