Coverage for flask/src/rasp_water/control/weather_sensor.py: 15%

1#!/usr/bin/env python3 

2""" 

3雨量データを取得します。 

4 

5Usage: 

6 weather_sensor.py [-c CONFIG] [-p HOURS] [-D] 

7 

8Options: 

9 -c CONFIG : CONFIG を設定ファイルとして読み込んで実行します。[default: config.yaml] 

10 -p HOURS : 集計する時間を指定します。[default: 12] 

11 -D : デバッグモードで動作します。 

12""" 

13 

14import datetime 

15import logging 

16 

17import my_lib.sensor_data 

18import my_lib.time 

19import rasp_water.control.scheduler 

20 

21 

22def hours_since_last_watering(): 

23 schedule_data = rasp_water.control.scheduler.schedule_load() 

24 

25 now = my_lib.time.now() 

26 

27 last_date = [] 

28 for schedule in schedule_data: 

29 if not schedule["is_active"]: 

30 continue 

31 time_str = schedule["time"] 

32 hour, minute = map(int, time_str.split(":")) 

33 

34 for days_ago in range(7): 

35 day = now - datetime.timedelta(days=days_ago) 

36 wday_index = day.weekday() 

37 if schedule["wday"][wday_index]: 

38 scheduled_date = day.replace(hour=hour, minute=minute, second=0, microsecond=0) 

39 last_date.append(scheduled_date) 

40 break 

41 

42 if len(last_date) == 0: 

43 return 24 * 7 

44 

45 minutes = (now - max(last_date)).total_seconds() / 60 

46 

47 hours = int(minutes // 60) 

48 if minutes % 60 >= 30: 

49 hours += 1 

50 

51 return hours 

52 

53 

54def get_rain_fall_sum(config, hours): 

55 return my_lib.sensor_data.get_hour_sum( 

56 config["influxdb"], 

57 config["weather"]["rain_fall"]["sensor"]["measure"], 

58 config["weather"]["rain_fall"]["sensor"]["hostname"], 

59 "rain", 

60 hours=hours, 

61 ) 

62 

63 

64def get_rain_fall(config): 

65 hours = hours_since_last_watering() 

66 # InfluxDBクエリエラーを避けるため、最小1時間に設定 

67 hours = max(1, hours) 

68 

69 try: 

70 rain_fall_sum = get_rain_fall_sum(config, hours) 

71 except Exception as e: 

72 logging.warning("Failed to get rain fall data, assuming no rain: %s", e) 

73 rain_fall_sum = 0.0 

74 

75 logging.info("Rain fall sum since last watering: %.1f (%d hours)", rain_fall_sum, hours) 

76 

77 rainfall_judge = rain_fall_sum > config["weather"]["rain_fall"]["sensor"]["threshold"]["sum"] 

78 logging.info("Rain fall sensor judge: %s", rainfall_judge) 

79 

80 return (rainfall_judge, rain_fall_sum) 

81 

82 

83if __name__ == "__main__": 

84 # TEST Code 

85 import docopt 

86 import my_lib.config 

87 import my_lib.logger 

88 import my_lib.webapp.config 

89 

90 args = docopt.docopt(__doc__) 

91 

92 config_file = args["-c"] 

93 hours = int(args["-p"]) 

94 debug_mode = args["-D"] 

95 

96 my_lib.logger.init("test", level=logging.DEBUG if debug_mode else logging.INFO) 

97 

98 config = my_lib.config.load(config_file) 

99 

100 my_lib.webapp.config.init(config) 

101 rasp_water.control.scheduler.init() 

102 

103 logging.info("Sum of rainfall is %.1f (%d hours)", get_rain_fall_sum(config, hours), hours) 

104 

105 logging.info("Finish.")