Coverage for flask/src/rasp_water/valve.py: 98%

1#!/usr/bin/env python3

2import inspect

3import logging

4import os

5import pathlib

6import threading

7import time

8import traceback

9from builtins import open as valve_open

10from enum import IntEnum

12import my_lib.footprint

13import my_lib.rpi

14import my_lib.webapp.config

16# バルブを一定期間開く際に作られるファイル．

17# ファイルの内容はバルブを閉じるべき UNIX 時間．

18STAT_PATH_VALVE_CONTROL_COMMAND = None

20# 実際にバルブを開いた際に作られるファイル．

21# 実際にバルブを閉じた際に削除される．

22STAT_PATH_VALVE_OPEN = None

24# 実際にバルブを閉じた際に作られるファイル．

25# 実際にバルブを開いた際に削除される．

26STAT_PATH_VALVE_CLOSE = None

28# 電磁弁制御用の GPIO 端子番号．

29# この端子が H になった場合に，水が出るように回路を組んでおく．

30GPIO_PIN_DEFAULT = 18

33# 流量計の A/D 値が 5V の時の流量

34FLOW_SCALE_MAX = 12

36# 異常とみなす流量

37FLOW_ERROR_TH = 20

39# 流量計をモニタする ADC の設定 (ADS1015 のドライバ ti_ads1015 が公開)

40ADC_SCALE_PATH = "/sys/bus/iio/devices/iio:device0/in_voltage0_scale"

41ADC_SCALE_VALUE = 3

43# 流量計のアナログ出力値 (ADS1015 のドライバ ti_ads1015 が公開)

44ADC_VALUE_PATH = "/sys/bus/iio/devices/iio:device0/in_voltage0_raw"

46# 電磁弁を開いてからこの時間経過しても，水が流れていなかったらエラーにする

47TIME_CLOSE_FAIL = 45

49# 電磁弁を閉じてからこの時間経過しても，水が流れていたらエラーにする

50# (Pytest によるテストの際，時間を分単位で制御する関係上，60 より大きい値にしておく)

51TIME_OPEN_FAIL = 61

53# この時間の間，異常な流量になっていたらエラーにする

54TIME_OVER_FAIL = 5

56# この時間の間，流量が 0 だったら，今回の計測を停止する．

57TIME_ZERO_TAIL = 5

60class VALVE_STATE(IntEnum): # noqa: N801

61 OPEN = my_lib.rpi.gpio.level.HIGH.value

62 CLOSE = my_lib.rpi.gpio.level.LOW.value

65class CONTROL_MODE(IntEnum): # noqa: N801

66 TIMER = 1

67 IDLE = 0

70if (os.environ.get("DUMMY_MODE", "false") != "true") and (

71 os.environ.get("TEST", "false") != "true"

72): # pragma: no cover

74 def conv_rawadc_to_flow(adc, offset):

75 flow = max(((adc * ADC_SCALE_VALUE * FLOW_SCALE_MAX) / 5000.0) - offset, 0)

76 if flow < 0.01:

77 flow = 0

79 return flow

81 def get_flow(offset=0):

82 try:

83 with pathlib.Path(ADC_VALUE_PATH).open(mode="r") as f:

84 return {"flow": conv_rawadc_to_flow(int(f.read()), offset), "result": "success"}

85 except Exception:

86 return {"flow": 0, "result": "fail"}

88else:

89 import random

91 def get_flow(offset=0): # noqa: ARG001

92 if STAT_PATH_VALVE_OPEN.exists():

93 if get_flow.prev_flow == 0:

94 flow = FLOW_SCALE_MAX

95 else:

96 flow = max(

97 0,

98 min(

99 get_flow.prev_flow + (random.random() - 0.5) * (FLOW_SCALE_MAX / 5.0), # noqa: S311

100 FLOW_SCALE_MAX,

101 ),

102 )

103

104 get_flow.prev_flow = flow

105

106 return {"flow": flow, "result": "success"}

107 else:

108 if get_flow.prev_flow > 1:

109 get_flow.prev_flow /= 5

110 else:

111 get_flow.prev_flow = max(0, get_flow.prev_flow - 0.5)

