Coverage for custom_components/supernotify/binary_sensor.py: 100%

169 statements  

« prev     ^ index     » next       coverage.py v7.16.2, created at 2026-11-08 17:42 +0000

1"""Binary sensor platform: scenario and recipient state as real Home Assistant entities. 

2 

3Forwarded to from async_setup_entry in __init__.py once the SupernotifyEngine (entry. 

4runtime_data) is fully initialized - scenario_registry.scenarios and people_registry.people are 

5already populated by then, mirroring notify.py's own async_setup_entry for recipient notify 

6entities. 

7 

8These replace the raw entity registry and hass.states.async_set() writes previously used 

9for these binary_sensors (see upstream issue #175, "Part B"): real 

10BinarySensorEntity objects grouped under a single SuperNotify device, instead of a bare 

11entity_registry entry with a hand-written state and no Entity object behind it. 

12 

13Both are read-only: enabling and disabling a scenario or recipient is done by its switch entity 

14(switch.py). The scenario binary_sensor is the only place a scenario's state - whether its 

15conditions currently hold, as opposed to whether it is enabled - is exposed, so is kept for 

16everyone, for any scenario that hasn't opted out with expose_state. For a scenario with conditions 

17it is read-only, while one without conditions has a manual binary_sensor, which is the control for 

18whether it applies, set from outside Supernotify. The recipient binary_sensor only mirrors the switch, so is deprecated, and only kept 

19for an existing install that already has it, never created for a new one. A one-off repair tells 

20anyone with it enabled that it will be removed in a future version (see repairs.py). 

21 

22The delivery and transport binary_sensors are deprecated in the same way: enabling and disabling 

23moved to their switches, so these only mirror whether each is enabled, are kept only for an 

24existing install that already has them, and only for a delivery or transport that is loaded. A 

25one-off repair tells anyone with one enabled that they will be removed in a future version. 

26 

27entity_id and unique_id are chosen deliberately to line up with the raw state writes these 

28entities replaced (binary_sensor.supernotify_scenario_<name> / _recipient_<name>, unique_id 

29"scenario_<name>" / "recipient_<name>" / "delivery_<name>" / "transport_<name>" with no 

30config-entry prefix) so that upgrading an existing installation adopts the same registry entry 

31and history instead of creating a duplicate. 

32""" 

33 

34from __future__ import annotations 

35 

36import logging 

37from typing import TYPE_CHECKING 

38 

39from homeassistant.components.binary_sensor import BinarySensorEntity 

40from homeassistant.const import STATE_OFF, STATE_ON, EntityCategory, Platform 

41from homeassistant.core import callback 

42from homeassistant.helpers import entity_registry as er 

43from homeassistant.helpers import issue_registry as ir 

44from homeassistant.helpers.event import async_track_state_change_event 

45from homeassistant.helpers.restore_state import RestoreEntity 

46 

47from . import DOMAIN 

48from .common import sanitize 

49from .const import ( 

50 CONF_DEPRECATED_ENTITIES, 

51 CONF_HOUSEKEEPING, 

52 DELIVERY_UNRECORDED_ATTRIBUTES, 

53 OVERRIDE_KIND_DELIVERY, 

54 OVERRIDE_KIND_TRANSPORT, 

55 TRANSPORT_UNRECORDED_ATTRIBUTES, 

56) 

57from .hass_api import ha_device_info 

58from .repairs import ( 

59 DELIVERY_TRANSPORT_BINARY_SENSOR_DEPRECATED_ISSUE_ID, 

60 RECIPIENT_BINARY_SENSOR_DEPRECATED_ISSUE_ID, 

61 async_create_delivery_transport_binary_sensor_deprecated_issue, 

62 async_create_recipient_binary_sensor_deprecated_issue, 

63) 

64 

65_LOGGER = logging.getLogger(__name__) 

66 

67if TYPE_CHECKING: 

68 from homeassistant.core import Event, HomeAssistant 

69 from homeassistant.helpers.device_registry import DeviceInfo 

70 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback 

71 from homeassistant.helpers.event import EventStateChangedData 

72 

73 from . import SupernotifyConfigEntry 

74 from .delivery import Delivery, DeliveryRegistry 

75 from .people import PeopleRegistry, Recipient 

76 from .scenario import Scenario, ScenarioRegistry 

77 from .transport import Transport 

78 

79# Entities are pushed, never polled, and their actions only change local state 

80PARALLEL_UPDATES = 0 

81 

82 

83async def async_setup_entry( 

84 hass: HomeAssistant, 

85 entry: SupernotifyConfigEntry, 

86 async_add_entities: AddConfigEntryEntitiesCallback, 

87) -> None: 

