Coverage for src / rasp_shutter / config.py: 98%

1#!/usr/bin/env python3 

2""" 

3設定ファイルの型定義 

4 

5設計方針: 

6- dataclass で型安全な設定を定義 

7- パスは pathlib.Path で統一 

8- None の使用を最小限に 

9- my_lib の型を直接使用 

10""" 

11 

12from __future__ import annotations 

13 

14import pathlib 

15from dataclasses import dataclass 

16from typing import Any 

17 

18import my_lib.config 

19import my_lib.notify.slack 

20import my_lib.sensor_data 

21import my_lib.webapp.config 

22 

23# my_lib から re-export 

24InfluxDBConfig = my_lib.sensor_data.InfluxDBConfig 

25SlackErrorOnlyConfig = my_lib.notify.slack.SlackErrorOnlyConfig 

26SlackEmptyConfig = my_lib.notify.slack.SlackEmptyConfig 

27SlackErrorConfig = my_lib.notify.slack.SlackErrorConfig 

28SlackChannelConfig = my_lib.notify.slack.SlackChannelConfig 

29 

30# Slack 設定の型エイリアス 

31SlackConfigType = SlackErrorOnlyConfig | SlackEmptyConfig 

32 

33__all__ = [ 

34 "AppConfig", 

35 "InfluxDBConfig", 

36 "LivenessConfig", 

37 "LivenessFileConfig", 

38 "LocationConfig", 

39 "MetricsConfig", 

40 "SensorConfig", 

41 "SensorSpecConfig", 

42 "ShutterConfig", 

43 "ShutterEndpointConfig", 

44 "SlackChannelConfig", 

45 "SlackConfigType", 

46 "SlackEmptyConfig", 

47 "SlackErrorConfig", 

48 "SlackErrorOnlyConfig", 

49 "WebappConfig", 

50 "WebappDataConfig", 

51 "load", 

52 "parse_config", 

53] 

54 

55 

56# === Webapp === 

57@dataclass(frozen=True) 

58class WebappDataConfig: 

59 """webapp.data セクションの設定""" 

60 

61 schedule_file_path: pathlib.Path 

62 log_file_path: pathlib.Path 

63 stat_dir_path: pathlib.Path 

64 

65 

66@dataclass(frozen=True) 

67class WebappConfig: 

68 """webapp セクションの設定""" 

69 

70 static_dir_path: pathlib.Path 

71 data: WebappDataConfig 

72 

73 

74# === Sensor === 

75@dataclass(frozen=True) 

76class SensorSpecConfig: 

77 """センサー指定""" 

78 

79 name: str 

80 measure: str 

81 hostname: str 

82 

83 

84@dataclass(frozen=True) 

85class SensorConfig: 

86 """sensor セクションの設定""" 

87 

88 influxdb: InfluxDBConfig 

89 lux: SensorSpecConfig 

90 solar_rad: SensorSpecConfig 

91 

92 

93# === Location === 

94@dataclass(frozen=True) 

95class LocationConfig: 

96 """location セクションの設定""" 

97 

98 latitude: float 

99 longitude: float 

100 

101 

102# === Metrics === 

103@dataclass(frozen=True) 

104class MetricsConfig: 

105 """metrics セクションの設定""" 

106 

107 data: pathlib.Path 

108 

109 

110# === Liveness === 

111@dataclass(frozen=True) 

112class LivenessFileConfig: 

113 """Liveness ファイルパス設定""" 

114 

115 scheduler: pathlib.Path 

116 

117 

118@dataclass(frozen=True) 

119class LivenessConfig: 

120 """liveness セクションの設定""" 

121 

122 file: LivenessFileConfig 

123 

124 

125# === Shutter === 

126@dataclass(frozen=True) 

127class ShutterEndpointConfig: 

128 """シャッターエンドポイント設定""" 

129 

130 open: str 

131 close: str 

132 

133 

134@dataclass(frozen=True) 

135class ShutterConfig: 

136 """シャッター設定""" 

137 

138 name: str 

139 endpoint: ShutterEndpointConfig 

140 

141 

142# === メイン設定クラス === 

143@dataclass(frozen=True) 

144class AppConfig: 

145 """アプリケーション設定""" 

146 

147 webapp: WebappConfig 

148 sensor: SensorConfig 

149 location: LocationConfig 

150 metrics: MetricsConfig 

151 liveness: LivenessConfig 

152 shutter: list[ShutterConfig] 

153 slack: SlackConfigType 

154 

155 

156# === パース関数 === 

157def _parse_webapp_data(data: dict[str, Any]) -> WebappDataConfig: 

158 return WebappDataConfig( 

159 schedule_file_path=pathlib.Path(data["schedule_file_path"]).resolve(), 

160 log_file_path=pathlib.Path(data["log_file_path"]).resolve(), 

161 stat_dir_path=pathlib.Path(data["stat_dir_path"]).resolve(), 

162 ) 

163 