112

113 return {"flow": get_flow.prev_flow, "result": "success"}

114

115 get_flow.prev_flow = 0

116

117

118pin_no = GPIO_PIN_DEFAULT

119worker = None

120should_terminate = threading.Event()

121

122

123# NOTE: STAT_PATH_VALVE_CONTROL_COMMAND の内容に基づいて，

124# バルブを一定時間開けます．

125# 時間を操作したテストを行うため，この関数の中では，

126# time.time() の代わりに my_lib.rpi.gpio_time() を使う．

127def control_worker(config, queue): # noqa: PLR0912, PLR0915, C901

128 global should_terminate

129

130 sleep_sec = 0.1

131

132 liveness_file = pathlib.Path(config["liveness"]["file"]["valve_control"])

133

134 logging.info("Start valve control worker")

135

136 time_open_start = None

137 time_close = None

138 flow = 0

139 flow_sum = 0

140 count_flow = 0

141 count_zero = 0

142 count_over = 0

143 notify_last_time = None

144 notify_last_flow_sum = 0

145 notify_last_count = 0

146 stop_measure = False

147

148 i = 0

149 while True:

150 if should_terminate.is_set():

151 break

152

153 if time_open_start is not None:

154 flow = get_flow(config["flow"]["offset"])["flow"]

155 flow_sum += flow

156 count_flow += 1

157

158 if (my_lib.rpi.gpio_time() - notify_last_time) > 10:

159 # NOTE: 10秒ごとに途中集計を報告する

160 queue.put(

161 {

162 "type": "instantaneous",

163 "flow": float(flow_sum - notify_last_flow_sum) / (count_flow - notify_last_count),

164 }

165 )

166

167 notify_last_time = my_lib.rpi.gpio_time()

168 notify_last_flow_sum = flow_sum

169 notify_last_count = count_flow

170

171 # NOTE: 以下の処理はファイルシステムへのアクセスが発生するので，実施頻度を落とす

172 if i % 5 == 0:

173 if time_open_start is None:

174 if STAT_PATH_VALVE_OPEN.exists():

175 # NOTE: バルブが開かれていたら，状態を変更してトータルの水量の集計を開始する

176 time_open_start = my_lib.rpi.gpio_time()

177 notify_last_time = time_open_start

178 # NOTE: バルブを閉じてから流量が 0 になるまでに再度開いた場合にエラーにならないようにする

179 time_close = None

180 else:

181 if STAT_PATH_VALVE_CONTROL_COMMAND.exists():

182 # NOTE: バルブコマンドが存在したら，閉じる時間をチェックして，必要に応じて閉じる

183 try:

184 with valve_open(STAT_PATH_VALVE_CONTROL_COMMAND) as f:

185 time_to_close = float(f.read())

186

187 # NOTE: テストの際に時間を操作する関係で，

188 # 単純な大小比較だけではなく差分絶対値の比較も行う

189 if (my_lib.rpi.gpio_time() > time_to_close) or ( 189 ↛ 199line 189 didn't jump to line 199

190 abs(my_lib.rpi.gpio_time() - time_to_close) < 0.01

191 ):

192 logging.info("Times is up, close valve")

193 # NOTE: 下記の関数の中で

194 # STAT_PATH_VALVE_CONTROL_COMMAND は削除される

195 set_state(VALVE_STATE.CLOSE)

196 time_close = my_lib.rpi.gpio_time()

197 except Exception:

198 logging.warning(traceback.format_exc())

199 if (time_close is None) and STAT_PATH_VALVE_CLOSE.exists():

200 # NOTE: 常にバルブコマンドで制御するので，基本的にここには来ない

201 logging.warning("BUG?")

202 time_close = my_lib.rpi.gpio_time()

203

204 if (time_close is not None) and (time_open_start is not None):

205 period_sec = my_lib.rpi.gpio_time() - time_open_start

206

207 # NOTE: バルブが閉じられた後，流量が 0 になっていたらトータル流量を報告する

208 if flow < 0.1:

209 count_zero += 1

210

211 if flow > FLOW_ERROR_TH:

212 count_over += 1

