Coverage for flask/src/rasp_shutter/control/webapi/control.py: 85%

1#!/usr/bin/env python3

2import enum

3import logging

4import os

5import pathlib

6import threading

8import my_lib.flask_util

9import my_lib.footprint

10import my_lib.webapp.config

11import my_lib.webapp.log

12import rasp_shutter.control.config

13import rasp_shutter.metrics.collector

14import requests

16import flask

18# この時間内に同じ制御がスケジューラで再度リクエストされた場合、

19# 実行をやめる。

20EXEC_INTERVAL_SCHEDULE_HOUR = 12

21# この時間内に同じ制御が手動で再度リクエストされた場合、

22# 実行をやめる。

23EXEC_INTERVAL_MANUAL_MINUTES = 1

26class SHUTTER_STATE(enum.IntEnum): # noqa: N801

27 OPEN = 0

28 CLOSE = 1

29 UNKNOWN = 2

32class CONTROL_MODE(enum.Enum): # noqa: N801

33 MANUAL = "🔧手動"

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

35 AUTO = "🤖自動"

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

40control_lock = threading.Lock()

41cmd_hist = []

44def init():

45 global cmd_hist # noqa: PLW0603

46 cmd_hist = []

49def time_str(time_val):

50 if time_val >= (60 * 60):

51 unit = ["分", "時間"]

52 time_val /= 60

53 else:

54 unit = ["秒", "分"]

56 upper = 0

57 if time_val >= 60:

58 upper = int(time_val / 60)

59 time_val -= upper * 60

60 time_val = int(time_val)

62 if upper != 0:

63 if time_val == 0:

64 return f"{upper}{unit[1]}"

65 else:

66 return f"{upper}{unit[1]}{time_val}{unit[0]}"

67 else:

68 return f"{time_val}{unit[0]}"

71def call_shutter_api(config, index, state):

72 cmd_hist_push(

73 {

74 "index": index,

75 "state": state,

76 }

77 )

79 if os.environ.get("DUMMY_MODE", "false") == "true":

80 return True

82 result = True

83 shutter = config["shutter"][index]

85 logging.debug("Request %s", shutter["endpoint"][state])

86 if requests.get(shutter["endpoint"][state], timeout=5).status_code != 200:

87 result = False

89 return result

92def exec_stat_file(state, index):

93 return pathlib.Path(str(rasp_shutter.control.config.STAT_EXEC_TMPL[state]).format(index=index))

96def clean_stat_exec(config):

97 for index in range(len(config["shutter"])):

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

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

100

101 my_lib.footprint.clear(rasp_shutter.control.config.STAT_PENDING_OPEN)

102 my_lib.footprint.clear(rasp_shutter.control.config.STAT_AUTO_CLOSE)

103

104

105def get_shutter_state(config):

106 state_list = []

107 for index, shutter in enumerate(config["shutter"]):

108 shutter_state = {

109 "name": shutter["name"],

110 }

111

112 exec_stat_open = exec_stat_file("open", index)

113 exec_stat_close = exec_stat_file("close", index)

114

115 if my_lib.footprint.exists(exec_stat_open):

116 if my_lib.footprint.exists(exec_stat_close):

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

118 shutter_state["state"] = SHUTTER_STATE.OPEN

119 else:

120 shutter_state["state"] = SHUTTER_STATE.CLOSE

121 else:

122 shutter_state["state"] = SHUTTER_STATE.OPEN

123 else: # noqa: PLR5501

124 if my_lib.footprint.exists(exec_stat_close):

125 shutter_state["state"] = SHUTTER_STATE.CLOSE

126 else:

127 shutter_state["state"] = SHUTTER_STATE.UNKNOWN

128 state_list.append(shutter_state)

129

130 return {

131 "state": state_list,

132 "result": "success",

133 }

134

135

136def set_shutter_state_impl(config, index, state, mode, sense_data, user): # noqa: PLR0913

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

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

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

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

141 exec_hist = exec_stat_file(state, index)

142 diff_sec = my_lib.footprint.elapsed(exec_hist)

143

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

145 if mode == CONTROL_MODE.MANUAL: 145 ↛ 160line 145 didn't jump to line 160 because the condition on line 145 was always true

146 if (diff_sec / 60) < EXEC_INTERVAL_MANUAL_MINUTES:

147 my_lib.webapp.log.info(

148 (

149 "🔔 {name}のシャッターを{state}るのを見合わせました。"

150 "{time_diff_str}前に{state}ています。{by}"

151 ).format(

152 name=config["shutter"][index]["name"],

153 state="開け" if state == "open" else "閉め",

154 time_diff_str=time_str(diff_sec),

155 by=f"(by {user})" if user != "" else "",

156 )

157 )

158 return

159

160 elif mode == CONTROL_MODE.SCHEDULE:

161 if (diff_sec / (60 * 60)) < EXEC_INTERVAL_SCHEDULE_HOUR:

162 my_lib.webapp.log.info(

163 (

164 "🔔 スケジュールに従って{name}のシャッターを{state}るのを見合わせました。"

165 "{time_diff_str}前に{state}ています。{by}"

166 ).format(

167 name=config["shutter"][index]["name"],

168 state="開け" if state == "open" else "閉め",

169 time_diff_str=time_str(diff_sec),

170 by=f"(by {user})" if user != "" else "",

171 )

172 )

173 return

174 elif mode == CONTROL_MODE.AUTO:

175 if (diff_sec / (60 * 60)) < EXEC_INTERVAL_SCHEDULE_HOUR:

