Coverage for src / anpr2mqtt / hass.py: 60%

1import datetime as dt 

2import json 

3from io import BytesIO 

4from pathlib import Path 

5from typing import Any 

6 

7import paho.mqtt.client as mqtt 

8import structlog 

9from PIL import Image 

10 

11import anpr2mqtt 

12from anpr2mqtt.settings import EventSettings 

13 

14from .const import ImageInfo 

15 

16log = structlog.get_logger() 

17 

18 

19def post_discovery_message( 

20 client: mqtt.Client, 

21 discovery_topic_prefix: str, 

22 state_topic: str, 

23 image_topic: str, 

24 event_config: EventSettings, 

25 device_creation: bool = True, 

26) -> None: 

27 topic = f"{discovery_topic_prefix}/sensor/{event_config.camera}/{event_config.event}/config" 

28 name: str = event_config.description or f"{event_config.event} {event_config.camera}" 

29 payload: dict[str, Any] = { 

30 "o": { 

31 "name": "anpr2mqtt", 

32 "sw": anpr2mqtt.version, # pyright: ignore[reportAttributeAccessIssue] 

33 "url": "https://anpt2mqtt.rhizomatics.org.uk", 

34 }, 

35 "device_class": None, 

36 "value_template": "{{ value_json.target }}", 

37 "unique_id": f"{event_config.event}_{event_config.camera}", 

38 "state_topic": state_topic, 

39 "json_attributes_topic": state_topic, 

40 "icon": "mdi:car-back", 

41 "name": name, 

42 } 

43 if device_creation: 

44 add_device_info(payload, event_config) 

45 client.publish(topic, payload=json.dumps(payload), qos=0, retain=True) 

46 log.info("Published HA MQTT sensor Discovery message to %s", topic) 

47 

48 topic = f"{discovery_topic_prefix}/image/{event_config.camera}/{event_config.event}/config" 

49 payload = { 

50 "o": { 

51 "name": "anpr2mqtt", 

52 "sw": anpr2mqtt.version, # pyright: ignore[reportAttributeAccessIssue] 

53 "url": "https://anpt2mqtt.rhizomatics.org.uk", 

54 }, 

55 "device_class": None, 

56 "unique_id": f"{event_config.event}_{event_config.camera}", 

57 "image_topic": image_topic, 

58 "json_attributes_topic": state_topic, 

59 "icon": "mdi:car-back", 

60 "name": name, 

61 } 

62 if device_creation: 

63 add_device_info(payload, event_config) 

64 client.publish(topic, payload=json.dumps(payload), qos=0, retain=True) 

65 log.info("Published HA MQTT Discovery message to %s", topic) 

66 

67 

68def add_device_info(payload: dict[str, Any], event_config: EventSettings) -> None: 

69 payload["dev"] = { 

70 "name": f"anpr2mqtt on {event_config.camera}", 

71 "sw_version": anpr2mqtt.version, # pyright: ignore[reportAttributeAccessIssue] 

72 "manufacturer": "rhizomatics", 

73 "identifiers": [f"{event_config.event}_{event_config.camera}.anpr2mqtt"], 

74 } 

75 if event_config.area: 

76 payload["dev"]["suggested_area"] = event_config.area 

77 

78 

79def post_state_message( 

80 client: mqtt.Client, 

81 topic: str, 

82 target: str | None, 

83 event_config: EventSettings, 

84 ocr_fields: dict[str, str | None], 

85 image_info: ImageInfo | None = None, 

86 classification: dict[str, Any] | None = None, 

87 previous_sightings: int | None = None, 

88 last_sighting: dt.datetime | None = None, 

89 url: str | None = None, 

90 error: str | None = None, 

91 file_path: Path | None = None, 

92 reg_info: Any = None, 

93) -> None: 

94 payload: dict[str, Any] = { 

95 "target": target, 

96 "target_type": event_config.target_type, 

97 event_config.target_type: target, 

98 "event": event_config.event, 

99 "camera": event_config.camera or "UNKNOWN", 

100 "area": event_config.area, 

101 "reg_info": reg_info, 

102 } 

103 payload.update(ocr_fields) 

104 if error: 104 ↛ 105line 104 didn't jump to line 105 because the condition on line 104 was never true

105 payload["error"] = error 

106 if url is not None: 106 ↛ 108line 106 didn't jump to line 108 because the condition on line 106 was always true

107 payload["event_image_url"] = url 

108 if file_path is not None: 108 ↛ 110line 108 didn't jump to line 110 because the condition on line 108 was always true

109 payload["file_path"] = str(file_path) 

110 if classification is not None: 

111 payload.update(classification) 

112 if previous_sightings is not None: 

113 payload["previous_sightings"] = previous_sightings 

114 if last_sighting is not None: 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true

115 payload["last_sighting"] = last_sighting.isoformat() 

116 

117 try: 

118 if image_info: 

119 payload.update( 

120 { 

121 "event_time": image_info.timestamp.isoformat(), 

122 "image_event": image_info.event, 

123 "ext": image_info.ext, 

124 "image_size": image_info.size, 

125 } 

126 ) 

127 

128 client.publish(topic, payload=json.dumps(payload), qos=0, retain=True) 

129 log.debug("Published HA MQTT State message to %s: %s", topic, payload) 

130 except Exception as e: 

131 log.error("Failed to publish event %s: %s", payload, e, exc_info=1) 

132 

133 

134def post_image_message(client: mqtt.Client, topic: str, image: Image.Image, img_format: str = "JPEG") -> None: 

135 try: 

136 img_byte_arr = BytesIO() 

137 image.save(img_byte_arr, format=img_format) 

138 img_bytes = img_byte_arr.getvalue() 

139 

140 client.publish(topic, payload=img_bytes, qos=0, retain=True) 

141 log.debug("Published HA MQTT Image message to %s: %s bytes", topic, len(img_bytes)) 

142 except Exception as e: 

143 log.error("Failed to publish image entity: %s", e, exc_info=1)