Coverage for src / updates2mqtt / hass_formatter.py: 67%

1from typing import Any 

2 

3import structlog 

4 

5import updates2mqtt 

6from updates2mqtt.model import Discovery 

7 

8log = structlog.get_logger() 

9HASS_UPDATE_SCHEMA = [ 

10 "installed_version", 

11 "latest_version", 

12 "title", 

13 "release_summary", 

14 "release_url", 

15 "entity_picture", 

16 "in_progress", 

17 "update_percentage", 

18] 

19 

20 

21def hass_format_config( 

22 discovery: Discovery, 

23 object_id: str, 

24 state_topic: str, 

25 command_topic: str | None, 

26 attrs_topic: str | None, 

27 force_command_topic: bool | None, 

28 device_creation: bool = True, 

29 area: str | None = None, 

30) -> dict[str, Any]: 

31 config: dict[str, Any] = { 

32 "name": discovery.title, 

33 "device_class": None, # not firmware, so defaults to null 

34 "unique_id": object_id, 

35 "state_topic": state_topic, 

36 "supported_features": discovery.features, 

37 "default_entity_id": f"update.{discovery.node}_{discovery.provider.source_type}_{discovery.name}", 

38 "origin": { 

39 "name": f"{discovery.node} updates2mqtt", 

40 "sw_version": updates2mqtt.version, # pyright: ignore[reportAttributeAccessIssue] 

41 "support_url": "https://github.com/rhizomatics/updates2mqtt/issues", 

42 }, 

43 } 

44 if attrs_topic: 44 ↛ 46line 44 didn't jump to line 46 because the condition on line 44 was always true

45 config["json_attributes_topic"] = attrs_topic 

46 if discovery.entity_picture_url: 46 ↛ 47line 46 didn't jump to line 47 because the condition on line 46 was never true

47 config["entity_picture"] = discovery.entity_picture_url 

48 if discovery.device_icon: 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true

49 config["icon"] = discovery.device_icon 

50 if device_creation: 

51 config["device"] = { 

52 "name": f"{discovery.node} updates2mqtt", 

53 "sw_version": updates2mqtt.version, # pyright: ignore[reportAttributeAccessIssue] 

54 "manufacturer": "rhizomatics", 

55 "identifiers": [f"{discovery.node}.updates2mqtt"], 

56 } 

57 if area: 

58 config["device"]["suggested_area"] = area 

59 if command_topic and (discovery.can_update or force_command_topic): 

60 config["command_topic"] = command_topic 

61 if discovery.can_update: 

62 config["payload_install"] = f"{discovery.source_type}|{discovery.name}|install" 

63 

64 return config 

65 

66 

67def hass_format_state(discovery: Discovery, in_progress: bool = False, release_summary_max_size: int = 6144) -> dict[str, Any]: 

68 state: dict[str, str | dict | list | bool | None] = { 

69 "installed_version": discovery.current_version, 

70 "latest_version": discovery.latest_version, 

71 "title": discovery.title, 

72 "in_progress": in_progress, 

73 } 

74 if discovery.release_detail: 74 ↛ 75line 74 didn't jump to line 75 because the condition on line 74 was never true

75 if discovery.release_detail.summary: 

76 if len(discovery.release_detail.summary) > release_summary_max_size: 

77 log.warn("Release notes for %s truncated to %s", discovery.name, release_summary_max_size) 

78 state["release_summary"] = discovery.release_detail.summary[:release_summary_max_size] 

79 if discovery.release_detail.notes_url: 

80 state["release_url"] = discovery.release_detail.notes_url 

81 

82 return state