88 """Expose the state of each scenario, and keep the deprecated recipient, delivery and 

89 transport binary_sensors for an install that has them.""" 

90 service = entry.runtime_data 

91 device_info = ha_device_info(entry.entry_id) 

92 

93 entity_registry = er.async_get(hass) 

94 entities: list[BinarySensorEntity] = [] 

95 for scenario in service.context.scenario_registry.scenarios.values(): 

96 if service.context.scenario_registry.scenario_has_state(scenario): 

97 sensor_class = SupernotifyScenarioManualBinarySensor if scenario.is_manual else SupernotifyScenarioBinarySensor 

98 entities.append(sensor_class(scenario, service.context.scenario_registry, device_info)) 

99 elif entity_id := entity_registry.async_get_entity_id(Platform.BINARY_SENSOR, DOMAIN, f"scenario_{scenario.name}"): 

100 # an install from before that has since opted out with expose_state - remove it 

101 # rather than leave it as an entity that's no longer provided 

102 _LOGGER.info("SUPERNOTIFY Removing binary_sensor for scenario %s, it has no state to show", scenario.name) 

103 entity_registry.async_remove(entity_id) 

104 hass.states.async_remove(entity_id) 

105 

106 # The recipient, delivery and transport binary_sensors are deprecated, so only kept for an 

107 # existing install that already has them, and never published for anyone new - which 

108 # includes a recipient, delivery or transport added to an existing install. Whether an entity 

109 # exists is known from its registry entry, which is only created by adding the entity. 

110 # Housekeeping's deprecated_entities switched off removes them, for good. 

111 if not entry.options.get(CONF_HOUSEKEEPING, {}).get(CONF_DEPRECATED_ENTITIES, True): 

112 async_remove_deprecated_entities(hass, entry.entry_id) 

113 async_add_entities(entities) 

114 return 

115 in_use = False 

116 for recipient in service.context.people_registry.people.values(): 

117 registry_entry = _existing_row(entity_registry, f"recipient_{recipient.name}") 

118 if registry_entry is not None: 

119 entities.append(SupernotifyRecipientBinarySensor(recipient, service.context.people_registry, device_info)) 

120 # only worth a warning if somebody could still be relying on it 

121 in_use = in_use or registry_entry.disabled_by is None 

122 

123 delivery_registry = service.context.delivery_registry 

124 legacy_in_use = False 

125 legacy_sensors: list[SupernotifyLegacyBinarySensor] = [ 

126 SupernotifyTransportBinarySensor(transport, delivery_registry, device_info) 

127 for transport in delivery_registry.transports.values() 

128 ] 

129 legacy_sensors.extend( 

130 SupernotifyDeliveryBinarySensor(delivery, delivery_registry, device_info) 

131 for delivery in delivery_registry.deliveries.values() 

132 ) 

133 for legacy_sensor in legacy_sensors: 

134 registry_entry = _existing_row(entity_registry, legacy_sensor.unique_id or "") 

135 if registry_entry is not None: 

136 entities.append(legacy_sensor) 

137 legacy_in_use = legacy_in_use or registry_entry.disabled_by is None 

138 

139 async_add_entities(entities) 

140 if in_use: 

141 async_create_recipient_binary_sensor_deprecated_issue(hass) 

142 if legacy_in_use: 

143 async_create_delivery_transport_binary_sensor_deprecated_issue(hass) 

144 

145 

146def _existing_row(entity_registry: er.EntityRegistry, unique_id: str) -> er.RegistryEntry | None: 

147 """The registry entry of one of our binary_sensors, if an earlier version already created it.""" 

148 entity_id = entity_registry.async_get_entity_id(Platform.BINARY_SENSOR, DOMAIN, unique_id) 

149 return entity_registry.async_get(entity_id) if entity_id else None 

150 

151 

152class SupernotifyScenarioBinarySensor(BinarySensorEntity): 

153 """A scenario's evaluated condition state - whether it currently applies, as opposed to 

154 whether it is enabled, which is the scenario switch. A scenario with no conditions to 

155 evaluate gets SupernotifyScenarioManualBinarySensor instead. 

156 

157 Read-only: writing its state does not enable or disable the scenario. 

158 

159 See ScenarioRegistry._scenario_state()/scenario_is_on() for how ON/OFF/unknown is derived, 

160 and async_refresh_scenario_states() for what triggers a re-read of this entity's state. 

161 """ 

162 

163 _attr_has_entity_name = True 

164 _attr_entity_category = EntityCategory.DIAGNOSTIC 

165 _attr_translation_key = "scenario" 

166 _attr_should_poll = False 

167 

168 def __init__(self, scenario: Scenario, registry: ScenarioRegistry, device_info: DeviceInfo) -> None: 

