Coverage for core / src / sensorkit / common / condition.py: 96%

1# SPDX-License-Identifier: Apache-2.0 

2"""Condition types for threshold and change-detection logic over value streams.""" 

3 

4from __future__ import annotations 

5 

6from abc import ABC, abstractmethod 

7from typing import Literal, override 

8 

9from pydantic import BaseModel 

10 

11 

12def resolve_field(obj: object, field_path: str) -> object: 

13 """Resolve a dot-separated field path on an object or dict.""" 

14 value = obj 

15 for part in field_path.split("."): 

16 if isinstance(value, dict): 

17 value = value.get(part) 

18 else: 

19 value = getattr(value, part, None) 

20 if value is None: 

21 return None 

22 return value 

23 

24 

25def _coerce_to_threshold_type(value: object, threshold: float | str | bool | None) -> object: 

26 """Coerce a value to the threshold's type for comparison.""" 

27 if threshold is None: 

28 return value 

29 if isinstance(threshold, bool): 

30 if isinstance(value, bool): 

31 return value 

32 if value is None: 

33 return None 

34 return bool(value) 

35 if isinstance(threshold, (int, float)): 

36 try: 

37 return float(value) 

38 except (TypeError, ValueError): 

39 return None 

40 return value 

41 

42 

43class Condition(BaseModel, ABC): 

44 """Abstract condition evaluated against a value stream. Discriminated on `kind`.""" 

45 

46 kind: str 

47 

48 @abstractmethod 

49 def evaluate( 

50 self, 

51 current: object, 

52 previous: object, 

53 was_active: bool, 

54 ) -> tuple[bool, bool]: 

55 """Evaluate the condition. 

56 

57 Returns: 

58 A `(should_notify, is_active)` tuple. 

59 

60 *should_notify* -- `True` if a notification should be sent this 

61 tick. 

62 

63 *is_active* -- `True` if the condition is currently in an "active" 

64 zone (used for deadband/hysteresis tracking). For conditions 

65 without deadband the two values are identical. 

66 """ 

67 ... 

68 

69 

70class ChangesCondition(Condition): 

71 """Fires on any value change.""" 

72 

73 kind: Literal["changes"] = "changes" 

74 

75 @override 

76 def evaluate(self, current, previous, was_active): 

77 """Return `(True, True)` when `current != previous`, otherwise `(False, False)`.""" 

78 fired = current != previous 

79 return (fired, fired) 

80 

81 

82class AboveCondition(Condition): 

83 """Fires every update while value > threshold. 

84 

85 With *deadband*: remains active until value < threshold - deadband. 

86 """ 

87 

88 kind: Literal["above"] = "above" 

89 threshold: float 

90 deadband: float = 0.0 

91 

92 @override 

93 def evaluate(self, current, previous, was_active): 

94 """Fire while the numeric value exceeds the threshold, applying deadband hysteresis.""" 

95 try: 

96 v = float(current) 

97 except (TypeError, ValueError): 

98 return (False, False) 

99 if was_active: 

100 still = v >= self.threshold - self.deadband 

101 return (still, still) 

102 newly = v > self.threshold 

103 return (newly, newly) 

104 

105 

106class BelowCondition(Condition): 

107 """Fires every update while value < threshold. 

108 

109 With *deadband*: remains active until value > threshold + deadband. 

110 """ 

111 

112 kind: Literal["below"] = "below" 

113 threshold: float 

114 deadband: float = 0.0 

115 

116 @override 

117 def evaluate(self, current, previous, was_active): 

118 """Fire while the numeric value is below the threshold, applying deadband hysteresis.""" 

119 try: 

120 v = float(current) 

121 except (TypeError, ValueError): 

122 return (False, False) 

123 if was_active: 

124 still = v <= self.threshold + self.deadband 

125 return (still, still) 

126 newly = v < self.threshold 

127 return (newly, newly) 

128 

129 

130class EqualsCondition(Condition): 

131 """Fires every update while value equals threshold.""" 

132 

133 kind: Literal["equals"] = "equals" 

134 threshold: float | str | bool 

135 

136 @override 

137 def evaluate(self, current, previous, was_active): 

138 """Fire on every update where current (coerced to the threshold type) equals the threshold.""" 

139 matches = _coerce_to_threshold_type(current, self.threshold) == self.threshold 

140 return (matches, matches) 

141 

142 

143class BecomesCondition(Condition): 

144 """Fires once when value transitions to equal threshold.""" 

145 

146 kind: Literal["becomes"] = "becomes" 

147 threshold: float | str | bool | None 

148 

149 @override 

150 def evaluate(self, current, previous, was_active): 

151 """Fire exactly once when the value transitions from not-equal to equal to the threshold.""" 

152 coerced_cur = _coerce_to_threshold_type(current, self.threshold) 

153 coerced_prev = _coerce_to_threshold_type(previous, self.threshold) 

154 fired = coerced_cur == self.threshold and coerced_prev != self.threshold 

155 return (fired, fired) 

156 

157 

158class CrossesAboveCondition(Condition): 

159 """Fires once on upward crossing. 

160 

161 With *deadband*: won't re-fire until value drops below 

162 threshold - deadband and crosses back above. 

163 """ 

164 

165 kind: Literal["crosses_above"] = "crosses_above" 

166 threshold: float 

167 deadband: float = 0.0 

168 

169 @override 

170 def evaluate(self, current, previous, was_active): 

171 """Fire once when value crosses upward through the threshold; deadband prevents re-firing.""" 

172 try: 

173 v = float(current) 

174 p = float(previous) 

175 except (TypeError, ValueError): 

176 return (False, False) 

177 if was_active: 

178 still = v >= self.threshold - self.deadband 

179 return (False, still) 

180 crossed = v > self.threshold and p <= self.threshold 

181 return (crossed, crossed) 

182 

183 

184class CrossesBelowCondition(Condition): 

185 """Fires once on downward crossing. 

186 

187 With *deadband*: won't re-fire until value rises above 

188 threshold + deadband and crosses back below. 

189 """ 

190 

191 kind: Literal["crosses_below"] = "crosses_below" 

192 threshold: float 

193 deadband: float = 0.0 

194 

195 @override 

196 def evaluate(self, current, previous, was_active): 

197 """Fire once when value crosses downward through the threshold; deadband prevents re-firing.""" 

198 try: 

199 v = float(current) 

200 p = float(previous) 

201 except (TypeError, ValueError): 

202 return (False, False) 

203 if was_active: 

204 still = v <= self.threshold + self.deadband 

205 return (False, still) 

206 crossed = v < self.threshold and p >= self.threshold 

207 return (crossed, crossed) 

208 

209 

210type AnyCondition = ( 

211 ChangesCondition 

212 | AboveCondition 

213 | BelowCondition 

214 | EqualsCondition 

215 | BecomesCondition 

216 | CrossesAboveCondition 

217 | CrossesBelowCondition 

218)