Coverage for custom_components/camera_autonotify/hub.py: 98%

1"""Reviews from Frigate over MQTT, one subscription for each topic prefix, shared by the entries that need it, and 

2each entry's watcher, which times what its tracker holds back and sends what it decides""" 

3 

4from __future__ import annotations 

5 

6import logging 

7import time 

8from dataclasses import dataclass, field 

9from typing import TYPE_CHECKING, Any 

10 

11from homeassistant.components import mqtt 

12from homeassistant.const import CONF_TYPE 

13from homeassistant.core import CALLBACK_TYPE, Context, HomeAssistant, callback 

14from homeassistant.helpers.event import async_call_later 

15 

16from .const import CONF_SOURCE_NAME, DOMAIN, TYPE_ZONE 

17from .notifier import Notifier 

18from .tracker import Decision, ReviewTracker, Settings, Wait 

19 

20if TYPE_CHECKING: 

21 from datetime import datetime 

22 

23 from homeassistant.components.mqtt import ReceiveMessage 

24 from homeassistant.config_entries import ConfigEntry 

25 

26_LOGGER = logging.getLogger(__name__) 

27 

28REVIEWS_TOPIC = "{prefix}/reviews" 

29DATA_HUBS = "hubs" 

30 

31 

32class ReviewWatcher: 

33 """The reviews of one camera's or zone's entry""" 

34 

35 def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None: 

36 self.hass = hass 

37 self.entry = entry 

38 settings = Settings.from_options(entry.options) 

39 name = entry.data[CONF_SOURCE_NAME] 

40 if entry.data.get(CONF_TYPE) == TYPE_ZONE: 

41 self.tracker = ReviewTracker(settings, zone=name) 

42 else: 

43 self.tracker = ReviewTracker(settings, camera=name) 

44 self.notifier = Notifier(hass, entry) 

45 # each review's notifications tied together, the later ones as children of the first 

46 self._contexts: dict[str, Context] = {} 

47 self._timers: dict[str, CALLBACK_TYPE] = {} 

48 

49 @callback 

50 def async_review(self, payload: str | bytes | bytearray) -> None: 

51 try: 

52 outcome = self.tracker.on_message(payload, time.time()) 

53 except ValueError, KeyError, TypeError: 

54 _LOGGER.warning("CAMERA_AUTONOTIFY Review not understood: %s", payload) 

55 return 

56 self._act(outcome) 

57 

58 @callback 

59 def async_stop(self) -> None: 

60 for cancel in self._timers.values(): 

61 cancel() 

62 self._timers.clear() 

63 

64 @property 

65 def reviews(self) -> list[str]: 

66 return self.tracker.tracked 

67 

68 def _act(self, outcome: Decision | Wait | None) -> None: 

69 if isinstance(outcome, Wait): 

70 self._wait(outcome) 

71 elif isinstance(outcome, Decision): 

72 self.entry.async_create_background_task( 

73 self.hass, 

74 self.notifier.async_notify(outcome, self._context(outcome.review.id)), 

75 f"{DOMAIN} notify {outcome.review.id}", 

76 ) 

77 # done with, once the tracker is 

78 for review_id in set(self._contexts) - set(self.tracker.tracked): 

79 del self._contexts[review_id] 

80 

81 def _wait(self, wait: Wait) -> None: 

82 if cancel := self._timers.pop(wait.review_id, None): 

83 cancel() 

84 

85 @callback 

86 def due(now: datetime) -> None: 

87 self._timers.pop(wait.review_id, None) 

88 self._act(self.tracker.on_timer(wait.review_id, time.time())) 

89 

90 self._timers[wait.review_id] = async_call_later(self.hass, max(0.0, wait.at - time.time()), due) 

91 

92 def _context(self, review_id: str) -> Context: 

93 first = self._contexts.get(review_id) 

94 if first is None: 

95 first = self._contexts[review_id] = Context() 

96 return first 

97 return Context(parent_id=first.id) 

98 

99 

100@dataclass 

101class _Hub: 

102 unsubscribe: CALLBACK_TYPE 

103 watchers: list[ReviewWatcher] = field(default_factory=list) 

104 

105 

106def _hubs(hass: HomeAssistant) -> dict[str, _Hub]: 

107 data: dict[str, Any] = hass.data.setdefault(DOMAIN, {}) 

108 return data.setdefault(DATA_HUBS, {}) 

109 

110 

111async def async_watch(hass: HomeAssistant, prefix: str, watcher: ReviewWatcher) -> CALLBACK_TYPE: 

112 """Start the watcher on the reviews of a Frigate server, returning what stops it""" 

113 hubs = _hubs(hass) 

114 hub = hubs.get(prefix) 

115 if hub is None: 

116 

117 @callback 

118 def received(message: ReceiveMessage) -> None: 

119 for each in list(hubs[prefix].watchers): 

120 each.async_review(message.payload) 

121 

122 unsubscribe = await mqtt.async_subscribe(hass, REVIEWS_TOPIC.format(prefix=prefix), received) 

123 # another entry may have subscribed while this one was waiting 

124 if (hub := hubs.get(prefix)) is None: 

125 hub = hubs[prefix] = _Hub(unsubscribe) 

126 else: 

127 unsubscribe() 

128 hub.watchers.append(watcher) 

129 

130 @callback 

131 def stop() -> None: 

132 watcher.async_stop() 

133 if watcher in hub.watchers: 

134 hub.watchers.remove(watcher) 

135 if not hub.watchers and hubs.get(prefix) is hub: 

136 hub.unsubscribe() 

137 del hubs[prefix] 

138 

139 return stop