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

1"""What's known of a Frigate server, read from the Frigate integration, and the links its notification proxy serves""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from homeassistant.core import HomeAssistant, callback 

8from homeassistant.helpers import entity_registry as er 

9from homeassistant.helpers.network import NoURLAvailableError, get_url 

10from homeassistant.util import slugify 

11 

12from .const import FRIGATE_DOMAIN 

13 

14DEFAULT_TOPIC_PREFIX = "frigate" 

15# the Frigate integration keeps each server's config, as Frigate itself reports it, under its entry id 

16FRIGATE_CONFIG = "config" 

17# served without a login, so a phone can fetch it, for as long as the Frigate entry allows 

18NOTIFICATIONS_PATH = "/api/frigate/notifications" 

19 

20 

21@callback 

22def frigate_config(hass: HomeAssistant, source_entry_id: str) -> dict[str, Any] | None: 

23 """The config of a Frigate server, once its entry is set up""" 

24 data = hass.data.get(FRIGATE_DOMAIN, {}).get(source_entry_id) 

25 return data.get(FRIGATE_CONFIG) if isinstance(data, dict) else None 

26 

27 

28def topic_prefix(config: dict[str, Any] | None) -> str: 

29 return ((config or {}).get("mqtt") or {}).get("topic_prefix") or DEFAULT_TOPIC_PREFIX 

30 

31 

32@callback 

33def base_url(hass: HomeAssistant) -> str | None: 

34 """The address a phone can reach Home Assistant at, away from home if there's one for that""" 

35 try: 

36 return get_url(hass, prefer_external=True) 

37 except NoURLAvailableError: 

38 return None 

39 

40 

41def snapshot_url(base: str, detection_id: str) -> str: 

42 return f"{base}{NOTIFICATIONS_PATH}/{detection_id}/snapshot.jpg" 

43 

44 

45def hls_url(base: str, detection_id: str, camera: str) -> str: 

46 """The clip as HLS, which iOS plays when tapped, where an attached `clip.mp4` fails to show at all""" 

47 return f"{base}{NOTIFICATIONS_PATH}/{detection_id}/{camera}/master.m3u8" 

48 

49 

50def review_thumbnail_url(base: str, review_id: str, camera: str) -> str: 

51 return f"{base}{NOTIFICATIONS_PATH}/{review_id}/{camera}/review_thumbnail.webp" 

52 

53 

54@callback 

55def camera_entity_id(hass: HomeAssistant, source_entry_id: str, camera: str) -> str | None: 

56 """The camera entity the Frigate integration has for a camera, by the unique id it gives it""" 

57 return er.async_get(hass).async_get_entity_id("camera", FRIGATE_DOMAIN, f"{source_entry_id}:camera:{camera}") 

58 

59 

60def _labels(camera: dict[str, Any]) -> set[str]: 

61 """What a camera can see or hear, and what its reviews are made of""" 

62 review = camera.get("review") or {} 

63 return { 

64 *((camera.get("objects") or {}).get("track") or []), 

65 *((camera.get("audio") or {}).get("listen") or []), 

66 *((review.get("alerts") or {}).get("labels") or []), 

67 *((review.get("detections") or {}).get("labels") or []), 

68 } 

69 

70 

71def camera_choices(config: dict[str, Any] | None, camera: str) -> tuple[list[str], list[str]]: 

72 """The labels and zones of a camera, as its config has them, to choose from""" 

73 settings = ((config or {}).get("cameras") or {}).get(camera) or {} 

74 return sorted(_labels(settings)), list(settings.get("zones") or {}) 

75 

76 

77def zone_labels(config: dict[str, Any] | None, zone: str) -> list[str]: 

78 """The labels of every camera a zone is on, the zone as slugified for its device""" 

79 cameras = ((config or {}).get("cameras") or {}).values() 

80 return sorted({ 

81 label for camera in cameras if zone in {slugify(z) for z in camera.get("zones") or {}} for label in _labels(camera) 

82 })