Coverage for flask/src/rasp_shutter/webapp_control.py: 99%

1#!/usr/bin/env python3 

2import enum 

3import logging 

4import os 

5import pathlib 

6import threading 

7 

8import my_lib.flask_util 

9import my_lib.footprint 

10import my_lib.webapp.config 

11import my_lib.webapp.log 

12import rasp_shutter.config 

13import requests 

14 

15import flask 

16 

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

18# 実行をやめる。 

19EXEC_INTERVAL_SCHEDULE_HOUR = 12 

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

21# 実行をやめる。 

22EXEC_INTERVAL_MANUAL_MINUTES = 1 

23 

24 

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

26 OPEN = 0 

27 CLOSE = 1 

28 UNKNOWN = 2 

29 

30 

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

32 MANUAL = "🔧手動" 

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

34 AUTO = "🤖自動" 

35 

36 

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

38 

39control_lock = threading.Lock() 

40cmd_hist = [] 

41 

42 

43def init(): 

44 global cmd_hist # noqa: PLW0603 

45 cmd_hist = [] 

46 

47 

48def time_str(time_val): 

49 if time_val >= (60 * 60): 

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

51 time_val /= 60 

52 else: 

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

54 

55 upper = 0 

56 if time_val >= 60: 

57 upper = int(time_val / 60) 

58 time_val -= upper * 60 

59 time_val = int(time_val) 

60 

61 if upper != 0: 

62 if time_val == 0: 

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

64 else: 

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

66 else: 

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

68 

69 

70def call_shutter_api(config, index, state): 

71 cmd_hist_push( 

72 { 

73 "index": index, 

74 "state": state, 

75 } 

76 ) 

77 

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

79 return True 

80 

81 result = True 

82 shutter = config["shutter"][index] 

83 

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

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

86 result = False 

87 

88 return result 

89 

90 

91def exec_stat_file(state, index): 

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

93 

94 

95def clean_stat_exec(config): 

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

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

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

99 

100 

101def get_shutter_state(config): 

102 state_list = [] 

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

104 shutter_state = { 

105 "name": shutter["name"], 

106 } 

107 

108 exec_stat_open = exec_stat_file("open", index) 

109 exec_stat_close = exec_stat_file("close", index) 

110 

111 if my_lib.footprint.exists(exec_stat_open): 

112 if my_lib.footprint.exists(exec_stat_close): 

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

114 shutter_state["state"] = SHUTTER_STATE.OPEN 

115 else: 

116 shutter_state["state"] = SHUTTER_STATE.CLOSE 

117 else: 

118 shutter_state["state"] = SHUTTER_STATE.OPEN 

119 else: # noqa: PLR5501 

120 if my_lib.footprint.exists(exec_stat_close): 

121 shutter_state["state"] = SHUTTER_STATE.CLOSE 

122 else: 

123 shutter_state["state"] = SHUTTER_STATE.UNKNOWN 

124 state_list.append(shutter_state) 

125 

126 return { 

127 "state": state_list, 

128 "result": "success", 

129 } 

130 

131 

132def set_shutter_state_impl(config, index, state, mode, sense_data=None, user=""): # noqa: PLR0913 

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

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

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

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

137 exec_hist = exec_stat_file(state, index) 

138 diff_sec = my_lib.footprint.elapsed(exec_hist) 

139 

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

141 if mode == CONTROL_MODE.MANUAL: 

142 if (diff_sec / 60) < EXEC_INTERVAL_MANUAL_MINUTES: 

143 my_lib.webapp.log.info( 

144 ( 

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

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

147 ).format( 

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

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

150 time_diff_str=time_str(diff_sec), 

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

152 ) 

153 ) 

154 return 

155 

156 elif mode == CONTROL_MODE.SCHEDULE: 

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

158 my_lib.webapp.log.info( 

159 ( 

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

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

162 ).format( 

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

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

165 time_diff_str=time_str(diff_sec), 

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

167 ) 

168 ) 

169 return 

170 elif mode == CONTROL_MODE.AUTO: 

171 if (diff_sec / (60 * 60)) < EXEC_INTERVAL_SCHEDULE_HOUR: # pragma: no cover 

172 # NOTE: shutter_auto_close の段階で撥ねられているので、ここには来ない。 

173 my_lib.webapp.log.info( 

174 ( 

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

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

177 ).format( 

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

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

180 time_diff_str=time_str(diff_sec), 

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

182 ) 

183 ) 

184 return 

185 else: # pragma: no cover 

186 pass 

187 

188 result = call_shutter_api(config, index, state) 

189 

190 my_lib.footprint.update(exec_hist) 

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

192 my_lib.footprint.clear(exec_inv_hist) 

193 

194 if result: 

195 my_lib.webapp.log.info( 

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

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

198 mode=mode.value, 

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

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

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

202 ) 

203 ) 

204 else: 

205 my_lib.webapp.log.error( 

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

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

208 mode=mode.value, 

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

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

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

212 ) 

213 ) 

214 

215 

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

217 if state == "open": 

218 if mode != CONTROL_MODE.MANUAL: 

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

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

221 my_lib.footprint.clear(rasp_shutter.config.STAT_AUTO_CLOSE) 

222 else: 

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

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

225 my_lib.footprint.clear(rasp_shutter.config.STAT_PENDING_OPEN) 

226 

227 with control_lock: 

228 for index in index_list: 

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

230 

231 return get_shutter_state(config) 

232 

233 

234def sensor_text(sense_data): 

235 if sense_data is None: 235 ↛ 236line 235 didn't jump to line 236 because the condition on line 235 was never true

236 return "" 

237 else: 

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

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

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

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

242 ) 

243 

244 

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

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

247 global cmd_hist 

248 

249 cmd_hist.append(cmd) 

250 if len(cmd_hist) > 20: 

251 cmd_hist.pop(0) 

252 

253 

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

255@my_lib.flask_util.support_jsonp 

256def api_shutter_ctrl(): 

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

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

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

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

261 

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

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

264 

265 if cmd == 1: 

266 return flask.jsonify( 

267 dict( 

268 {"cmd": "set"}, 

269 **set_shutter_state( 

270 config, 

271 index_list, 

272 state, 

273 CONTROL_MODE.MANUAL, 

274 user=my_lib.flask_util.auth_user(flask.request), 

275 ), 

276 ) 

277 ) 

278 else: 

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

280 

281 

282# NOTE: テスト用 

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

284@my_lib.flask_util.support_jsonp 

285def api_shutter_ctrl_log(): 

286 global cmd_hist # noqa: PLW0603 

287 

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

289 if cmd == "clear": 

290 cmd_hist = [] 

291 return flask.jsonify( 

292 { 

293 "result": "success", 

294 } 

295 ) 

296 else: 

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

298 

299 

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

301@my_lib.flask_util.support_jsonp 

302def api_shutter_list(): 

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

304 

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

306 

307 

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

309@my_lib.flask_util.support_jsonp 

310def api_dummy_open(): 

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

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

313 

314 

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

316@my_lib.flask_util.support_jsonp 

317def api_dummy_close(): 

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

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