Coverage for flask/src/rasp_shutter/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_shutter.scheduler 

13 

14import flask 

15 

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

17 

18 

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

20 

21schedule_lock = threading.Lock() 

22schedule_queue = None 

23worker = None 

24 

25 

26def init(config): 

27 global worker # noqa: PLW0603 

28 global schedule_queue # noqa: PLW0603 

29 

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

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

32 

33 schedule_queue = multiprocessing.Queue() 

34 rasp_shutter.scheduler.init() 

35 worker = threading.Thread( 

36 target=rasp_shutter.scheduler.schedule_worker, 

37 args=( 

38 config, 

39 schedule_queue, 

40 ), 

41 ) 

42 worker.start() 

43 

44 

45def term(): 

46 global worker # noqa: PLW0603 

47 

48 if worker is None: 

49 return 

50 

51 rasp_shutter.scheduler.term() 

52 worker.join() 

53 

54 worker = None 

55 

56 

57def wday_str_list(wday_list): 

58 wday_str = WDAY_STR 

59 

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

61 

62 

63def schedule_entry_str(name, entry): 

64 return "{name} {time} {solar_rad} W/mm^2 {lux} LUX {altitude} deg {wday}".format( 

65 name=name.upper(), 

66 time=entry["time"], 

67 solar_rad=entry["solar_rad"], 

68 lux=entry["lux"], 

69 altitude=entry["altitude"], 

70 wday=",".join(wday_str_list(entry["wday"])), 

71 ) 

72 

73 

74def schedule_str(schedule_data): 

75 str_buf = [] 

76 for name in ["open", "close"]: 

77 entry = schedule_data[name] 

78 if not entry["is_active"]: 

79 continue 

80 str_buf.append(schedule_entry_str(name, entry)) 

81 

82 if len(str_buf) == 0: 

83 return "∅ 全て無効" 

84 

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

86 

87 

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

89@my_lib.flask_util.support_jsonp 

90@flask_cors.cross_origin() 

91def api_schedule_ctrl(): 

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

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

94 if cmd == "set": 

95 schedule_data = json.loads(data) 

96 

97 if not rasp_shutter.scheduler.schedule_validate(schedule_data): 

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

99 return flask.jsonify(rasp_shutter.scheduler.schedule_load()) 

100 

101 with schedule_lock: 

102 schedule_data = json.loads(data) 

103 

104 endpoint = urllib.parse.urljoin( 

105 flask.request.url_root, 

106 flask.url_for("rasp-shutter-control.api_shutter_ctrl"), 

107 ) 

108 

109 for entry in schedule_data.values(): 

110 entry["endpoint"] = endpoint 

111 schedule_queue.put(schedule_data) 

112 

113 rasp_shutter.scheduler.schedule_store(schedule_data) 

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

115 

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

117 my_lib.webapp.log.info( 

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

119 schedule=schedule_str(schedule_data), 

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

121 ) 

122 ) 

123 

124 return flask.jsonify(rasp_shutter.scheduler.schedule_load())