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

1#!/usr/bin/env python3 

2import json 

3import multiprocessing 

4import os 

5import threading 

6import urllib.parse 

7 

8import flask_cors 

9import my_lib.flask_util 

10import my_lib.webapp.config 

11import my_lib.webapp.event 

12import my_lib.webapp.log 

13import rasp_shutter.control.scheduler 

14 

15import flask 

16 

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

18 

19_schedule_lock = {} 

20_schedule_queue = {} 

21_worker_thread = {} 

22 

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

24 

25 

26def init(config): 

27 init_impl(config) 

28 

29 

30def term(): 

31 if get_worker_thread() is None: 31 ↛ 32line 31 didn't jump to line 32 because the condition on line 31 was never true

32 return 

33 

34 rasp_shutter.control.scheduler.term() 

35 get_worker_thread().join() 

36 del _worker_thread[get_worker_id()] 

37 

38 

39def init_impl(config): 

40 if get_worker_thread() is not None: 40 ↛ 41line 40 didn't jump to line 41 because the condition on line 40 was never true

41 raise ValueError("worker should be None") # noqa: TRY003 

42 

43 worker_id = get_worker_id() 

44 

45 _schedule_queue[worker_id] = multiprocessing.Queue() 

46 _schedule_lock[worker_id] = threading.RLock() 

47 rasp_shutter.control.scheduler.init() 

48 

49 _worker_thread[worker_id] = threading.Thread( 

50 target=rasp_shutter.control.scheduler.schedule_worker, 

51 args=( 

52 config, 

53 get_schedule_queue(), 

54 ), 

55 ) 

56 _worker_thread[worker_id].start() 

57 

58 

59def get_worker_id(): 

60 return os.environ.get("PYTEST_XDIST_WORKER", "") 

61 

62 

63def get_schedule_lock(): 

64 return _schedule_lock.get(get_worker_id(), None) 

65 

66 

67def get_schedule_queue(): 

68 return _schedule_queue.get(get_worker_id(), None) 

69 

70 

71def get_worker_thread(): 

72 return _worker_thread.get(get_worker_id(), None) 

73 

74 

75def wday_str_list(wday_list): 

76 wday_str = WDAY_STR 

77 

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

79 

80 

81def schedule_entry_str(name, entry): 

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

83 name=name.upper(), 

84 time=entry["time"], 

85 solar_rad=entry["solar_rad"], 

86 lux=entry["lux"], 

87 altitude=entry["altitude"], 

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

89 ) 

90 

91 

92def schedule_str(schedule_data): 

93 str_buf = [] 

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

95 entry = schedule_data[name] 

96 if not entry["is_active"]: 

97 continue 

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

99 

100 if len(str_buf) == 0: 

101 return "∅ 全て無効" 

102 

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

104 

105 

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

107@my_lib.flask_util.support_jsonp 

108@flask_cors.cross_origin() 

109def api_schedule_ctrl(): 

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

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

112 if cmd == "set": 

113 schedule_data = json.loads(data) 

114 

115 if not rasp_shutter.control.scheduler.schedule_validate(schedule_data): 

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

117 return flask.jsonify(rasp_shutter.control.scheduler.schedule_load()) 

118 

119 with get_schedule_lock(): 

120 schedule_data = json.loads(data) 

121 

122 endpoint = urllib.parse.urljoin( 

123 flask.request.url_root, 

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

125 ) 

126 

127 for entry in schedule_data.values(): 

128 entry["endpoint"] = endpoint 

129 get_schedule_queue().put(schedule_data) 

130 

131 rasp_shutter.control.scheduler.schedule_store(schedule_data) 

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

133 

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

135 my_lib.webapp.log.info( 

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

137 schedule=schedule_str(schedule_data), 

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

139 ) 

140 ) 

141 

142 return flask.jsonify(rasp_shutter.control.scheduler.schedule_load())