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

1#!/usr/bin/env python3 

2import json 

3import multiprocessing 

4import threading 

5import urllib.parse 

6 

7import flask_cors 

8import my_lib.flask_util 

9import my_lib.webapp.config 

10import my_lib.webapp.event 

11import my_lib.webapp.log 

12import rasp_water.scheduler 

13 

14import flask 

15 

16blueprint = flask.Blueprint("rasp-water-schedule", __name__, url_prefix=my_lib.webapp.config.URL_PREFIX) 

17 

18schedule_lock = threading.Lock() 

19schedule_queue = None 

20worker = None 

21 

22WDAY_STR = ["日", "月", "火", "水", "木", "金", "土"] 

23 

24 

25def init(config): 

26 global worker # noqa: PLW0603 

27 global schedule_queue # noqa: PLW0603 

28 

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

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

31 

32 schedule_queue = multiprocessing.Queue() 

33 rasp_water.scheduler.init() 

34 worker = threading.Thread( 

35 target=rasp_water.scheduler.schedule_worker, 

36 args=( 

37 config, 

38 schedule_queue, 

39 ), 

40 ) 

41 worker.start() 

42 

43 

44def term(): 

45 global worker # noqa: PLW0603 

46 

47 if worker is None: 

48 return 

49 

50 rasp_water.scheduler.should_terminate.set() 

51 worker.join() 

52 

53 worker = None 

54 

55 

56def wday_str_list(wday_list): 

57 wday_str = WDAY_STR 

58 return [wday_str[i] for i in range(len(wday_list)) if wday_list[i]] 

59 

60 

61def schedule_entry_str(entry): 

62 return "{} 開始 {} 分間 {}".format(entry["time"], entry["period"], ",".join(wday_str_list(entry["wday"]))) 

63 

64 

65def schedule_str(schedule): 

66 str_buf = [] 

67 for entry in schedule: 

68 if not entry["is_active"]: 

69 continue 

70 str_buf.append(schedule_entry_str(entry)) 

71 

72 if len(str_buf) == 0: 

73 return "∅ 全て無効" 

74 

75 return "、\n".join(str_buf) 

76 

77 

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

79@my_lib.flask_util.support_jsonp 

80@flask_cors.cross_origin() 

81def api_schedule_ctrl(): 

82 cmd = flask.request.args.get("cmd", None) 

83 data = flask.request.args.get("data", None) 

84 if cmd == "set": 

85 schedule_data = json.loads(data) 

86 

87 if not rasp_water.scheduler.schedule_validate(schedule_data): 

88 my_lib.webapp.log.error("😵 スケジュールの指定が不正です。") 

89 return flask.jsonify(rasp_water.scheduler.schedule_load()) 

90 

91 with schedule_lock: 

92 endpoint = urllib.parse.urljoin( 

93 flask.request.url_root, 

94 flask.url_for("rasp-water-valve.api_valve_ctrl"), 

95 ) 

96 

97 for entry in schedule_data: 

98 entry["endpoint"] = endpoint 

99 schedule_queue.put(schedule_data) 

100 

101 rasp_water.scheduler.schedule_store(schedule_data) 

102 my_lib.webapp.event.notify_event(my_lib.webapp.event.EVENT_TYPE.SCHEDULE) 

103 

104 user = my_lib.flask_util.auth_user(flask.request) 

105 my_lib.webapp.log.info( 

106 "📅 スケジュールを更新しました。\n{schedule}\n{by}".format( 

107 schedule=schedule_str(schedule_data), 

108 by=f"by {user}" if user != "" else "", 

109 ) 

110 ) 

111 

112 return flask.jsonify(rasp_water.scheduler.schedule_load())