176 my_lib.webapp.log.info(

177 (

178 "🔔 自動で{name}のシャッターを{state}るのを見合わせました。"

179 "{time_diff_str}前に{state}ています。{by}"

180 ).format(

181 name=config["shutter"][index]["name"],

182 state="開け" if state == "open" else "閉め",

183 time_diff_str=time_str(diff_sec),

184 by=f"(by {user})" if user != "" else "",

185 )

186 )

187 return

188 else: # pragma: no cover

189 pass

190

191 result = call_shutter_api(config, index, state)

192

193 my_lib.footprint.update(exec_hist)

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

195 my_lib.footprint.clear(exec_inv_hist)

196

197 if result:

198 my_lib.webapp.log.info(

199 "{name}のシャッターを{mode}で{state}ました。{sensor_text}{by}".format(

200 name=config["shutter"][index]["name"],

201 mode=mode.value,

202 state="開け" if state == "open" else "閉め",

203 sensor_text=sensor_text(sense_data) if mode != CONTROL_MODE.MANUAL else "",

204 by=f"\n(by {user})" if user != "" else "",

205 )

206 )

207

208 # メトリクス収集

209 try:

210 mode_str = (

211 "manual"

212 if mode == CONTROL_MODE.MANUAL

213 else "schedule"

214 if mode == CONTROL_MODE.SCHEDULE

215 else "auto"

216 )

217 metrics_data_path = config.get("metrics", {}).get("data")

218 rasp_shutter.metrics.collector.record_shutter_operation(

219 state, mode_str, metrics_data_path, sense_data

220 )

221 except Exception as e:

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

223 else:

224 my_lib.webapp.log.error(

225 "{name}のシャッターを{mode}で{state}るのに失敗しました。{sensor_text}{by}".format(

226 name=config["shutter"][index]["name"],

227 mode=mode.value,

228 state="開け" if state == "open" else "閉め",

229 sensor_text=sensor_text(sense_data) if mode != CONTROL_MODE.MANUAL else "",

230 by=f"\n(by {user})" if user != "" else "",

231 )

232 )

233

234 # 失敗メトリクス収集

235 try:

236 metrics_data_path = config.get("metrics", {}).get("data")

237 rasp_shutter.metrics.collector.record_failure(metrics_data_path)

238 except Exception as e:

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

240

241

242def set_shutter_state(config, index_list, state, mode, sense_data, user=""): # noqa: PLR0913

243 logging.debug(

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

245 )

246

247 if state == "open":

248 if mode != CONTROL_MODE.MANUAL: 248 ↛ 251line 248 didn't jump to line 251 because the condition on line 248 was never true

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

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

251 my_lib.footprint.clear(rasp_shutter.control.config.STAT_AUTO_CLOSE)

252 else:

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

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

255 my_lib.footprint.clear(rasp_shutter.control.config.STAT_PENDING_OPEN)

256

257 with control_lock:

258 for index in index_list:

259 try:

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

261 except Exception: # noqa: PERF203

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

263 continue

264

265 return get_shutter_state(config)

266

267

268def sensor_text(sense_data):

269 if sense_data is None:

270 return ""

271 else:

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

273 solar_rad=sense_data["solar_rad"]["value"],

274 lux=sense_data["lux"]["value"],

275 altitude=sense_data["altitude"]["value"],

276 )

277

278

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

280def cmd_hist_push(cmd): # pragma: no cover

281 global cmd_hist

282

283 cmd_hist.append(cmd)

284 if len(cmd_hist) > 20:

285 cmd_hist.pop(0)

286

287

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

289@my_lib.flask_util.support_jsonp

290def api_shutter_ctrl():

291 cmd = flask.request.args.get("cmd", 0, type=int)

292 index = flask.request.args.get("index", -1, type=int)

293 state = flask.request.args.get("state", "close", type=str)

294 config = flask.current_app.config["CONFIG"]

295

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

297 index_list = list(range(len(config["shutter"]))) if index == -1 else [index]

298

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

300

301 if cmd == 1:

302 return flask.jsonify(

303 dict(

304 {"cmd": "set"},

305 **set_shutter_state(

306 config,

307 index_list,

308 state,

309 CONTROL_MODE.MANUAL,

310 sense_data,

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

312 ),

313 )

314 )

315 else:

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

317

318

319# NOTE: テスト用

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

321@my_lib.flask_util.support_jsonp

322def api_shutter_ctrl_log():

323 global cmd_hist # noqa: PLW0603

324

325 cmd = flask.request.args.get("cmd", "get")

326 if cmd == "clear":

327 cmd_hist = []

328 return flask.jsonify(

329 {

330 "result": "success",

331 }

332 )

333 else:

334 return flask.jsonify({"result": "success", "log": cmd_hist})

335

336

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

338@my_lib.flask_util.support_jsonp

339def api_shutter_list():

340 config = flask.current_app.config["CONFIG"]

341

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

343

344

345if os.environ.get("DUMMY_MODE", "false") == "true": 345 ↛ exitline 345 didn't exit the module because the condition on line 345 was always true

346

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

348 @my_lib.flask_util.support_jsonp

349 def api_dummy_open():

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

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

352

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

354 @my_lib.flask_util.support_jsonp

355 def api_dummy_close():

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

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

358

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

360 @my_lib.flask_util.support_jsonp

361 def api_test_control_clear():

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

363 config = flask.current_app.config["CONFIG"]

364 clean_stat_exec(config)

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