Coverage for src/rasp_shutter/control/webapi/control.py: 92%

1#!/usr/bin/env python3

2import dataclasses

3import enum

4import logging

5import pathlib

6import threading

7import typing

9import flask

10import flask_cors

11import my_lib.flask_util

12import my_lib.footprint

13import my_lib.pytest_util

14import my_lib.webapp.config

15import my_lib.webapp.log

16import requests

17from flask_pydantic import validate

19import rasp_shutter.config

20import rasp_shutter.control.config

21import rasp_shutter.control.webapi.sensor

22import rasp_shutter.metrics.collector

23import rasp_shutter.type_defs

24import rasp_shutter.util

25from rasp_shutter.schemas import CtrlLogRequest, ShutterCtrlRequest

28class SHUTTER_STATE(enum.IntEnum):

29 OPEN = 0

30 CLOSE = 1

31 UNKNOWN = 2

34class CONTROL_MODE(enum.Enum):

35 MANUAL = "🔧手動"

36 SCHEDULE = "⏰スケジューラ"

37 AUTO = "🤖自動"

40MODE_TO_STR: dict[CONTROL_MODE, str] = {

41 CONTROL_MODE.MANUAL: "manual",

42 CONTROL_MODE.SCHEDULE: "schedule",

43 CONTROL_MODE.AUTO: "auto",

44}

47class ModeIntervalConfig(typing.NamedTuple):

48 """モード別の制御間隔設定"""

50 divisor: float # diff_secを割る値（60=分単位、3600=時間単位）

51 interval_threshold: float # この値より短い場合は制御をスキップ

52 log_prefix: str # ログメッセージのプレフィックス

55_cfg = rasp_shutter.control.config

56MODE_INTERVAL_CONFIG: dict[CONTROL_MODE, ModeIntervalConfig] = {

57 CONTROL_MODE.MANUAL: ModeIntervalConfig(60, _cfg.EXEC_INTERVAL_MANUAL_MINUTES, ""),

58 CONTROL_MODE.SCHEDULE: ModeIntervalConfig(3600, _cfg.EXEC_INTERVAL_SCHEDULE_HOUR, "スケジュールに従って"),

59 CONTROL_MODE.AUTO: ModeIntervalConfig(3600, _cfg.EXEC_INTERVAL_SCHEDULE_HOUR, "自動で"),

60}

63blueprint = flask.Blueprint("rasp-shutter-control", __name__, url_prefix=my_lib.webapp.config.URL_PREFIX)

65control_lock = threading.Lock()

67# ワーカー固有の制御履歴（pytest-xdist並列実行対応）

68_cmd_hist: dict[str, list[dict]] = {}

71def _get_cmd_hist() -> list[dict]:

72 """ワーカー固有の制御履歴リストを取得"""

73 worker_id = my_lib.pytest_util.get_worker_id()

74 if worker_id not in _cmd_hist: 74 ↛ 75line 74 didn't jump to line 75 because the condition on line 74 was never true

75 _cmd_hist[worker_id] = []

76 return _cmd_hist[worker_id]

79def _clear_cmd_hist() -> None:

80 """ワーカー固有の制御履歴をクリア"""

81 worker_id = my_lib.pytest_util.get_worker_id()

82 _cmd_hist[worker_id] = []

85def init() -> None:

86 _clear_cmd_hist()

89# 公開API: 制御履歴の取得・クリア用

90class _CmdHistWrapper:

91 """ワーカー固有の制御履歴へのアクセスを提供するラッパークラス"""

93 def __iter__(self):

94 return iter(_get_cmd_hist())

96 def __len__(self) -> int:

97 return len(_get_cmd_hist())

99 def __getitem__(self, key: int) -> dict:

100 return _get_cmd_hist()[key]

101

102 def append(self, item: dict) -> None:

103 _get_cmd_hist().append(item)

104

105 def clear(self) -> None:

106 _clear_cmd_hist()

107

108 def copy(self) -> list[dict]:

109 return _get_cmd_hist().copy()

110

111

112cmd_hist = _CmdHistWrapper()

113

114

115def time_str(time_val: float) -> str:

116 """秒数を人間が読みやすい形式に変換"""

117 if time_val >= 3600:

118 hours, remainder = divmod(int(time_val), 3600)

119 minutes = remainder // 60

120 if minutes > 0:

121 return f"{hours}時間{minutes}分"

122 return f"{hours}時間"

123 elif time_val >= 60:

124 minutes, seconds = divmod(int(time_val), 60)

125 if seconds > 0:

126 return f"{minutes}分{seconds}秒"

127 return f"{minutes}分"

128 return f"{int(time_val)}秒"

129

130

131def call_shutter_api(config: rasp_shutter.config.AppConfig, index: int, state: str) -> bool:

132 cmd_hist_push(

133 {

134 "index": index,

135 "state": state,

136 }

137 )

138

139 if rasp_shutter.util.is_dummy_mode():

140 return True

141

142 result = True

143 shutter = config.shutter[index]

144

145 endpoint = shutter.endpoint.open if state == "open" else shutter.endpoint.close

146 logging.debug("Request %s", endpoint)

