Coverage for custom_components/supernotify/switch.py: 100%
161 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-08 17:42 +0000
1"""Switch platform: enable or disable each scenario, recipient, delivery and transport.
3Forwarded to from async_setup_entry in __init__.py once the SupernotifyEngine (entry.
4runtime_data) is fully initialized.
6A scenario's binary_sensor (binary_sensor.py) reports whether its *conditions* currently hold, so
7it can't also be the control for enabling and disabling the scenario - writing its state used to do
8both, and the two meanings fought each other. Recipient, delivery and transport binary_sensors had
9no such conflict, but are treated the same way for consistency. These switches are the control,
10and the binary_sensors are now read-only and kept only for backward compatibility.
12A delivery and its transport each have their own switch, and their own flag: switching a
13transport off suppresses all of its deliveries, without changing the delivery switches.
15Each switch overrides the `enabled` value configured in YAML, and the override is kept across a
16restart or reload (RestoreEntity) for as long as that configured value - its own `enabled`, or for
17a delivery without one, its transport's - is unchanged. Editing it in YAML hands control back to
18the configuration.
19"""
21from __future__ import annotations
23import logging
24from dataclasses import dataclass
25from typing import TYPE_CHECKING, Any, Protocol
27from homeassistant.components.switch import SwitchEntity
28from homeassistant.const import EntityCategory
29from homeassistant.core import Context, callback
30from homeassistant.helpers.restore_state import ExtraStoredData, RestoreEntity
31from homeassistant.util import slugify
33from . import DOMAIN
34from .common import sanitize
35from .const import (
36 ATTR_OVERRIDDEN,
37 DELIVERY_UNRECORDED_ATTRIBUTES,
38 OVERRIDE_KIND_DELIVERY,
39 OVERRIDE_KIND_RECIPIENT,
40 OVERRIDE_KIND_SCENARIO,
41 OVERRIDE_KIND_TRANSPORT,
42 SUBENTRY_TYPE_TRANSPORT,
43 TRANSPORT_UNRECORDED_ATTRIBUTES,
44 transport_display_name,
45)
46from .hass_api import ha_device_info, ha_transport_device_info
48if TYPE_CHECKING:
49 from homeassistant.core import HomeAssistant
50 from homeassistant.helpers.device_registry import DeviceInfo
51 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
53 from . import SupernotifyConfigEntry
54 from .delivery import Delivery, DeliveryRegistry
55 from .people import PeopleRegistry, Recipient
56 from .scenario import Scenario, ScenarioRegistry
57 from .transport import Transport
59_LOGGER = logging.getLogger(__name__)
61# Entities are pushed, never polled, and their actions only change local state
62PARALLEL_UPDATES = 0
65async def async_setup_entry(
66 hass: HomeAssistant,
67 entry: SupernotifyConfigEntry,
68 async_add_entities: AddConfigEntryEntitiesCallback,
69) -> None:
70 """Add a switch for each scenario, recipient, and loaded transport and delivery."""
71 _ = hass
72 context = entry.runtime_data.context
73 switches = entry.runtime_data.override_switches
74 device_info = ha_device_info(entry.entry_id)
75 entities: list[SwitchEntity] = [
76 SupernotifyScenarioSwitch(scenario, context.scenario_registry, device_info, switches)
77 for scenario in context.scenario_registry.scenarios.values()
78 ]
79 # Only for what is loaded. The switch of a transport that isn't loaded this time, and of its
80 # deliveries, is left in the entity registry rather than removed: a transport can be missing
81 # just while what it depends on is still starting up
82 delivery_registry = context.delivery_registry
83 async_add_entities(entities)
84 # each transport's on its own service device in its subentry, along with its deliveries' -
85 # and each recipient's in its subentry, where it has one, with no device, so a recipient is
86 # only listed once in the integration
87 transport_subentries = {
88 s.unique_id: s.subentry_id for s in entry.subentries.values() if s.subentry_type == SUBENTRY_TYPE_TRANSPORT
89 }
90 for transport in delivery_registry.transports.values():
91 # named by its alias, imported from YAML, else its name - and renamed in the UI as any device is
92 transport_device = ha_transport_device_info(
93 entry.entry_id, transport.name, transport.alias or transport_display_name(transport.name)
94 )
95 transport_entities: list[SwitchEntity] = [
96 SupernotifyTransportSwitch(transport, delivery_registry, transport_device, switches)
97 ]
98 transport_entities.extend(
99 SupernotifyDeliverySwitch(delivery, delivery_registry, transport_device, switches)
100 for delivery in delivery_registry.deliveries.values()
101 if delivery.transport.name == transport.name
102 )
103 async_add_entities(transport_entities, config_subentry_id=transport_subentries.get(transport.name))
104 subentry_ids = context.people_registry.subentry_ids(entry.subentries)
105 for recipient in context.people_registry.people.values():
106 async_add_entities(
107 [SupernotifyRecipientSwitch(recipient, context.people_registry, switches)],
108 config_subentry_id=subentry_ids.get(recipient.key or ""),
109 )
112class Overridable(Protocol):
113 """Anything with a configured enabled flag that a switch can override at runtime."""
115 name: str
116 enabled: bool
117 config_enabled: bool
120@dataclass(frozen=True)
121class OverrideStoredData(ExtraStoredData):
122 """What a switch keeps across a restart or reload: its state, and the configured value it
123 was overriding, so that an override is dropped once the configuration changes."""
125 enabled: bool
126 config_enabled: bool
128 def as_dict(self) -> dict[str, Any]:
129 return {"enabled": self.enabled, "config_enabled": self.config_enabled}
131 @classmethod
132 def from_dict(cls, data: object) -> OverrideStoredData | None:
133 """None unless data holds both values, as real booleans."""
134 if isinstance(data, dict) and isinstance(data.get("enabled"), bool) and isinstance(data.get("config_enabled"), bool):
135 return cls(enabled=data["enabled"], config_enabled=data["config_enabled"])
136 return None
139class SupernotifyOverridableSwitch(SwitchEntity, RestoreEntity):
140 """A switch overriding the configured enabled flag of a scenario, recipient, delivery or
141 transport, and keeping that override across a restart or reload while the configured value -
142 its own `enabled`, or for a delivery without one, its transport's - is unchanged."""
144 _attr_has_entity_name = True
145 _attr_entity_category = EntityCategory.CONFIG
146 _attr_should_poll = False
148 def __init__(
149 self,
150 kind: str,
151 target: Overridable,
152 device_info: DeviceInfo | None,
153 switches: dict[str, SupernotifyOverridableSwitch],
154 ) -> None:
155 self._target = target
156 self._switches = switches
157 self._key = f"{kind}_{target.name}"
158 self._attr_unique_id = self._key
159 self._attr_device_info = device_info
161 @property
162 def is_on(self) -> bool:
163 return bool(self._target.enabled)
165 def _with_overridden(self, attributes: dict[str, object]) -> dict[str, object]:
166 """The switched item's own attributes, and whether the switch was changed at runtime - then
167 it no longer matches the configured `enabled`, and reset_overrides would put it back."""
168 return {**attributes, ATTR_OVERRIDDEN: bool(self._target.enabled) != bool(self._target.config_enabled)}
170 @property
171 def extra_restore_state_data(self) -> OverrideStoredData:
172 # kept trivial: some Home Assistant versions don't guard this getter against errors
173 return OverrideStoredData(enabled=bool(self._target.enabled), config_enabled=bool(self._target.config_enabled))
175 async def async_added_to_hass(self) -> None:
176 await super().async_added_to_hass()
177 self._switches[self._key] = self
178 last = await self.async_get_last_extra_data()
179 stored = OverrideStoredData.from_dict(last.as_dict()) if last else None
180 if stored is None:
181 return
182 if stored.config_enabled != self._target.config_enabled:
183 _LOGGER.info("SUPERNOTIFY Configuration of %s changed, discarding its previous override", self._key)
184 return
185 if stored.enabled != self._target.enabled:
186 _LOGGER.info("SUPERNOTIFY Restoring override of %s to %s", self._key, "on" if stored.enabled else "off")
187 # no state write here - Home Assistant writes the first state once this returns
188 self._apply_enabled(stored.enabled)
190 async def async_will_remove_from_hass(self) -> None:
191 self._switches.pop(self._key, None)
192 await super().async_will_remove_from_hass()
194 async def async_turn_on(self, **kwargs: Any) -> None:
195 self.async_set_enabled(True)
197 async def async_turn_off(self, **kwargs: Any) -> None:
198 self.async_set_enabled(False)
200 @callback
201 def async_set_enabled(self, enabled: bool, context: Context | None = None) -> bool:
202 """Change the flag and publish it, returning whether anything changed.
204 `context` is the action call that asked for the change, when that isn't this switch's own
205 turn_on/turn_off, for which Home Assistant has already set it."""
206 if enabled == self._target.enabled:
207 return False
208 if context is not None:
209 self.async_set_context(context)
210 self._apply_enabled(enabled)
211 self.async_write_ha_state()
212 return True
214 def _apply_enabled(self, enabled: bool) -> None:
215 self._target.enabled = enabled
216 self._refresh_related()
218 def _refresh_related(self) -> None:
219 """Re-publish any other entity whose state follows this flag."""
222class SupernotifyScenarioSwitch(SupernotifyOverridableSwitch):
223 """Whether a scenario is enabled, and so able to apply to notifications."""
225 _attr_translation_key = "scenario_enabled"
227 def __init__(
228 self,
229 scenario: Scenario,
230 registry: ScenarioRegistry,
231 device_info: DeviceInfo,
232 switches: dict[str, SupernotifyOverridableSwitch],
233 ) -> None:
234 super().__init__(OVERRIDE_KIND_SCENARIO, scenario, device_info, switches)
235 self._scenario = scenario
236 self._registry = registry
237 self._attr_translation_placeholders = {"scenario": scenario.alias or scenario.name}
238 # Setting entity_id directly is not preferred Home Assistant practice - entities should
239 # leave it to be derived from the device and entity name. It's done here so that a new
240 # install gets the entity_id documented for scenarios (and used by the binary_sensor
241 # this switch sits beside), which HA's own derivation would not produce. It has no effect
242 # on an existing install: the entity registry entry found by unique_id always wins, so
243 # nobody's entity_id, including one they have renamed, is changed by this.
244 self.entity_id = f"switch.{DOMAIN}_scenario_{scenario.name}"
246 @property
247 def extra_state_attributes(self) -> dict[str, object]:
248 return self._with_overridden(sanitize(self._scenario.attributes(include_condition=False)))
250 def _refresh_related(self) -> None:
251 # the condition state of a disabled scenario is always off, so that changes with this
252 self._registry.async_refresh_entity(self._scenario.name, self._context)
255class SupernotifyRecipientSwitch(SupernotifyOverridableSwitch):
256 """Whether a recipient is enabled, and so able to be notified."""
258 _attr_translation_key = "recipient_enabled"
260 def __init__(
261 self,
262 recipient: Recipient,
263 registry: PeopleRegistry,
264 switches: dict[str, SupernotifyOverridableSwitch],
265 ) -> None:
266 super().__init__(OVERRIDE_KIND_RECIPIENT, recipient, None, switches)
267 self._attr_translation_placeholders = {"recipient": recipient.alias or recipient.name}
268 self._recipient = recipient
269 self._registry = registry
270 # Setting entity_id directly is not preferred Home Assistant practice - see
271 # SupernotifyScenarioSwitch for why it's done anyway, and why it can't change the
272 # entity_id of an existing install.
273 self.entity_id = f"switch.{DOMAIN}_recipient_{recipient.name}"
275 @property
276 def extra_state_attributes(self) -> dict[str, object]:
277 return self._with_overridden(sanitize(self._recipient.attributes()))
279 def _refresh_related(self) -> None:
280 # the deprecated binary_sensor mirrors enabled
281 self._registry.async_refresh_entity(self._recipient.name)
284class SupernotifyDeliverySwitch(SupernotifyOverridableSwitch):
285 """Whether a delivery is enabled, and so able to be used for notifications."""
287 _attr_translation_key = "delivery_enabled"
288 _unrecorded_attributes = DELIVERY_UNRECORDED_ATTRIBUTES
290 def __init__(
291 self,
292 delivery: Delivery,
293 registry: DeliveryRegistry,
294 device_info: DeviceInfo | None,
295 switches: dict[str, SupernotifyOverridableSwitch],
296 ) -> None:
297 super().__init__(OVERRIDE_KIND_DELIVERY, delivery, device_info, switches)
298 self._delivery = delivery
299 self._registry = registry
300 self._attr_translation_placeholders = {"delivery": delivery.alias or delivery.name}
301 # Setting entity_id directly is not preferred Home Assistant practice - see
302 # SupernotifyScenarioSwitch for why it's done anyway. Slugified, as a delivery name can be
303 # any string.
304 self.entity_id = f"switch.{DOMAIN}_delivery_{slugify(delivery.name)}"
306 @property
307 def extra_state_attributes(self) -> dict[str, object]:
308 return self._with_overridden(sanitize(self._delivery.attributes()))
310 def _refresh_related(self) -> None:
311 # the deprecated binary_sensor mirrors enabled
312 self._registry.async_refresh_entity(self._key)
315class SupernotifyTransportSwitch(SupernotifyOverridableSwitch):
316 """Whether a transport is enabled - when off, none of its deliveries are used."""
318 _attr_translation_key = "transport_enabled"
319 _unrecorded_attributes = TRANSPORT_UNRECORDED_ATTRIBUTES
321 def __init__(
322 self,
323 transport: Transport,
324 registry: DeliveryRegistry,
325 device_info: DeviceInfo | None,
326 switches: dict[str, SupernotifyOverridableSwitch],
327 ) -> None:
328 super().__init__(OVERRIDE_KIND_TRANSPORT, transport, device_info, switches)
329 self._transport = transport
330 self._registry = registry
331 # named after its device, the transport's own
332 # Setting entity_id directly is not preferred Home Assistant practice - see
333 # SupernotifyScenarioSwitch and SupernotifyDeliverySwitch.
334 self.entity_id = f"switch.{DOMAIN}_transport_{slugify(transport.name)}"
336 @property
337 def extra_state_attributes(self) -> dict[str, object]:
338 return self._with_overridden(sanitize(self._transport.attributes()))
340 def _refresh_related(self) -> None:
341 # the deprecated binary_sensor mirrors enabled
342 self._registry.async_refresh_entity(self._key)
343 # and each of this transport's deliveries shows it, as transport_enabled
344 for delivery in self._registry.deliveries.values():
345 if delivery.transport is not self._transport:
346 continue
347 key = f"{OVERRIDE_KIND_DELIVERY}_{delivery.name}"
348 switch = self._switches.get(key)
349 if switch is not None:
350 if self._context is not None:
351 switch.async_set_context(self._context)
352 switch.async_write_ha_state()
353 self._registry.async_refresh_entity(key)