Coverage for core / src / sensorkit / std / weather.py: 98%

1# SPDX-License-Identifier: Apache-2.0 

2from __future__ import annotations 

3 

4import functools 

5from typing import TYPE_CHECKING, Literal, TypedDict, override 

6 

7from pydantic import BaseModel 

8 

9import sensorkit.api as sk 

10from sensorkit.auto.constraint import Constraint, ConstraintEvaluator 

11from sensorkit.common.keyword import validate_keyword_json 

12from sensorkit.core.client import SensorKit 

13 

14if TYPE_CHECKING: 

15 import astropy.units as u 

16 

17 

18class RefractionArgs(TypedDict): 

19 temperature: u.Quantity | None 

20 relative_humidity: u.Quantity | None 

21 pressure: u.Quantity | None 

22 

23 

24@sk.declare_keyword 

25class BasicWeather(BaseModel): 

26 """Ambient weather conditions keyword, with all fields optional. 

27 

28 Units follow the ASCOM IObservingConditions convention. 

29 

30 Attributes: 

31 temperature: Ambient air temperature, °C. 

32 humidity: Relative humidity, percent (0-100). 

33 pressure: Barometric pressure, hPa, absolute at the observatory altitude rather 

34 than corrected to sea level. 

35 cloud_cover: Sky covered by cloud, percent (0-100). 

36 dew_point: Dew point temperature, °C. 

37 rain_rate: Rainfall intensity, mm/h. 

38 wind_direction: Direction the wind blows from, degrees clockwise from true north 

39 (0-360). 

40 wind_speed: Wind speed, m/s. 

41 """ 

42 temperature: float | None = None 

43 humidity: float | None = None 

44 pressure: float | None = None 

45 cloud_cover: float | None = None 

46 dew_point: float | None = None 

47 rain_rate: float | None = None 

48 wind_direction: float | None = None 

49 wind_speed: float | None = None 

50 

51 def to_astropy(self) -> RefractionArgs: 

52 """Return the refraction arguments accepted by astropy's AltAz frame. 

53 

54 Keys match the AltAz frame attributes, so the result can be splatted straight 

55 into AltAz or into SkyCoord with an AltAz frame. Astropy normalizes the percent 

56 humidity to the 0-1 fraction it works in. 

57 """ 

58 import astropy.units as u 

59 

60 return RefractionArgs( 

61 temperature=self.temperature * u.deg_C if self.temperature is not None else None, 

62 relative_humidity=self.humidity * u.percent if self.humidity is not None else None, 

63 pressure=self.pressure * u.hPa if self.pressure is not None else None, 

64 ) 

65 

66 def get_fits_cards(self): 

67 if self.temperature is not None: 

68 yield "SKWXTEMP", (self.temperature, "Ambient air temperature [C]") 

69 if self.humidity is not None: 

70 yield "SKWXHUM", (self.humidity, "Relative humidity [%]") 

71 if self.pressure is not None: 

72 yield "SKWXPRES", (self.pressure, "Barometric pressure [hPa]") 

73 if self.cloud_cover is not None: 

74 yield "SKWXCLD", (self.cloud_cover, "Cloud cover [%]") 

75 if self.dew_point is not None: 

76 yield "SKWXDEW", (self.dew_point, "Dew point [C]") 

77 if self.rain_rate is not None: 

78 yield "SKWXRAIN", (self.rain_rate, "Rain rate [mm/h]") 

79 if self.wind_speed is not None: 

80 yield "SKWXWSPD", (self.wind_speed, "Wind speed [m/s]") 

81 if self.wind_direction is not None: 

82 yield "SKWXWDIR", (self.wind_direction, "Wind direction [deg]") 

83 

84 

85WeatherProvider = sk.declare_trait( 

86 "WeatherProvider", 

87 required_keywords=("BasicWeather",), 

88) 

89 

90StandardWeather = sk.declare_archetype( 

91 "weather", 

92 required_traits=(WeatherProvider,), 

93) 

94"""Standard archetype for ambient weather telemetry providers.""" 

95 

96 

97class WeatherFieldEvaluator: 

98 

99 def __init__(self, name: str, threshold: float, deadband: float): 

100 self.name = name 

101 self.threshold = threshold 

102 self.deadband = deadband 

103 self._exceeded = False 

104 

105 def eval_threshold(self, weather: BasicWeather) -> float: 

106 value: float | None = getattr(weather, self.name, None) 

107 

108 if value is None: 

109 return float("inf") 

110 

111 over = value - self.threshold 

112 over_deadband = over + self.deadband 

113 self._exceeded = (over_deadband if self._exceeded else over) > 0 

114 

115 return over_deadband if self._exceeded else over 

116 

117 

118class WeatherConstraint(Constraint): 

119 """Constraint that monitors a weather provider and activates when conditions exceed thresholds. 

120 

121 Thresholds and deadbands are compared directly against the matching BasicWeather field 

122 and so carry its units: humidity in percent, wind speed in m/s, rain rate in mm/h. 

123 

124 Attributes: 

125 provider: Entity name of the weather provider to consume BasicWeather from. 

126 humidity_max: Relative humidity ceiling, percent. 

127 humidity_deadband: Percent below humidity_max the reading must fall to clear. 

128 wind_max: Wind speed ceiling, m/s. 

129 wind_deadband: m/s below wind_max the reading must fall to clear. 

130 rain_max: Rainfall intensity ceiling, mm/h. 

131 rain_deadband: mm/h below rain_max the reading must fall to clear. 

132 """ 

133 

134 kind: Literal["weather"] = "weather" 

135 provider: str 

136 humidity_max: float | None = None 

137 humidity_deadband: float = 0.0 

138 wind_max: float | None = None 

139 wind_deadband: float = 0.0 

140 rain_max: float | None = None 

141 rain_deadband: float = 0.0 

142 

143 def _get_field_evaluators(self): 

144 if self.humidity_max is not None: 

145 yield WeatherFieldEvaluator("humidity", self.humidity_max, self.humidity_deadband) 

146 

147 if self.wind_max is not None: 

148 yield WeatherFieldEvaluator("wind_speed", self.wind_max, self.wind_deadband) 

149 

150 if self.rain_max is not None: 

151 yield WeatherFieldEvaluator("rain_rate", self.rain_max, self.rain_deadband) 

152 

153 @functools.cached_property 

154 def _field_evaluators(self) -> tuple[WeatherFieldEvaluator, ...]: 

155 return tuple(self._get_field_evaluators()) 

156 

157 def check_weather(self, weather: BasicWeather) -> list[str]: 

158 errors = [] 

159 

160 for field in self._field_evaluators: 

161 delta = field.eval_threshold(weather) 

162 

163 if delta == float("inf"): 

164 errors.append(f"{field.name} data missing") 

165 elif delta >= 0: 

166 errors.append(f"{field.name} is {delta:.1f} too high") 

167 

168 return errors 

169 

170 @override 

171 async def check_task(self, evaluator: ConstraintEvaluator, kit: SensorKit): 

172 provider = kit.entity(self.provider) 

173 consumer = await provider._stream.consume("BasicWeather") 

174 

175 async for msg in consumer: 

176 try: 

177 weather = validate_keyword_json("BasicWeather", msg.data) 

178 except Exception: 

179 continue 

180 

181 errors = self.check_weather(weather) 

182 

183 if errors: 

184 evaluator.constrain(", ".join(errors)) 

185 else: 

186 evaluator.clear() 

187 

188 evaluator.ready()