Coverage for custom_components/autoarm/frigate.py: 100%

1"""Keep the active Frigate profile in step with the alarm state 

2 

3See https://docs.frigate.video/configuration/profiles/ 

4""" 

5 

6import logging 

7from collections.abc import Callable 

8 

9from homeassistant.components.select.const import ATTR_OPTIONS 

10from homeassistant.components.select.const import DOMAIN as SELECT_DOMAIN 

11from homeassistant.const import ATTR_ENTITY_ID, ATTR_OPTION, SERVICE_SELECT_OPTION, STATE_UNAVAILABLE, STATE_UNKNOWN 

12from homeassistant.core import Context, Event, EventStateChangedData, HomeAssistant 

13from homeassistant.exceptions import HomeAssistantError 

14from homeassistant.helpers import entity_registry as er 

15from homeassistant.helpers.event import async_track_state_change_event 

16 

17from .const import EVENT_TRIGGERED, FRIGATE_NO_PROFILE, PUBLIC_ALARM_STATES 

18from .helpers import child_context 

19 

20_LOGGER = logging.getLogger(__name__) 

21 

22# ends the unique id of the Frigate integration's profile select, after its config entry id 

23FRIGATE_PROFILE_UNIQUE_ID_SUFFIX = ":select:profile" 

24 

25 

26def match_profile(state: str, available: list[str]) -> str | None: 

27 """The profile named for an alarm state, ignoring case, or else one named armed for any of the armed states""" 

28 by_name = {p.lower(): p for p in available if p != FRIGATE_NO_PROFILE} 

29 return by_name.get(state) or (by_name.get("armed") if state.startswith("armed_") else None) 

30 

31 

32def frigate_profile_selects(hass: HomeAssistant) -> list[str]: 

33 """Profile selects of the Frigate integration, one for each Frigate server with profiles""" 

34 return [ 

35 entry.entity_id 

36 for entry in er.async_get(hass).entities.values() 

37 if entry.platform == "frigate" and entry.unique_id.endswith(FRIGATE_PROFILE_UNIQUE_ID_SUFFIX) 

38 ] 

39 

40 

41def frigate_profiles(hass: HomeAssistant, profile_select: str | None) -> list[str]: 

42 """Profiles of the profile select, or of any Frigate profile select before one is chosen""" 

43 selects = [profile_select] if profile_select else frigate_profile_selects(hass) 

44 return list( 

45 dict.fromkeys( 

46 p 

47 for entity_id in selects 

48 if (state := hass.states.get(entity_id)) 

49 for p in state.attributes.get(ATTR_OPTIONS, []) 

50 if p != FRIGATE_NO_PROFILE 

51 ) 

52 ) 

53 

54 

55async def async_track_frigate_profile( 

56 hass: HomeAssistant, alarm_panel: str, profile_select: str, profiles: dict[str, str] 

57) -> Callable[[], None]: 

58 """Switch Frigate to the profile given for each alarm state, and to no profile for the others 

59 

60 A state never given a profile is matched by name to one of the select's profiles, 

61 while transitional states, like arming or triggered, leave the profile as it is 

62 """ 

63 

64 async def sync(context: Context | None = None) -> None: 

65 panel = hass.states.get(alarm_panel) 

66 select = hass.states.get(profile_select) 

67 if ( 

68 panel is None 

69 or panel.state not in PUBLIC_ALARM_STATES 

70 or select is None 

71 or select.state in (STATE_UNAVAILABLE, STATE_UNKNOWN) 

72 ): 

73 return 

74 available: list[str] = select.attributes.get(ATTR_OPTIONS, []) 

75 profile = ( 

76 profiles[panel.state] if panel.state in profiles else match_profile(panel.state, available) or FRIGATE_NO_PROFILE 

77 ) 

78 if profile not in available: 

79 # gone from Frigate, so back to the base config rather than staying on the profile of an earlier state 

80 _LOGGER.warning("AUTOARM No Frigate profile %s on %s for alarm state %s", profile, profile_select, panel.state) 

81 profile = FRIGATE_NO_PROFILE 

82 if profile == select.state: 

83 return 

84 _LOGGER.info("AUTOARM Frigate profile %s-->%s for alarm state %s", select.state, profile, panel.state) 

85 context = context or Context() 

86 if context.origin_event is None: 

87 # the logbook takes the first event of a context as its cause, otherwise that's the select action 

88 hass.bus.async_fire(EVENT_TRIGGERED, {ATTR_ENTITY_ID: profile_select, "alarm_state": panel.state}, context=context) 

89 try: 

90 await hass.services.async_call( 

91 SELECT_DOMAIN, 

92 SERVICE_SELECT_OPTION, 

93 {ATTR_ENTITY_ID: profile_select, ATTR_OPTION: profile}, 

94 blocking=True, 

95 context=context, 

96 ) 

97 except HomeAssistantError as e: 

98 _LOGGER.warning("AUTOARM Failed to set Frigate profile %s on %s: %s", profile, profile_select, e) 

99 

100 async def on_panel_change(event: Event[EventStateChangedData]) -> None: 

101 await sync(child_context(event.context)) 

102 

103 async def on_select_change(event: Event[EventStateChangedData]) -> None: 

104 # only catching up as Frigate comes back, so a profile chosen by hand is left alone 

105 old = event.data["old_state"] 

106 if old is None or old.state in (STATE_UNAVAILABLE, STATE_UNKNOWN): 

107 await sync(child_context(event.context)) 

108 

109 unsubscribes = [ 

110 async_track_state_change_event(hass, [alarm_panel], on_panel_change), 

111 async_track_state_change_event(hass, [profile_select], on_select_change), 

112 ] 

113 await sync() 

114 

115 def unsubscribe() -> None: 

116 for unsub in unsubscribes: 

117 unsub() 

118 

119 return unsubscribe