Coverage for custom_components/supernotify/transports/kodi.py: 97%
113 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-01 18:25 +0000
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-01 18:25 +0000
1"""Kodi transport for SuperNotify.
3Shows on-screen overlay notifications on Kodi media centers using Home
4Assistant's `kodi` integration, calling the entity-based `kodi.call_method`
5service with the JSON-RPC method `GUI.ShowNotification`.
7Supported data keys (all optional):
8 kodi_displaytime int overlay duration in milliseconds
9 (default: 10000). Kodi enforces a minimum
10 of 1500 ms, lower values are clamped.
11 kodi_icon str "info" | "warning" | "error" or an image
12 URL. Overrides the priority-derived icon.
13 kodi_attach_image bool use the camera snapshot as the notification
14 icon, resolved to a URL reachable from the
15 Kodi host (default: False). When an image
16 URL is resolved it wins over kodi_icon.
18Priority to native icon mapping (when kodi_icon is not set):
19 critical -> "error", high -> "warning",
20 medium / low / minimum -> "info"
22Notes on the `kodi.call_method` service:
23- The service is entity-based: Kodi instances are `media_player` entities
24 created by the kodi config entry. Targets are pre-filtered here to
25 `media_player.*` entity ids and passed via target_data; other targets
26 are dropped with a debug log.
27- `GUI.ShowNotification` requires a non-empty `title`: when the envelope
28 has no title, a "Notification" fallback is used.
29- Kodi runs on a remote host, so the snapshot image must be a URL that the
30 Kodi box can fetch, NEVER a local HA filesystem path. Resolution order:
31 1. `envelope.media[snapshot_url]` (absolutised against the HA base URL)
32 2. `envelope.grab_image()` + `media_storage.object_url()` (already an
33 absolute URL served by the HA web server)
34 The HA Internal URL should be a direct IP (e.g. http://192.168.0.123:8123)
35 rather than an mDNS `.local` hostname, which some players cannot resolve.
36- No residual data passthrough: any extra key in the payload is forwarded
37 by `kodi.call_method` as a JSON-RPC parameter and makes the whole
38 `GUI.ShowNotification` call fail on the Kodi side. Residual generic data
39 keys are therefore dropped with a debug log, unlike the standard
40 transport pattern.
41- Pure overlay: no action buttons and no user interaction.
43Internal data keys filtered upstream by notification.py and NOT popped here:
44 force_resend, spoken_message
46References:
47- Kodi integration: https://www.home-assistant.io/integrations/kodi/
48- JSON-RPC GUI.ShowNotification: https://kodi.wiki/view/JSON-RPC_API
50"""
52from __future__ import annotations
54import logging
55import re
56import urllib.parse
57from typing import TYPE_CHECKING, Any
59from homeassistant.const import ATTR_ENTITY_ID
61from custom_components.supernotify.common import boolify
62from custom_components.supernotify.const import (
63 ATTR_MEDIA_SNAPSHOT_URL,
64 OPTION_TARGET_CATEGORIES,
65 OPTION_TARGET_SELECT,
66 TRANSPORT_KODI,
67)
68from custom_components.supernotify.model import DebugTrace, TargetRequired, TransportConfig, TransportFeature
69from custom_components.supernotify.transport import Transport
71if TYPE_CHECKING:
72 from custom_components.supernotify.envelope import Envelope
74_LOGGER = logging.getLogger(__name__)
76RE_VALID_KODI = r"media_player\.[A-Za-z0-9_]+"
78_KODI_ENTITY_RE = re.compile(rf"^{RE_VALID_KODI}$")
80# GUI.ShowNotification displaytime constraints (milliseconds)
81DEFAULT_DISPLAYTIME = 10000
82MIN_DISPLAYTIME = 1500
84# GUI.ShowNotification requires a non-empty title
85DEFAULT_TITLE = "Notification"
87# SuperNotify priority -> Kodi native notification icon
88_PRIORITY_ICON = {
89 "critical": "error",
90 "high": "warning",
91 "medium": "info",
92 "low": "info",
93 "minimum": "info",
94}
95_DEFAULT_ICON = "info"
98def _coerce_int(value: Any) -> int | None:
99 """Best-effort int coercion. Returns None on failure."""
100 if value is None:
101 return None
102 try:
103 return int(value)
104 except (TypeError, ValueError):
105 try:
106 return int(float(value))
107 except (TypeError, ValueError):
108 return None
111class KodiTransport(Transport):
112 """Notify via Kodi on-screen overlay using the kodi integration."""
114 def __init__(self, *args: Any, **kwargs: Any) -> None:
115 super().__init__(*args, **kwargs)
117 name = TRANSPORT_KODI
119 @property
120 def supported_features(self) -> TransportFeature:
121 return TransportFeature.MESSAGE | TransportFeature.TITLE | TransportFeature.IMAGES | TransportFeature.SNAPSHOT_IMAGE
123 @property
124 def default_config(self) -> TransportConfig:
125 config = TransportConfig()
126 config.delivery_defaults.action = "kodi.call_method"
127 config.delivery_defaults.target_required = TargetRequired.ALWAYS
128 config.delivery_defaults.options = {
129 OPTION_TARGET_SELECT: [RE_VALID_KODI],
130 OPTION_TARGET_CATEGORIES: [ATTR_ENTITY_ID],
131 }
132 return config
134 def validate_action(self, action: str | None) -> bool:
135 """Validate that action is the kodi call_method service."""
136 return action == "kodi.call_method"
138 def select_targets(self, envelope: Envelope) -> list[str]:
139 """Filter envelope targets down to media_player entity ids.
141 `kodi.call_method` only accepts media_player entities; anything else
142 is dropped with a debug log. Duplicates are removed preserving order.
143 """
144 raw_targets: list[str] = envelope.target.entity_ids or [] if envelope.target else []
145 targets: list[str] = []
146 for target in raw_targets:
147 if isinstance(target, str) and _KODI_ENTITY_RE.match(target):
148 if target not in targets:
149 targets.append(target)
150 else:
151 _LOGGER.debug("SUPERNOTIFY kodi: skipping invalid target %r", target)
152 return targets
154 def _absolute_url(self, url: str) -> str:
155 """Make a relative URL absolute so the Kodi host can fetch it.
157 Prefers the HA internal URL since Kodi is typically a LAN device;
158 falls back to the external URL. Absolute URLs pass through as-is.
159 """
160 if not url or url.startswith(("http://", "https://")):
161 return url
162 base = self.hass_api.internal_url or self.hass_api.external_url
163 if not base:
164 _LOGGER.warning("SUPERNOTIFY kodi: no base url to absolutise %s", url)
165 return url
166 if not url.startswith("/"):
167 url = "/" + url
168 return urllib.parse.urljoin(base, url)
170 async def _resolve_image_url(self, envelope: Envelope) -> str | None:
171 """Resolve a snapshot image URL reachable from the Kodi host.
173 Order of resolution:
174 1. envelope.media snapshot_url (absolutised against HA base URL)
175 2. envelope.grab_image() + media_storage.object_url() (served by
176 the HA web server via the registered media path)
177 3. None (caller keeps the priority/override icon)
178 """
179 snapshot_url = envelope.media.get(ATTR_MEDIA_SNAPSHOT_URL) if envelope.media else None
180 if snapshot_url:
181 return self._absolute_url(str(snapshot_url))
183 image_path = None
184 try:
185 image_path = await envelope.grab_image()
186 except Exception as e:
187 _LOGGER.warning("SUPERNOTIFY kodi: failed to grab image: %s", e)
188 if image_path:
189 try:
190 object_url = await self.context.media_storage.object_url(image_path)
191 except Exception as e:
192 _LOGGER.debug("SUPERNOTIFY kodi: object_url failed for %s: %s", image_path, e)
193 object_url = None
194 if object_url:
195 return object_url
196 _LOGGER.debug("SUPERNOTIFY kodi: no shareable URL for %s, image skipped", image_path)
197 return None
199 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool:
200 _LOGGER.debug("SUPERNOTIFY kodi %s", envelope.message)
202 raw_data: dict[str, Any] = dict(envelope.data) if envelope.data else {}
204 # Pop Kodi-specific data keys
205 displaytime_raw = raw_data.pop("kodi_displaytime", None)
206 icon_override = raw_data.pop("kodi_icon", None)
207 attach_image = boolify(raw_data.pop("kodi_attach_image", False), default=False)
209 # Resolve and pre-validate media_player targets
210 targets = self.select_targets(envelope)
211 if not targets:
212 _LOGGER.warning("SUPERNOTIFY kodi: no valid media_player targets")
213 self.record_error("no valid Kodi media_player targets", "deliver")
214 return False
216 # Coerce displaytime and clamp to the Kodi minimum
217 displaytime = _coerce_int(displaytime_raw)
218 if displaytime is None:
219 if displaytime_raw is not None:
220 _LOGGER.warning(
221 "SUPERNOTIFY kodi: invalid kodi_displaytime %r, using default %d ms",
222 displaytime_raw,
223 DEFAULT_DISPLAYTIME,
224 )
225 displaytime = DEFAULT_DISPLAYTIME
226 if displaytime < MIN_DISPLAYTIME:
227 _LOGGER.debug(
228 "SUPERNOTIFY kodi: kodi_displaytime %d below Kodi minimum, clamping to %d ms",
229 displaytime,
230 MIN_DISPLAYTIME,
231 )
232 displaytime = MIN_DISPLAYTIME
234 # Icon: priority-derived default, then kodi_icon override, then
235 # snapshot image URL (which wins when resolvable)
236 icon: str = _PRIORITY_ICON.get(envelope.priority or "medium", _DEFAULT_ICON)
237 if icon_override:
238 icon = str(icon_override)
239 if attach_image:
240 image_url = await self._resolve_image_url(envelope)
241 if image_url:
242 icon = image_url
243 else:
244 _LOGGER.debug("SUPERNOTIFY kodi: no image URL available, keeping icon %r", icon)
246 # Build the JSON-RPC payload. Every key in action_data beyond
247 # `method` is forwarded as a GUI.ShowNotification parameter, so
248 # residual generic data keys are NOT merged: an unknown parameter
249 # fails the whole JSON-RPC call on the Kodi side.
250 action_data: dict[str, Any] = {
251 "method": "GUI.ShowNotification",
252 "title": envelope.title or DEFAULT_TITLE,
253 "message": envelope.message or "",
254 "image": icon,
255 "displaytime": displaytime,
256 }
258 if raw_data:
259 _LOGGER.debug(
260 "SUPERNOTIFY kodi: dropping data keys not supported by GUI.ShowNotification: %s",
261 sorted(raw_data),
262 )
264 return await self.call_action(
265 envelope,
266 action_data=action_data,
267 target_data={ATTR_ENTITY_ID: targets},
268 )