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

1"""Sensor platform for remote_logger.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from typing import TYPE_CHECKING, Any 

7 

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

9from homeassistant.const import EntityCategory 

10from homeassistant.helpers import device_registry as dr 

11from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo 

12from homeassistant.util import slugify 

13 

14from .const import DOMAIN 

15 

16if TYPE_CHECKING: 

17 from collections.abc import Callable 

18 

19 from homeassistant.core import HomeAssistant 

20 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback 

21 

22 from custom_components.remote_logger.exporter import LogExporter 

23 from custom_components.remote_logger.remote_logger import RemoteLoggerConfigEntry 

24 

25 

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

27class RemoteLoggerDiagnosticEntityDescription(SensorEntityDescription): 

28 """Describes diagnostic sensor entity.""" 

29 

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

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

32 

33 

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

35 RemoteLoggerDiagnosticEntityDescription( 

36 key="format_errors", 

37 translation_key="format_errors", 

38 native_unit_of_measurement="error", 

39 entity_category=EntityCategory.DIAGNOSTIC, 

40 state_class=SensorStateClass.TOTAL_INCREASING, 

41 value_fn=lambda exporter: exporter.format_error_count, 

42 attr_fn=lambda exporter: { 

43 "last_error_time": exporter.last_format_error, 

44 "last_error_message": exporter.last_format_error_message, 

45 }, 

46 ), 

47 RemoteLoggerDiagnosticEntityDescription( 

48 key="posting_errors", 

49 translation_key="posting_errors", 

50 native_unit_of_measurement="error", 

51 entity_category=EntityCategory.DIAGNOSTIC, 

52 state_class=SensorStateClass.TOTAL_INCREASING, 

53 value_fn=lambda exporter: exporter.posting_error_count, 

54 attr_fn=lambda exporter: { 

55 "last_error_time": exporter.last_posting_error, 

56 "last_error_message": exporter.last_posting_error_message, 

57 }, 

58 ), 

59 RemoteLoggerDiagnosticEntityDescription( 

60 key="events", 

61 translation_key="events", 

62 native_unit_of_measurement="event", 

63 entity_category=EntityCategory.DIAGNOSTIC, 

64 state_class=SensorStateClass.TOTAL_INCREASING, 

65 value_fn=lambda exporter: exporter.event_count, 

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

67 ), 

68 RemoteLoggerDiagnosticEntityDescription( 

69 key="postings", 

70 translation_key="postings", 

71 native_unit_of_measurement="posting", 

72 entity_category=EntityCategory.DIAGNOSTIC, 

73 state_class=SensorStateClass.TOTAL_INCREASING, 

74 value_fn=lambda exporter: exporter.posting_count, 

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

76 ), 

77) 

78 

79 

80class LoggerEntity(SensorEntity): 

81 """Represent a diagnostic tracking logger.""" 

82 

83 _attr_should_poll = True 

84 _attr_has_entity_name = True 

85 entity_description: RemoteLoggerDiagnosticEntityDescription # pyright: ignore[reportIncompatibleVariableOverride] 

86 

87 def __init__( 

88 self, 

89 exporter: LogExporter, 

90 description: RemoteLoggerDiagnosticEntityDescription, 

91 device_info: DeviceInfo, 

92 ) -> None: 

93 super().__init__() 

94 self._exporter: LogExporter = exporter 

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

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

97 self._attr_device_info = device_info 

98 

99 @property 

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

101 """Entity state.""" 

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

103 

104 @property 

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

106 """Entity state attributes.""" 

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

108 

109 

110async def async_setup_entry( 

111 hass: HomeAssistant, 

112 entry: RemoteLoggerConfigEntry, 

113 async_add_entities: AddConfigEntryEntitiesCallback, 

114) -> None: 

115 """Set up remote_logger sensors from a config entry.""" 

116 exporter: LogExporter = entry.runtime_data.exporter 

117 device_info = DeviceInfo( 

118 entry_type=DeviceEntryType.SERVICE, 

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

120 manufacturer="Rhizomatics", 

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

122 ) 

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

124 

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

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

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

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

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

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

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

132 registry.async_remove_device(device.id)