213

214 if count_over > TIME_OVER_FAIL:

215 set_state(VALVE_STATE.CLOSE)

216 queue.put({"type": "error", "message": "😵水が流れすぎています。"})

217

218 if count_zero > TIME_ZERO_TAIL:

219 # NOTE: 流量(L/min)の平均を求めてから期間(min)を掛ける

220 total = float(flow_sum) / count_flow * period_sec / 60

221

222 queue.put(

223 {

224 "type": "total",

225 "period": period_sec,

226 "total": total,

227 }

228 )

229

230 if (period_sec > TIME_CLOSE_FAIL) and (total < 1):

231 queue.put(

232 {

233 "type": "error",

234 "message": "😵 元栓が閉まっている可能性があります。",

235 }

236 )

237

238 stop_measure = True

239 elif (my_lib.rpi.gpio_time() - time_close) > TIME_OPEN_FAIL:

240 set_state(VALVE_STATE.CLOSE)

241 queue.put(

242 {

243 "type": "error",

244 "message": "😵 バルブを閉めても水が流れ続けています。",

245 }

246 )

247 stop_measure = True

248

249 if stop_measure:

250 stop_measure = False

251 time_open_start = None

252 time_close = None

253 flow_sum = 0

254 count_flow = 0

255 count_zero = 0

256 count_over = 0

257

258 notify_last_time = None

259 notify_last_flow_sum = 0

260 notify_last_count = 0

261

262 time.sleep(sleep_sec)

263

264 if i % (10 / sleep_sec) == 0:

265 my_lib.footprint.update(liveness_file)

266

267 i += 1

268

269 logging.info("Terminate valve control worker")

270

271

272def init(config, queue, pin=GPIO_PIN_DEFAULT):

273 global worker # noqa: PLW0603

274 global pin_no # noqa: PLW0603

275 global STAT_PATH_VALVE_CONTROL_COMMAND # noqa: PLW0603

276 global STAT_PATH_VALVE_OPEN # noqa: PLW0603

277 global STAT_PATH_VALVE_CLOSE # noqa: PLW0603

278

279 STAT_PATH_VALVE_CONTROL_COMMAND = my_lib.webapp.config.STAT_DIR_PATH / "valve" / "control" / "command"

280 STAT_PATH_VALVE_OPEN = my_lib.webapp.config.STAT_DIR_PATH / "valve" / "open"

281 STAT_PATH_VALVE_CLOSE = my_lib.webapp.config.STAT_DIR_PATH / "valve" / "close"

282

283 if worker is not None: 283 ↛ 284line 283 didn't jump to line 284 because the condition on line 283 was never true

284 raise ValueError("worker should be None") # noqa: TRY003, EM101

285

286 pin_no = pin

287

288 set_state(VALVE_STATE.CLOSE)

289

290 logging.info("Setting scale of ADC")

291 if pathlib.Path(ADC_SCALE_PATH).exists():

292 with pathlib.Path(ADC_SCALE_PATH).open(mode="w") as f:

293 f.write(str(ADC_SCALE_VALUE))

294

295 worker = threading.Thread(

296 target=control_worker,

297 args=(

298 config,

299 queue,

300 ),

301 )

302 worker.start()

303

304

305def term():

306 global worker # noqa: PLW0603

307

308 should_terminate.set()

309 worker.join()

310

311 worker = None

312 should_terminate.clear()

313

314 my_lib.rpi.gpio.cleanup()

315

316

317# NOTE: 実際にバルブを開きます．

318def set_state(valve_state):

319 global pin_no

320

321 logging.debug(

322 "set_state = %s from %s at %s:%d",

323 valve_state,

324 inspect.stack()[1].function,

325 inspect.stack()[1].filename,

326 inspect.stack()[1].lineno,

327 )

328

329 curr_state = get_state()

330

331 if valve_state != curr_state:

332 logging.info("VALVE: %s -> %s", curr_state.name, valve_state.name)

333

334 my_lib.rpi.gpio.setwarnings(False)

335 my_lib.rpi.gpio.setmode(my_lib.rpi.gpio.BCM)