169 self._scenario = scenario 

170 self._registry = registry 

171 self._attr_unique_id = f"scenario_{scenario.name}" 

172 self._attr_device_info = device_info 

173 self._attr_translation_placeholders = {"scenario": scenario.alias or scenario.name} 

174 # Setting entity_id directly is not preferred Home Assistant practice - see 

175 # SupernotifyScenarioSwitch for why it's done anyway, and why it can't change the 

176 # entity_id of an existing install. 

177 self.entity_id = f"binary_sensor.{DOMAIN}_scenario_{scenario.name}" 

178 

179 @property 

180 def is_on(self) -> bool | None: 

181 return self._registry.scenario_is_on(self._scenario) 

182 

183 @property 

184 def extra_state_attributes(self) -> dict[str, object]: 

185 return sanitize(self._scenario.attributes(include_condition=False)) 

186 

187 async def async_added_to_hass(self) -> None: 

188 await super().async_added_to_hass() 

189 self._registry.register_entity(self._scenario.name, self) 

190 

191 async def async_will_remove_from_hass(self) -> None: 

192 self._registry.unregister_entity(self._scenario.name) 

193 await super().async_will_remove_from_hass() 

194 

195 

196class SupernotifyScenarioManualBinarySensor(SupernotifyScenarioBinarySensor, RestoreEntity): 

197 """The state of a scenario that has no conditions of its own to evaluate, set from outside 

198 Supernotify - typically a script or automation writing its state, or calling a service on it. 

199 

200 Unlike the condition binary_sensor, this is the control: whatever it is set to is applied 

201 back to the scenario, which then evaluates true while it is on, in the same way as one whose 

202 conditions hold. It is restored across a restart. Whether the scenario is enabled is still 

203 the scenario switch, and a disabled scenario never applies whatever this is set to. 

204 """ 

205 

206 _attr_translation_key = "scenario_manual" 

207 

208 async def async_added_to_hass(self) -> None: 

209 await super().async_added_to_hass() 

210 last_state = await self.async_get_last_state() 

211 if last_state is not None and last_state.state in (STATE_ON, STATE_OFF): 

212 self._scenario.manual_active = last_state.state == STATE_ON 

213 self.async_on_remove(async_track_state_change_event(self.hass, [self.entity_id], self._async_state_changed)) 

214 

215 @callback 

216 def _async_state_changed(self, event: Event[EventStateChangedData]) -> None: 

217 """Apply a state written to this entity from outside back to the scenario""" 

218 new_state = event.data["new_state"] 

219 if new_state is None or new_state.state not in (STATE_ON, STATE_OFF): 

220 return 

221 active = new_state.state == STATE_ON 

222 if active != self._scenario.manual_active: 

223 _LOGGER.info("SUPERNOTIFY Scenario %s manually set %s", self._scenario.name, new_state.state) 

224 self._scenario.manual_active = active 

225 # put back the entity's own attributes, which a plain state write would have replaced, 

226 # attributed to whatever wrote the state 

227 self.async_set_context(event.context) 

228 self.async_write_ha_state() 

229 

230 

231class SupernotifyRecipientBinarySensor(BinarySensorEntity): 

232 """Whether a recipient is currently enabled for delivery. Deprecated, see the recipient switch. 

233 

234 Read-only: writing its state no longer enables or disables the recipient. 

235 """ 

236 

237 _attr_has_entity_name = True 

238 _attr_entity_category = EntityCategory.DIAGNOSTIC 

239 # No device_class: CONNECTIVITY (a previous version of this class) is semantically wrong 

240 # here - it means online/offline device reachability, not "enabled for delivery", and made 

241 # a disabled recipient show as "Disconnected" in the dashboard. Icon (see icons.json) and 

242 # translation_key below already carry the meaning without borrowing a misleading one. 

243 _attr_translation_key = "recipient" 

244 _attr_should_poll = False 

245 

246 def __init__(self, recipient: Recipient, registry: PeopleRegistry, device_info: DeviceInfo) -> None: 

247 self._recipient = recipient 

248 self._registry = registry 

249 self._attr_unique_id = f"recipient_{recipient.name}" 

250 self._attr_device_info = device_info 

251 self._attr_translation_placeholders = {"recipient": recipient.alias or recipient.name} 

252 # Setting entity_id directly is not preferred Home Assistant practice - see 

253 # SupernotifyScenarioSwitch for why it's done anyway, and why it can't change the 

254 # entity_id of an existing install. 

255 self.entity_id = f"binary_sensor.{DOMAIN}_recipient_{recipient.name}" 

256 

257 @property 

258 def is_on(self) -> bool: 

