Coverage for custom_components/autoarm/helpers.py: 99%

1import datetime as dt 

2import logging 

3import re 

4from typing import TYPE_CHECKING, Any 

5 

6import homeassistant.util.dt as dt_util 

7from homeassistant.components.alarm_control_panel.const import AlarmControlPanelState 

8from homeassistant.core import Context, HomeAssistant, State 

9from homeassistant.helpers.dispatcher import async_dispatcher_send 

10from homeassistant.helpers.json import ExtendedJSONEncoder 

11 

12from .const import SIGNAL_STATUS_UPDATED, ChangeSource 

13 

14if TYPE_CHECKING: 

15 from collections.abc import Callable 

16 

17_LOGGER = logging.getLogger(__name__) 

18 

19 

20def friendly_state(state: str) -> str: 

21 """Alarm state as shown to people, Armed Away rather than armed_away""" 

22 return str(state).replace("_", " ").title() 

23 

24 

25def alarm_state_as_enum(state_str: str | None) -> AlarmControlPanelState | None: 

26 if state_str is None: 

27 return None 

28 try: 

29 return AlarmControlPanelState(state_str) 

30 except ValueError as e: 

31 _LOGGER.warning("AUTOARM Invalid alarm state: %s", e) 

32 return None 

33 

34 

35def change_source_as_enum(source_str: str | None) -> ChangeSource | None: 

36 if source_str is None: 

37 return None 

38 try: 

39 return ChangeSource(source_str) 

40 except ValueError as e: 

41 _LOGGER.warning("AUTOARM Invalid change source: %s", e) 

42 return None 

43 

44 

45def child_context(parent: Context | None) -> Context: 

46 """Context for AutoArm's reaction to an event or action, linked back to what caused it""" 

47 if parent is None: 

48 return Context() 

49 return Context(user_id=parent.user_id, parent_id=parent.id) 

50 

51 

52def safe_state(state: State | None) -> str | None: 

53 try: 

54 return state.state if state is not None else None 

55 except Exception as e: 

56 _LOGGER.debug("AUTOARM Failed to load state %s: %s", state, e) 

57 return None 

58 

59 

60class Limiter: 

61 """Rate limiting tracker""" 

62 

63 def __init__(self, window: dt.timedelta, max_calls: int = 4) -> None: 

64 self.calls: list[dt.datetime] = [] 

65 self.window: dt.timedelta = window 

66 self.max_calls: int = max_calls 

67 _LOGGER.debug( 

68 "AUTOARM Rate limiter initialized with window %s and max_calls %s", 

69 window, 

70 max_calls, 

71 ) 

72 

73 def triggered(self) -> bool: 

74 """Register a call and check if window based rate limit triggered""" 

75 cut_off: dt.datetime = dt_util.now() - self.window 

76 self.calls.append(dt_util.now()) 

77 in_scope = 0 

78 

79 for call in self.calls[:]: 

80 if call >= cut_off: 

81 in_scope += 1 

82 else: 

83 self.calls.remove(call) 

84 

85 return in_scope > self.max_calls 

86 

87 

88def deobjectify(obj: object) -> dict[Any, Any] | str | int | float | bool | None: 

89 if obj is None or isinstance(obj, (str, int, float, bool)): 

90 return obj 

91 if isinstance(obj, (dt.datetime, dt.time, dt.date)): 

92 return obj.isoformat() 

93 as_dict: Callable[[], dict[Any, Any]] | None = getattr(obj, "as_dict", None) 

94 if as_dict is None: 

95 return str(obj) 

96 return as_dict() 

97 

98 

99class AppHealthTracker: 

100 def __init__(self, hass: HomeAssistant) -> None: 

101 self.hass = hass 

102 self.initialized = False 

103 self.initialization_errors: dict[str, int] = {} 

104 self.failures = 0 

105 

106 def app_initialized(self) -> None: 

107 self.initialized = True 

108 async_dispatcher_send(self.hass, SIGNAL_STATUS_UPDATED) 

109 

110 def record_initialization_error(self, stage: str) -> None: 

111 self.initialization_errors.setdefault(stage, 0) 

112 self.initialization_errors[stage] += 1 

113 self.failures += 1 

114 async_dispatcher_send(self.hass, SIGNAL_STATUS_UPDATED) 

115 

116 def record_runtime_error(self) -> None: 

117 self.failures += 1 

118 async_dispatcher_send(self.hass, SIGNAL_STATUS_UPDATED) 

119 

120 

121class ExtendedExtendedJSONEncoder(ExtendedJSONEncoder): 

122 def default(self, o: Any) -> Any: 

123 if isinstance(o, dt.time): 

124 return o.isoformat() 

125 if isinstance(o, dt.timedelta): 

126 td: dt.timedelta = o 

127 return f"{td.seconds} seconds" 

128 if isinstance(o, re.Pattern): 

129 return {o.pattern} 

130 return super().default(o)