147 if requests.get(endpoint, timeout=5).status_code != 200:

148 result = False

149

150 return result

151

152

153def exec_stat_file(state: str, index: int) -> pathlib.Path:

154 return rasp_shutter.control.config.get_exec_stat_path(state, index)

155

156

157def clean_stat_exec(config: rasp_shutter.config.AppConfig) -> None:

158 for index in range(len(config.shutter)):

159 my_lib.footprint.clear(exec_stat_file("open", index))

160 my_lib.footprint.clear(exec_stat_file("close", index))

161

162 my_lib.footprint.clear(rasp_shutter.control.config.STAT_PENDING_OPEN.to_path())

163 my_lib.footprint.clear(rasp_shutter.control.config.STAT_AUTO_CLOSE.to_path())

164

165

166def get_shutter_state(config: rasp_shutter.config.AppConfig) -> rasp_shutter.type_defs.ShutterStateResponse:

167 state_list: list[rasp_shutter.type_defs.ShutterStateEntry] = []

168 for index, shutter in enumerate(config.shutter):

169 exec_stat_open = exec_stat_file("open", index)

170 exec_stat_close = exec_stat_file("close", index)

171

172 if my_lib.footprint.exists(exec_stat_open):

173 if my_lib.footprint.exists(exec_stat_close):

174 if my_lib.footprint.compare(exec_stat_open, exec_stat_close):

175 state = SHUTTER_STATE.OPEN

176 else:

177 state = SHUTTER_STATE.CLOSE

178 else:

179 state = SHUTTER_STATE.OPEN

180 else:

181 if my_lib.footprint.exists(exec_stat_close): # noqa: SIM108

182 state = SHUTTER_STATE.CLOSE

183 else:

184 state = SHUTTER_STATE.UNKNOWN

185

186 state_list.append(rasp_shutter.type_defs.ShutterStateEntry(name=shutter.name, state=state))

187

188 return rasp_shutter.type_defs.ShutterStateResponse(state=state_list, result="success")

189

190

191def set_shutter_state_impl(

192 config: rasp_shutter.config.AppConfig,

193 index: int,

194 state: str,

195 mode: CONTROL_MODE,

196 sense_data: rasp_shutter.type_defs.SensorData | None,

197 user: str,

198) -> None:

199 # NOTE: 閉じている場合に再度閉じるボタンをおしたり、逆に開いている場合に再度

200 # 開くボタンを押すことが続くと、スイッチがエラーになるので exec_hist を使って

201 # 防止する。また、明るさに基づく自動の開閉が連続するのを防止する。

202 # exec_hist はこれ以外の目的で使わない。

203 exec_hist = exec_stat_file(state, index)

204 diff_sec = my_lib.footprint.elapsed(exec_hist)

205

206 shutter_name = config.shutter[index].name

207

208 # NOTE: 制御間隔が短く、実際には制御できなかった場合、ログを残す。

209 state_text = rasp_shutter.type_defs.state_to_action_text(state)

210 time_diff_str = time_str(diff_sec)

211 by_text = f"(by {user})" if user != "" else ""

212

213 # NamedTupleで制御間隔チェック

214 interval_config = MODE_INTERVAL_CONFIG[mode]

215 if (diff_sec / interval_config.divisor) < interval_config.interval_threshold:

216 my_lib.webapp.log.info(

217 f"🔔 {interval_config.log_prefix}{shutter_name}のシャッターを{state_text}るのを見合わせました。"

218 f"{time_diff_str}前に{state_text}ています。{by_text}"

219 )

220 return

221

222 result = call_shutter_api(config, index, state)

223

224 my_lib.footprint.update(exec_hist)

225 exec_inv_hist = exec_stat_file("close" if state == "open" else "open", index)

226 my_lib.footprint.clear(exec_inv_hist)

227

228 sensor_text_str = sensor_text(sense_data)

229 by_newline_text = f"\n(by {user})" if user != "" else ""

230

231 if result:

232 my_lib.webapp.log.info(

233 f"{shutter_name}のシャッターを{mode.value}で{state_text}ました。{sensor_text_str}{by_newline_text}"

234 )

235

236 # メトリクス収集

237 try:

238 mode_str = MODE_TO_STR[mode]

239 metrics_data_path = config.metrics.data

240 rasp_shutter.metrics.collector.record_shutter_operation(

241 state, mode_str, metrics_data_path, sense_data

242 )

243 except Exception as e:

244 logging.warning("メトリクス記録に失敗しました: %s", e)

245 else:

246 my_lib.webapp.log.error(

247 f"{shutter_name}のシャッターを{mode.value}で{state_text}るのに失敗しました。"

248 f"{sensor_text_str}{by_newline_text}"

249 )

250

251 # 失敗メトリクス収集

252 try:

253 metrics_data_path = config.metrics.data

254 rasp_shutter.metrics.collector.record_failure(metrics_data_path)

255 except Exception as e:

256 logging.warning("失敗メトリクス記録に失敗しました: %s", e)

257

258

