wip: feat: curve image

custom_components/boiler_controller/const.py

@@ -1,7 +1,7 @@
 DOMAIN = "boiler_controller"
 VERSION = "0.1.0"
 
-PLATFORMS = ["sensor", "select", "number", "button"]
+PLATFORMS = ["sensor", "select", "number", "button", "image"]
 
 # Configuration flow step IDs
 STEP_POWER_SENSOR = "power_sensor"

custom_components/boiler_controller/controller.py

@@ -23,8 +23,9 @@
     DIMMER_MODES,
 )
 from .shelly_client import ShellyClient
-from .calculator import Calculator
+from .calculator import Calculator, DEFAULT_CALIBRATION_PROFILE
 from .calibration import CalibrationStore, points_to_thresholds
+from .profile_image import ProfileImageManager
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -44,10 +45,14 @@ def __init__(self, hass: HomeAssistant, config_entry, integration_version: str |
         self._last_calculator_run = None
         self._shelly_status = None
         self._current_dimmer_percentage: int | None = None
+        self._active_plot_points: list[tuple[int, float]] = list(DEFAULT_CALIBRATION_PROFILE)
         self._dispatcher_signal = f"{DOMAIN}_{config_entry.entry_id}_shelly_status"
         self._mode_signal = f"{DOMAIN}_{config_entry.entry_id}_dimming_mode"
         self._manual_brightness_signal = f"{DOMAIN}_{config_entry.entry_id}_manual_brightness"
         self._calibration_signal = f"{DOMAIN}_{config_entry.entry_id}_calibration_state"
+        self._profile_image_signal = f"{DOMAIN}_{config_entry.entry_id}_profile_image"
+        self._profile_image_manager = ProfileImageManager(hass, config_entry.entry_id)
+        self._profile_image_updated_at = None
 
         # Configuration
         self.shelly_url = config_entry.data[CONF_SHELLY_URL]
@@ -406,6 +411,17 @@ def get_calibration_profile(self):
         """Return the active calibration profile, if any."""
         return self._calibration_profile
 
+    @property
+    def profile_image_manager(self) -> ProfileImageManager:
+        """Return the renderer that maintains the calibration curve image."""
+
+        return self._profile_image_manager
+
+    def get_active_plot_points(self) -> list[tuple[int, float]]:
+        """Return the latest calibration profile expressed as percentage/watt pairs."""
+
+        return list(self._active_plot_points)
+
     def get_shelly_status_signal(self):
         """Return dispatcher signal name for Shelly status updates."""
         return self._dispatcher_signal
@@ -422,6 +438,16 @@ def get_calibration_state_signal(self):
         """Dispatcher signal fired when calibration state changes."""
         return self._calibration_signal
 
+    def get_profile_image_signal(self):
+        """Dispatcher signal fired when the rendered profile image changes."""
+
+        return self._profile_image_signal
+
+    def get_profile_image_updated_at(self):
+        """Return timestamp of the last profile image refresh."""
+
+        return self._profile_image_updated_at
+
     @property
     def dimming_mode(self) -> str:
         return self._dimming_mode
@@ -548,7 +574,7 @@ async def async_run_calibration(
                     return None
 
                 profile = await self._calibration_store.async_save_points(measurements)
-                self._apply_calibration_profile(profile)
+                await self._apply_calibration_profile(profile)
                 detail_lines = [
                     f"  - {point['percentage']}% -> {point['watts']:.2f} W"
                     for point in measurements
@@ -798,20 +824,34 @@ async def _async_load_calibration_profile(self) -> None:
             _LOGGER.warning("Failed to load calibration profile: %s", err)
             profile = None
 
-        self._apply_calibration_profile(profile)
+        await self._apply_calibration_profile(profile)
         if profile:
             _LOGGER.info(
                 "Loaded calibration profile with %s points",
                 len(profile.get("points", [])),
             )
 
-    def _apply_calibration_profile(self, profile: dict | None) -> None:
+    async def _apply_calibration_profile(self, profile: dict | None) -> None:
         """Install the provided calibration profile or fall back to defaults."""
 
         self._calibration_profile = profile
         points = profile.get("points", []) if profile else []
         thresholds = points_to_thresholds(points)
+        plot_points = self._build_plot_points(thresholds)
         self._calculator.set_calibration_profile(thresholds if thresholds else None)
+        self._active_plot_points = plot_points
+        await self._profile_image_manager.async_update(plot_points)
+        self._profile_image_updated_at = dt_util.utcnow()
+        async_dispatcher_send(self.hass, self._profile_image_signal)
+
+    def _build_plot_points(self, thresholds: list[tuple[float, int]] | None) -> list[tuple[int, float]]:
+        if not thresholds:
+            return list(DEFAULT_CALIBRATION_PROFILE)
+        plot_points: list[tuple[int, float]] = []
+        for watts, percentage in thresholds:
+            plot_points.append((int(percentage), float(watts)))
+        plot_points.sort(key=lambda item: item[0])
+        return plot_points
 
     @staticmethod
     def _get_state_unit(state) -> str:

custom_components/boiler_controller/image.py

@@ -0,0 +1,71 @@
+"""Image entity exposing the calibration profile curve."""
+from __future__ import annotations
+
+from typing import Callable
+
+from homeassistant.components.image import ImageEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.util import dt as dt_util
+
+from .const import DOMAIN
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    controller = hass.data[DOMAIN][config_entry.entry_id]["controller"]
+    async_add_entities([BoilerControllerProfileImage(hass, controller, config_entry)])
+
+
+class BoilerControllerProfileImage(ImageEntity):
+    """Image entity showing the latest calibration curve."""
+
+    _attr_content_type = "image/svg+xml"
+    _attr_has_entity_name = True
+
+    def __init__(self, hass: HomeAssistant, controller, config_entry: ConfigEntry) -> None:
+        super().__init__(hass)
+        self._controller = controller
+        self._attr_unique_id = f"{config_entry.entry_id}_calibration_curve"
+        self._attr_name = "Calibration Curve"
+        self._manager = controller.profile_image_manager
+        self._attr_entity_picture_local = self._manager.local_url
+        self._attr_device_info = controller.device_info
+        self._attr_image_last_updated = controller.get_profile_image_updated_at()
+        self._unsub_dispatcher: Callable[[], None] | None = None
+
+    async def async_image(self) -> bytes | None:
+        data = await self._manager.async_get_bytes()
+        if data is None:
+            # Render the default curve if nothing exists yet.
+            await self._manager.async_update(self._controller.get_active_plot_points())
+            data = await self._manager.async_get_bytes()
+            if data is not None:
+                self._attr_image_last_updated = dt_util.utcnow()
+                self.async_write_ha_state()
+        return data
+
+    @property
+    def available(self) -> bool:
+        return True
+
+    async def async_added_to_hass(self) -> None:
+        await super().async_added_to_hass()
+
+        async def _handle_image_refresh() -> None:
+            self._attr_image_last_updated = dt_util.utcnow()
+            self.async_write_ha_state()
+
+        signal = self._controller.get_profile_image_signal()
+        self._unsub_dispatcher = async_dispatcher_connect(self.hass, signal, _handle_image_refresh)
+
+    async def async_will_remove_from_hass(self) -> None:
+        await super().async_will_remove_from_hass()
+        if self._unsub_dispatcher:
+            self._unsub_dispatcher()
+            self._unsub_dispatcher = None

custom_components/boiler_controller/profile_image.py

@@ -0,0 +1,116 @@
+"""Utilities for rendering calibration profile curves as SVG images."""
+from __future__ import annotations
+
+import asyncio
+import os
+from typing import Iterable
+
+from homeassistant.core import HomeAssistant
+
+SVG_WIDTH = 1200
+SVG_HEIGHT = 500
+SVG_PADDING = 60
+
+
+class ProfileImageManager:
+    """Generate and cache SVG curves for the calibration profile."""
+
+    def __init__(self, hass: HomeAssistant, entry_id: str) -> None:
+        self._hass = hass
+        self._entry_id = entry_id
+        self._lock = asyncio.Lock()
+        self._latest_svg: bytes | None = None
+        self._rel_path = os.path.join("boiler_controller", f"profile_{entry_id}.svg")
+
+    @property
+    def local_url(self) -> str:
+        """Return the /local/... path that hosts the cached image."""
+
+        return f"/local/{self._rel_path.replace(os.path.sep, '/')}"
+
+    async def async_get_bytes(self) -> bytes | None:
+        """Return the most recently rendered SVG bytes."""
+
+        async with self._lock:
+            if self._latest_svg is not None:
+                return self._latest_svg
+
+        path = self._absolute_path
+        if os.path.exists(path):
+            data = await self._hass.async_add_executor_job(self._read_file, path)
+            async with self._lock:
+                self._latest_svg = data
+            return data
+        return None
+
+    async def async_update(self, profile_points: Iterable[tuple[int, float]]) -> None:
+        """Render the provided calibration profile into an SVG image."""
+
+        svg_bytes = await self._hass.async_add_executor_job(
+            self._render_svg, list(profile_points)
+        )
+        async with self._lock:
+            self._latest_svg = svg_bytes
+        path = self._absolute_path
+        await self._hass.async_add_executor_job(self._write_file, path, svg_bytes)
+
+    @property
+    def _absolute_path(self) -> str:
+        return self._hass.config.path("www", self._rel_path)
+
+    @staticmethod
+    def _read_file(path: str) -> bytes:
+        with open(path, "rb") as handle:
+            return handle.read()
+
+    @staticmethod
+    def _write_file(path: str, data: bytes) -> None:
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "wb") as handle:
+            handle.write(data)
+
+    @staticmethod
+    def _render_svg(profile_points: list[tuple[int, float]]) -> bytes:
+        if not profile_points:
+            profile_points = [(0, 0.0)]
+
+        percentages = [float(point[0]) for point in profile_points]
+        watts_values = [float(point[1]) for point in profile_points]
+
+        min_pct = min(percentages)
+        max_pct = max(percentages)
+        pct_span = max(1.0, max_pct - min_pct)
+
+        max_watts = max(1.0, max(watts_values))
+        plot_width = SVG_WIDTH - 2 * SVG_PADDING
+        plot_height = SVG_HEIGHT - 2 * SVG_PADDING
+
+        def scale_x(value: float) -> float:
+            return SVG_PADDING + ((value - min_pct) / pct_span) * plot_width
+
+        def scale_y(value: float) -> float:
+            return SVG_HEIGHT - SVG_PADDING - (value / max_watts) * plot_height
+
+        polyline = " ".join(
+            f"{scale_x(pct):.2f},{scale_y(watts):.2f}"
+            for pct, watts in zip(percentages, watts_values)
+        )
+
+        y_axis = f"{SVG_PADDING},{SVG_PADDING} {SVG_PADDING},{SVG_HEIGHT - SVG_PADDING}"
+        x_axis = f"{SVG_PADDING},{SVG_HEIGHT - SVG_PADDING} {SVG_WIDTH - SVG_PADDING},{SVG_HEIGHT - SVG_PADDING}"
+
+        svg = f"""
+<svg xmlns=\"http://www.w3.org/2000/svg\" width=\"{SVG_WIDTH}\" height=\"{SVG_HEIGHT}\" viewBox=\"0 0 {SVG_WIDTH} {SVG_HEIGHT}\" role=\"img\">
+    <title>Boiler Controller Calibration Curve</title>
+    <rect width=\"100%\" height=\"100%\" fill=\"#ffffff\" />
+    <polyline fill=\"none\" stroke=\"#d3d3d3\" stroke-width=\"2\" points=\"{y_axis}\" />
+    <polyline fill=\"none\" stroke=\"#d3d3d3\" stroke-width=\"2\" points=\"{x_axis}\" />
+    <polyline fill=\"none\" stroke=\"#ff7f0e\" stroke-width=\"4\" stroke-linejoin=\"round\" stroke-linecap=\"round\" points=\"{polyline}\" />
+    <text x=\"{SVG_WIDTH / 2}\" y=\"{SVG_PADDING / 2}\" text-anchor=\"middle\" font-size=\"24\" font-family=\"sans-serif\">Calibration Curve</text>
+    <text x=\"{SVG_WIDTH / 2}\" y=\"{SVG_HEIGHT - SVG_PADDING / 4}\" text-anchor=\"middle\" font-size=\"18\" font-family=\"sans-serif\">Brightness (%)</text>
+    <g transform=\"rotate(-90)\">
+        <text x=\"{-SVG_HEIGHT / 2}\" y=\"{SVG_PADDING / 2}\" text-anchor=\"middle\" font-size=\"18\" font-family=\"sans-serif\">Watts</text>
+    </g>
+</svg>
+"""
+        return svg.encode("utf-8")