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

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, session: str, in_progress: bool = False) -> dict[str, Any]: # noqa: ARG001 

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_summary: 74 ↛ 75line 74 didn't jump to line 75 because the condition on line 74 was never true

75 state["release_summary"] = discovery.release_summary 

76 if discovery.release_url: 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true

77 state["release_url"] = discovery.release_url 

78 

79 return state