336 my_lib.rpi.gpio.setup(pin_no, my_lib.rpi.gpio.OUT)

337 my_lib.rpi.gpio.output(pin_no, valve_state.value)

338

339 if valve_state == VALVE_STATE.OPEN:

340 STAT_PATH_VALVE_CLOSE.unlink(missing_ok=True)

341 if not STAT_PATH_VALVE_OPEN.exists(): 341 ↛ 352line 341 didn't jump to line 352 because the condition on line 341 was always true

342 STAT_PATH_VALVE_OPEN.parent.mkdir(parents=True, exist_ok=True)

343 STAT_PATH_VALVE_OPEN.touch()

344 else:

345 STAT_PATH_VALVE_OPEN.unlink(missing_ok=True)

346 if not STAT_PATH_VALVE_CLOSE.exists():

347 STAT_PATH_VALVE_CLOSE.parent.mkdir(parents=True, exist_ok=True)

348 STAT_PATH_VALVE_CLOSE.touch()

349

350 STAT_PATH_VALVE_CONTROL_COMMAND.unlink(missing_ok=True)

351

352 return get_state()

353

354

355# NOTE: 実際のバルブの状態を返します

356def get_state():

357 global pin_no

358

359 my_lib.rpi.gpio.setwarnings(False)

360 my_lib.rpi.gpio.setmode(my_lib.rpi.gpio.BCM)

361 my_lib.rpi.gpio.setup(pin_no, my_lib.rpi.gpio.OUT)

362

363 if my_lib.rpi.gpio.input(pin_no) == 1:

364 return VALVE_STATE.OPEN

365 else:

366 return VALVE_STATE.CLOSE

367

368

369def set_control_mode(open_sec):

370 logging.info("Open valve for %d sec", open_sec)

371

372 set_state(VALVE_STATE.OPEN)

373

374 time_close = my_lib.rpi.gpio_time() + open_sec

375

376 STAT_PATH_VALVE_CONTROL_COMMAND.parent.mkdir(parents=True, exist_ok=True)

377

378 with pathlib.Path(STAT_PATH_VALVE_CONTROL_COMMAND).open(mode="w") as f:

379 f.write(f"{time_close:.3f}")

380

381

382def get_control_mode():

383 if STAT_PATH_VALVE_CONTROL_COMMAND.exists():

384 with pathlib.Path(STAT_PATH_VALVE_CONTROL_COMMAND).open() as f:

385 time_close = float(f.read())

386 time_now = my_lib.rpi.gpio_time()

387

388 if time_close >= time_now:

389 return {

390 "mode": CONTROL_MODE.TIMER,

391 "remain": time_close - time_now,

392 }

393 else:

394 if (time_now - time_close) > 1: 394 ↛ 396line 394 didn't jump to line 396 because the condition on line 394 was always true

395 logging.warning("Timer control of the valve may be broken")

396 return {"mode": CONTROL_MODE.TIMER, "remain": 0}

397 else:

398 return {"mode": CONTROL_MODE.IDLE, "remain": 0}

399

400

401if __name__ == "__main__":

402 from multiprocessing import Queue

403

404 import my_lib.config

405 import my_lib.logger

406

407 my_lib.logger.init("test", level=logging.INFO)

408

409 config = my_lib.load()

410 queue = Queue()

411 init(config, queue)

412

413 set_state(VALVE_STATE.OPEN)

414 time.sleep(0.5)

415 logging.info("Flow: %.2f", get_flow(config["flow"]["offset"])["flow"])

416 time.sleep(0.5)

417 logging.info("Flow: %.2f", get_flow(config["flow"]["offset"])["flow"])

418 set_state(VALVE_STATE.CLOSE)

419 logging.info("Flow: %.2f", get_flow(config["flow"]["offset"])["flow"])

420

421 set_control_mode(60)

422 time.sleep(1)

423 logging.info(get_control_mode())

424 time.sleep(1)

425 logging.info(get_control_mode())

426 time.sleep(2)

427 logging.info(get_control_mode())

428

429 while True:

430 info = queue.get()

431 logging.info(info)

432

433 if info["type"] == "total":

434 break

435

436 should_terminate.set()