259def set_shutter_state(

260 config: rasp_shutter.config.AppConfig,

261 index_list: list[int],

262 state: str,

263 mode: CONTROL_MODE,

264 sense_data: rasp_shutter.type_defs.SensorData | None,

265 user: str = "",

266) -> rasp_shutter.type_defs.ShutterStateResponse:

267 logging.debug(

268 "set_shutter_state index=[%s], state=%s, mode=%s", ",".join(str(n) for n in index_list), state, mode

269 )

270

271 if state == "open":

272 if mode != CONTROL_MODE.MANUAL:

273 # NOTE: 手動以外でシャッターを開けた場合は、

274 # 自動で閉じた履歴を削除する。

275 my_lib.footprint.clear(rasp_shutter.control.config.STAT_AUTO_CLOSE.to_path())

276 else:

277 # NOTE: シャッターを閉じる指示がされた場合は、

278 # 暗くて延期されていた開ける制御を取り消す。

279 my_lib.footprint.clear(rasp_shutter.control.config.STAT_PENDING_OPEN.to_path())

280

281 with control_lock:

282 for index in index_list:

283 try:

284 set_shutter_state_impl(config, index, state, mode, sense_data, user)

285 except Exception:

286 logging.exception("Failed to control shutter (index=%d)", index)

287 continue

288

289 return get_shutter_state(config)

290

291

292def sensor_text(sense_data: rasp_shutter.type_defs.SensorData | None) -> str:

293 if sense_data is None:

294 return ""

295 else:

296 solar_rad = sense_data.solar_rad.value

297 lux = sense_data.lux.value

298 altitude = sense_data.altitude.value

299 return f"(日射: {solar_rad:.1f} W/m^2, 照度: {lux:.1f} LUX, 高度: {altitude:.1f})"

300

301

302# NOTE: テスト用のコード

303def cmd_hist_push(cmd: dict) -> None: # pragma: no cover

304 hist = _get_cmd_hist()

305 hist.append(cmd)

306 if len(hist) > 20:

307 hist.pop(0)

308

309

310@blueprint.route("/api/shutter_ctrl", methods=["GET", "POST"])

311@flask_cors.cross_origin()

312@validate(query=ShutterCtrlRequest)

313def api_shutter_ctrl(query: ShutterCtrlRequest) -> flask.Response:

314 config: rasp_shutter.config.AppConfig = flask.current_app.config["CONFIG"]

315

316 # NOTE: シャッターが指定されていない場合は、全てを制御対象にする

317 index_list = list(range(len(config.shutter))) if query.index == -1 else [query.index]

318

319 sense_data = rasp_shutter.control.webapi.sensor.get_sensor_data(config)

320

321 if query.cmd == 1:

322 result = set_shutter_state(

323 config,

324 index_list,

325 query.state,

326 CONTROL_MODE.MANUAL,

327 sense_data,

328 my_lib.flask_util.auth_user(flask.request),

329 )

330 return flask.jsonify(dict({"cmd": "set"}, **dataclasses.asdict(result)))

331 else:

332 return flask.jsonify(dict({"cmd": "get"}, **dataclasses.asdict(get_shutter_state(config))))

333

334

335# NOTE: テスト用

336@blueprint.route("/api/ctrl/log", methods=["GET"])

337@flask_cors.cross_origin()

338@validate(query=CtrlLogRequest)

339def api_shutter_ctrl_log(query: CtrlLogRequest) -> flask.Response:

340 if query.cmd == "clear":

341 _clear_cmd_hist()

342 return flask.jsonify(

343 {

344 "result": "success",

345 }

346 )

347 else:

348 return flask.jsonify({"result": "success", "log": _get_cmd_hist()})

349

350

351@blueprint.route("/api/shutter_list", methods=["GET"])

352@flask_cors.cross_origin()

353def api_shutter_list() -> flask.Response:

354 config: rasp_shutter.config.AppConfig = flask.current_app.config["CONFIG"]

355

356 return flask.jsonify([shutter.name for shutter in config.shutter])

357

358

359if rasp_shutter.util.is_dummy_mode(): 359 ↛ exitline 359 didn't exit the module because the condition on line 359 was always true

360

361 @blueprint.route("/api/dummy/open", methods=["GET"])

362 @my_lib.flask_util.support_jsonp

363 def api_dummy_open() -> flask.Response:

364 logging.info("ダミーのシャッターが開きました。")

365 return flask.jsonify({"status": "OK"})

366

367 @blueprint.route("/api/dummy/close", methods=["GET"])

368 @my_lib.flask_util.support_jsonp

369 def api_dummy_close() -> flask.Response:

370 logging.info("ダミーのシャッターが閉じました。")

371 return flask.jsonify({"status": "OK"})

372

373 @blueprint.route("/api/ctrl/clear", methods=["POST"])

374 @my_lib.flask_util.support_jsonp

375 def api_test_control_clear() -> flask.Response:

376 """テスト用: 制御履歴をクリア"""

377 config: rasp_shutter.config.AppConfig = flask.current_app.config["CONFIG"]

378 clean_stat_exec(config)

379 return flask.jsonify({"status": "OK"})

Coverage for src / rasp_shutter / control / webapi / control.py: 92%

205 statements