164 

165def _parse_webapp(data: dict[str, Any]) -> WebappConfig: 

166 return WebappConfig( 

167 static_dir_path=pathlib.Path(data["static_dir_path"]).resolve(), 

168 data=_parse_webapp_data(data["data"]), 

169 ) 

170 

171 

172def _parse_sensor_spec(data: dict[str, Any]) -> SensorSpecConfig: 

173 return SensorSpecConfig( 

174 name=data["name"], 

175 measure=data["measure"], 

176 hostname=data["hostname"], 

177 ) 

178 

179 

180def _parse_influxdb(data: dict[str, Any]) -> InfluxDBConfig: 

181 return InfluxDBConfig( 

182 url=data["url"], 

183 org=data["org"], 

184 token=data["token"], 

185 bucket=data["bucket"], 

186 ) 

187 

188 

189def _parse_sensor(data: dict[str, Any]) -> SensorConfig: 

190 return SensorConfig( 

191 influxdb=_parse_influxdb(data["influxdb"]), 

192 lux=_parse_sensor_spec(data["lux"]), 

193 solar_rad=_parse_sensor_spec(data["solar_rad"]), 

194 ) 

195 

196 

197def _parse_location(data: dict[str, Any]) -> LocationConfig: 

198 return LocationConfig( 

199 latitude=float(data["latitude"]), 

200 longitude=float(data["longitude"]), 

201 ) 

202 

203 

204def _parse_metrics(data: dict[str, Any]) -> MetricsConfig: 

205 return MetricsConfig( 

206 data=pathlib.Path(data["data"]).resolve(), 

207 ) 

208 

209 

210def _parse_slack(data: dict[str, Any] | None) -> SlackConfigType: 

211 """Slack 設定をパースする""" 

212 if data is None or not data.get("bot_token"): 212 ↛ 213line 212 didn't jump to line 213 because the condition on line 212 was never true

213 return SlackEmptyConfig() 

214 

215 return SlackErrorOnlyConfig( 

216 bot_token=data["bot_token"], 

217 from_name=data["from"], 

218 error=SlackErrorConfig( 

219 channel=SlackChannelConfig( 

220 name=data["error"]["channel"]["name"], 

221 id=data["error"]["channel"]["id"], 

222 ), 

223 interval_min=int(data["error"]["interval_min"]), 

224 ), 

225 ) 

226 

227 

228def _parse_liveness_file(data: dict[str, Any]) -> LivenessFileConfig: 

229 return LivenessFileConfig( 

230 scheduler=pathlib.Path(data["scheduler"]).resolve(), 

231 ) 

232 

233 

234def _parse_liveness(data: dict[str, Any]) -> LivenessConfig: 

235 return LivenessConfig( 

236 file=_parse_liveness_file(data["file"]), 

237 ) 

238 

239 

240def _parse_shutter_endpoint(data: dict[str, Any]) -> ShutterEndpointConfig: 

241 return ShutterEndpointConfig( 

242 open=data["open"], 

243 close=data["close"], 

244 ) 

245 

246 

247def _parse_shutter(data: dict[str, Any]) -> ShutterConfig: 

248 return ShutterConfig( 

249 name=data["name"], 

250 endpoint=_parse_shutter_endpoint(data["endpoint"]), 

251 ) 

252 

253 

254def _parse_shutter_list(data: list[dict[str, Any]]) -> list[ShutterConfig]: 

255 return [_parse_shutter(item) for item in data] 

256 

257 

258def parse_config(data: dict[str, Any]) -> AppConfig: 

259 """設定辞書をパースして AppConfig を返す""" 

260 return AppConfig( 

261 webapp=_parse_webapp(data["webapp"]), 

262 sensor=_parse_sensor(data["sensor"]), 

263 location=_parse_location(data["location"]), 

264 metrics=_parse_metrics(data["metrics"]), 

265 liveness=_parse_liveness(data["liveness"]), 

266 shutter=_parse_shutter_list(data["shutter"]), 

267 slack=_parse_slack(data.get("slack")), 

268 ) 

269 

270 

271def to_my_lib_webapp_config(config: AppConfig) -> my_lib.webapp.config.WebappConfig: 

272 """AppConfig から my_lib.webapp.config.WebappConfig を生成""" 

273 return my_lib.webapp.config.WebappConfig( 

274 static_dir_path=config.webapp.static_dir_path, 

275 data=my_lib.webapp.config.WebappDataConfig( 

276 schedule_file_path=config.webapp.data.schedule_file_path, 

277 log_file_path=config.webapp.data.log_file_path, 

278 stat_dir_path=config.webapp.data.stat_dir_path, 

279 ), 

280 ) 

281 

282 

283def load(config_path: str, schema_path: pathlib.Path) -> AppConfig: 

284 """設定ファイルを読み込んで AppConfig を返す""" 

285 raw_config = my_lib.config.load(config_path, schema_path) 

286 return parse_config(raw_config)