First try at supporting rain graph

custom_components/irm_kmi/camera.py

@@ -33,6 +33,7 @@ def __init__(self,
         """Initialize IrmKmiRadar component."""
         super().__init__(coordinator)
         Camera.__init__(self)
+        self.content_type = 'image/svg+xml'
         self._name = f"Radar {entry.title}"
         self._attr_unique_id = entry.entry_id
         self._attr_device_info = DeviceInfo(
@@ -42,18 +43,19 @@ def __init__(self,
             name=f"Radar {entry.title}"
         )
 
-        self._image_index = 0
+        self._image_index = False
 
     @property
     def frame_interval(self) -> float:
         """Return the interval between frames of the mjpeg stream."""
-        return 0.3
+        return 20
 
     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')
+        # TODO make it a still image to avoid cuts in playback on the dashboard
+        return self.coordinator.data.get('animation', {}).get('svg').encode()
 
     async def async_camera_image(
             self,
@@ -74,12 +76,12 @@ async def handle_async_mjpeg_stream(self, request: web.Request) -> web.StreamRes
 
     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)
-            self._image_index = (self._image_index + 1) % len(sequence)
-            return r
-        return None
+        # If this is not done this way, the live view can only be opened once
+        self._image_index = not self._image_index
+        if self._image_index:
+            return self.camera_image()
+        else:
+            return None
 
     @property
     def name(self) -> str:

custom_components/irm_kmi/coordinator.py

@@ -2,12 +2,10 @@
 import asyncio
 import logging
 from datetime import datetime, timedelta
-from io import BytesIO
 from typing import Any, List, Tuple
 
 import async_timeout
 import pytz
-from PIL import Image, ImageDraw, ImageFont
 from homeassistant.components.weather import Forecast
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import ATTR_LATITUDE, ATTR_LONGITUDE, CONF_ZONE
@@ -18,11 +16,13 @@
                                                       UpdateFailed)
 
 from .api import IrmKmiApiClient, IrmKmiApiError
-from .const import CONF_DARK_MODE, CONF_STYLE, OPTION_STYLE_SATELLITE
+from .const import CONF_DARK_MODE, CONF_STYLE
 from .const import IRM_KMI_TO_HA_CONDITION_MAP as CDT_MAP
-from .const import LANGS, OUT_OF_BENELUX, STYLE_TO_PARAM_MAP
+from .const import (LANGS, OPTION_STYLE_SATELLITE, OUT_OF_BENELUX,
+                    STYLE_TO_PARAM_MAP)
 from .data import (AnimationFrameData, CurrentWeatherData, IrmKmiForecast,
                    ProcessedCoordinatorData, RadarAnimationData)
+from .rain_graph import RainGraph
 from .utils import disable_from_config, get_config_value
 
 _LOGGER = logging.getLogger(__name__)
@@ -104,13 +104,17 @@ async def _async_animation_data(self, api_data: dict) -> RadarAnimationData:
             _LOGGER.warning(f"Could not get images for weather radar")
             return RadarAnimationData()
 
-        localisation = Image.open(BytesIO(images_from_api[0])).convert('RGBA')
+        localisation = images_from_api[0]
         images_from_api = images_from_api[1:]
 
-        radar_animation = await self.merge_frames_from_api(animation_data, country, images_from_api, localisation)
-
         lang = self.hass.config.language if self.hass.config.language in LANGS else 'en'
-        radar_animation['hint'] = api_data.get('animation', {}).get('sequenceHint', {}).get(lang)
+        radar_animation = RadarAnimationData(
+            hint=api_data.get('animation', {}).get('sequenceHint', {}).get(lang),
+            unit=api_data.get('animation', {}).get('unit', {}).get(lang),
+            location=localisation
+        )
+        svg_str = self.create_rain_graph(radar_animation, animation_data, country, images_from_api)
+        radar_animation['svg'] = svg_str
         return radar_animation
 
     async def process_api_data(self, api_data: dict) -> ProcessedCoordinatorData:
@@ -142,67 +146,6 @@ async def download_images_from_api(self,
         _LOGGER.debug(f"Just downloaded {len(images_from_api)} images")
         return images_from_api
 
-    async def merge_frames_from_api(self,
-                                    animation_data: List[dict],
-                                    country: str,
-                                    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
-        satellite_mode = self._style == OPTION_STYLE_SATELLITE
-
-        if country == 'NL':
-            background = Image.open("custom_components/irm_kmi/resources/nl.png").convert('RGBA')
-            fill_color = (0, 0, 0)
-        else:
-            image_path = (f"custom_components/irm_kmi/resources/be_"
-                          f"{'satellite' if satellite_mode else 'black' if self._dark_mode else 'white'}.png")
-            background = (Image.open(image_path).convert('RGBA'))
-            fill_color = (255, 255, 255) if self._dark_mode or satellite_mode else (0, 0, 0)
-
-        most_recent_frame = None
-        tz = pytz.timezone(self.hass.config.time_zone)
-        current_time = datetime.now(tz=tz)
-        sequence: List[AnimationFrameData] = list()
-
-        for (idx, sequence_element) in enumerate(animation_data):
-            frame = images_from_api[idx]
-            layer = Image.open(BytesIO(frame)).convert('RGBA')
-            temp = Image.alpha_composite(background, layer)
-            temp = Image.alpha_composite(temp, localisation_layer)
-
-            draw = ImageDraw.Draw(temp)
-            font = ImageFont.truetype("custom_components/irm_kmi/resources/roboto_medium.ttf", 16)
-            time_image = (datetime.fromisoformat(sequence_element.get('time'))
-                          .astimezone(tz=tz))
-
-            time_str = time_image.isoformat(sep=' ', timespec='minutes')
-
-            draw.text((4, 4), time_str, fill_color, font=font)
-
-            bytes_img = BytesIO()
-            temp.save(bytes_img, 'png', compress_level=8)
-
-            sequence.append(
-                AnimationFrameData(
-                    time=time_image,
-                    image=bytes_img.getvalue()
-                )
-            )
-
-            if most_recent_frame is None and current_time < time_image:
-                most_recent_frame = idx - 1 if idx > 0 else idx
-
-        background.close()
-        most_recent_frame = most_recent_frame if most_recent_frame is not None else -1
-
-        return RadarAnimationData(
-            sequence=sequence,
-            most_recent_image_idx=most_recent_frame
-        )
 
     @staticmethod
     def current_weather_from_data(api_data: dict) -> CurrentWeatherData:
@@ -351,3 +294,50 @@ def daily_list_to_forecast(data: List[dict] | None) -> List[Forecast] | None:
                 n_days += 1
 
         return forecasts
+
+    def create_rain_graph(self,
+                          radar_animation: RadarAnimationData,
+                          api_animation_data: List[dict],
+                          country: str,
+                          images_from_api: Tuple[bytes],
+                          ) -> str:
+
+        sequence: List[AnimationFrameData] = list()
+        tz = pytz.timezone(self.hass.config.time_zone)
+        current_time = datetime.now(tz=tz)
+        most_recent_frame = None
+
+        for idx, item in enumerate(api_animation_data):
+            frame = AnimationFrameData(
+                image=images_from_api[idx],
+                time=datetime.fromisoformat(item.get('time')) if item.get('time', None) is not None else None,
+                value=item.get('value', 0),
+                position=item.get('position', 0),
+                position_lower=item.get('positionLower', 0),
+                position_higher=item.get('positionHigher', 0)
+            )
+            sequence.append(frame)
+
+            if most_recent_frame is None and current_time < frame['time']:
+                most_recent_frame = idx - 1 if idx > 0 else idx
+
+        radar_animation['sequence'] = sequence
+        radar_animation['most_recent_image_idx'] = most_recent_frame
+
+        satellite_mode = self._style == OPTION_STYLE_SATELLITE
+
+        if country == 'NL':
+            image_path = "custom_components/irm_kmi/resources/nl.png"
+            bg_size = (640, 600)
+        else:
+            image_path = (f"custom_components/irm_kmi/resources/be_"
+                          f"{'satellite' if satellite_mode else 'black' if self._dark_mode else 'white'}.png")
+            bg_size = (640, 490)
+
+        svg_str = RainGraph(radar_animation, image_path, bg_size,
+                            dark_mode=self._dark_mode,
+                            # tz=self.hass.config.time_zone
+                            ).get_svg_string()
+
+        # TODO return value
+        return svg_str

custom_components/irm_kmi/data.py

@@ -24,6 +24,7 @@ class CurrentWeatherData(TypedDict, total=False):
     pressure: float | None
 
 
+# TODO cleanup useless fields
 class AnimationFrameData(TypedDict, total=False):
     """Holds one single frame of the radar camera, along with the timestamp of the frame"""
     time: datetime | None
@@ -33,6 +34,7 @@ class AnimationFrameData(TypedDict, total=False):
     position_higher: float | None
     position_lower: float | None
     rain_graph: bytes | None
+    merged_image: bytes | None
 
 
 class RadarAnimationData(TypedDict, total=False):
@@ -41,6 +43,8 @@ class RadarAnimationData(TypedDict, total=False):
     most_recent_image_idx: int | None
     hint: str | None
     unit: str | None
+    location: bytes | None
+    svg: str | None
 
 
 class ProcessedCoordinatorData(TypedDict, total=False):

custom_components/irm_kmi/manifest.json

@@ -8,6 +8,10 @@
   "integration_type": "service",
   "iot_class": "cloud_polling",
   "issue_tracker": "https://github.com/jdejaegh/irm-kmi-ha/issues",
-  "requirements": [],
+  "requirements": [
+    "Pillow==10.1.0",
+    "pytz==2023.3.post1",
+    "svgwrite==1.4.3"
+  ],
   "version": "0.1.6-beta"
 }