Coverage for custom_components/remote_logger/sensor.py: 98%

1"""Binary sensor platform for remote_logger.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from typing import TYPE_CHECKING 

7 

8from homeassistant.components.sensor import SensorEntity, SensorEntityDescription, SensorStateClass 

9from homeassistant.helpers import device_registry as dr 

10from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo 

11from homeassistant.util import slugify 

12 

13from custom_components.remote_logger.exporter import LogExporter 

14 

15from .const import DOMAIN 

16from .remote_logger import REF_EXPORTER 

17 

18if TYPE_CHECKING: 

19 from collections.abc import Callable 

20 

21 from homeassistant.config_entries import ConfigEntry 

22 from homeassistant.core import HomeAssistant 

23 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback 

24 

25 from custom_components.remote_logger.exporter import LogExporter 

26 

27from typing import TYPE_CHECKING, Any 

28 

29from homeassistant.const import EntityCategory 

30 

31 

32@dataclass(frozen=True, kw_only=True) 

33class RemoteLoggerDiagnosticEntityDescription(SensorEntityDescription): 

34 """Describes diagnostic sensor entity.""" 

35 

36 value_fn: Callable[[LogExporter], str | int | float | None] 

37 attr_fn: Callable[[LogExporter], dict[str, Any]] = lambda _: {} 

38 

39 

40SENSORS: tuple[RemoteLoggerDiagnosticEntityDescription, ...] = ( 

41 RemoteLoggerDiagnosticEntityDescription( 

42 key="format_errors", 

43 translation_key="format_errors", 

44 native_unit_of_measurement="error", 

45 entity_category=EntityCategory.DIAGNOSTIC, 

46 state_class=SensorStateClass.MEASUREMENT, 

47 value_fn=lambda logger: logger.format_error_count, 

48 attr_fn=lambda exporter: { 

49 "last_error_time": exporter.last_format_error, 

50 "last_error_message": exporter.last_format_error_message, 

51 }, 

52 ), 

53 RemoteLoggerDiagnosticEntityDescription( 

54 key="posting_errors", 

55 translation_key="posting_errors", 

56 native_unit_of_measurement="error", 

57 entity_category=EntityCategory.DIAGNOSTIC, 

58 state_class=SensorStateClass.MEASUREMENT, 

59 value_fn=lambda logger: logger.posting_error_count, 

60 attr_fn=lambda exporter: { 

61 "last_error_time": exporter.last_posting_error, 

62 "last_error_message": exporter.last_posting_error_message, 

63 }, 

64 ), 

65 RemoteLoggerDiagnosticEntityDescription( 

66 key="events", 

67 translation_key="events", 

68 native_unit_of_measurement="event", 

69 entity_category=EntityCategory.DIAGNOSTIC, 

70 state_class=SensorStateClass.MEASUREMENT, 

71 value_fn=lambda exporter: exporter.event_count, 

72 attr_fn=lambda exporter: {"last_event_time": exporter.last_event}, 

73 ), 

74 RemoteLoggerDiagnosticEntityDescription( 

75 key="postings", 

76 translation_key="postings", 

77 native_unit_of_measurement="posting", 

78 entity_category=EntityCategory.DIAGNOSTIC, 

79 state_class=SensorStateClass.MEASUREMENT, 

80 value_fn=lambda exporter: exporter.posting_count, 

81 attr_fn=lambda exporter: {"last_posting_time": exporter.last_posting}, 

82 ), 

83) 

84 

85 

86class LoggerEntity(SensorEntity): 

87 """Represent a diagnostic tracking logger.""" 

88 

89 _attr_entity_category: EntityCategory = EntityCategory.DIAGNOSTIC # pyright: ignore[reportIncompatibleVariableOverride] 

90 _attr_should_poll = True 

91 _attr_has_entity_name = True 

92 

93 def __init__( 

94 self, exporter: LogExporter, description: RemoteLoggerDiagnosticEntityDescription, device_info: DeviceInfo 

95 ) -> None: 

96 super().__init__() 

97 self._exporter: LogExporter = exporter 

98 self.entity_description: RemoteLoggerDiagnosticEntityDescription = description # pyright: ignore[reportIncompatibleVariableOverride] 

99 self._attr_unique_id = slugify(f"{exporter.name}_{description.key}") 

100 self._attr_device_info = device_info 

101 self._attr_translation_key = description.translation_key 

102 

103 @property 

104 def native_value(self) -> str | int | float | None: # pyright: ignore[reportIncompatibleVariableOverride] 

105 """Return the state.""" 

106 return self.entity_description.value_fn(self._exporter) 

107 

108 @property 

109 def extra_state_attributes(self) -> dict[str, Any]: # pyright: ignore[reportIncompatibleVariableOverride] 

110 """Return the state attributes.""" 

111 return self.entity_description.attr_fn(self._exporter) 

112 

113 

114async def async_setup_entry( 

115 hass: HomeAssistant, 

116 entry: ConfigEntry, 

117 async_add_entities: AddConfigEntryEntitiesCallback, 

118) -> None: 

119 """Set up remote_logger binary sensor from a config entry.""" 

120 exporter: LogExporter = hass.data[DOMAIN][entry.entry_id][REF_EXPORTER] 

121 device_info = DeviceInfo( 

122 entry_type=DeviceEntryType.SERVICE, 

123 identifiers={(DOMAIN, entry.entry_id)}, 

124 manufacturer="Rhizomatics", 

125 name=f"{exporter.name} Remote Logger", 

126 ) 

127 async_add_entities(LoggerEntity(exporter, description, device_info) for description in SENSORS) 

128 

129 # Remove any disabled duplicate devices for this config entry that are not 

130 # the canonical one (identified by entry.entry_id). These can accumulate 

131 # when the integration previously used endpoint-derived identifiers. 

132 registry: dr.DeviceRegistry = dr.async_get(hass) 

133 canonical_id: tuple[str, str] = (DOMAIN, entry.entry_id) 

134 for device in dr.async_entries_for_config_entry(registry, entry.entry_id): 

135 if canonical_id not in device.identifiers and device.disabled: 

136 registry.async_remove_device(device.id)