259 return self._recipient.enabled 

260 

261 @property 

262 def extra_state_attributes(self) -> dict[str, object]: 

263 return sanitize(self._recipient.attributes()) 

264 

265 async def async_added_to_hass(self) -> None: 

266 await super().async_added_to_hass() 

267 self._registry.register_entity(self._recipient.name, self) 

268 

269 async def async_will_remove_from_hass(self) -> None: 

270 self._registry.unregister_entity(self._recipient.name) 

271 await super().async_will_remove_from_hass() 

272 

273 

274class SupernotifyLegacyBinarySensor(BinarySensorEntity): 

275 """Whether a delivery or transport is currently enabled. Deprecated, see its switch. 

276 

277 Read-only: writing its state no longer enables or disables anything. No entity_id is set, 

278 as none is needed: an existing registry entry, found by unique_id, always provides it. 

279 """ 

280 

281 _attr_has_entity_name = True 

282 _attr_entity_category = EntityCategory.DIAGNOSTIC 

283 _attr_entity_registry_enabled_default = False 

284 _attr_should_poll = False 

285 

286 def __init__(self, kind: str, model: Delivery | Transport, registry: DeliveryRegistry, device_info: DeviceInfo) -> None: 

287 self._model = model 

288 self._registry = registry 

289 self._key = f"{kind}_{model.name}" 

290 self._attr_unique_id = self._key 

291 self._attr_device_info = device_info 

292 self._attr_translation_placeholders = {kind: model.alias or model.name} 

293 

294 @property 

295 def is_on(self) -> bool: 

296 return bool(self._model.enabled) 

297 

298 @property 

299 def extra_state_attributes(self) -> dict[str, object]: 

300 return sanitize(self._model.attributes()) 

301 

302 async def async_added_to_hass(self) -> None: 

303 await super().async_added_to_hass() 

304 self._registry.register_entity(self._key, self) 

305 

306 async def async_will_remove_from_hass(self) -> None: 

307 self._registry.unregister_entity(self._key) 

308 await super().async_will_remove_from_hass() 

309 

310 

311class SupernotifyDeliveryBinarySensor(SupernotifyLegacyBinarySensor): 

312 """Whether a delivery is currently enabled. Deprecated, see the delivery switch.""" 

313 

314 _attr_translation_key = "delivery" 

315 _unrecorded_attributes = DELIVERY_UNRECORDED_ATTRIBUTES 

316 

317 def __init__(self, delivery: Delivery, registry: DeliveryRegistry, device_info: DeviceInfo) -> None: 

318 super().__init__(OVERRIDE_KIND_DELIVERY, delivery, registry, device_info) 

319 

320 

321class SupernotifyTransportBinarySensor(SupernotifyLegacyBinarySensor): 

322 """Whether a transport is currently enabled. Deprecated, see the transport switch.""" 

323 

324 _attr_translation_key = "transport" 

325 _unrecorded_attributes = TRANSPORT_UNRECORDED_ATTRIBUTES 

326 

327 def __init__(self, transport: Transport, registry: DeliveryRegistry, device_info: DeviceInfo) -> None: 

328 super().__init__(OVERRIDE_KIND_TRANSPORT, transport, registry, device_info) 

329 

330 

331_DEPRECATED_UNIQUE_ID_PREFIXES = ("recipient_", "delivery_", "transport_") 

332 

333 

334def deprecated_entity_ids(hass: HomeAssistant, entry_id: str) -> list[str]: 

335 """The deprecated recipient, delivery and transport binary_sensors this entry still has""" 

336 return [ 

337 e.entity_id 

338 for e in er.async_entries_for_config_entry(er.async_get(hass), entry_id) 

339 if e.domain == Platform.BINARY_SENSOR and e.unique_id.startswith(_DEPRECATED_UNIQUE_ID_PREFIXES) 

340 ] 

341 

342 

343def async_remove_deprecated_entities(hass: HomeAssistant, entry_id: str) -> None: 

344 """Remove the deprecated binary_sensors, and the repairs asking for them to be replaced""" 

345 entity_registry = er.async_get(hass) 

346 for entity_id in deprecated_entity_ids(hass, entry_id): 

347 _LOGGER.info("SUPERNOTIFY Removing deprecated %s", entity_id) 

348 entity_registry.async_remove(entity_id) 

349 hass.states.async_remove(entity_id) 

350 for issue_id in (RECIPIENT_BINARY_SENSOR_DEPRECATED_ISSUE_ID, DELIVERY_TRANSPORT_BINARY_SENSOR_DEPRECATED_ISSUE_ID): 

351 ir.async_delete_issue(hass, DOMAIN, issue_id)