Coverage for flask/src/rasp_water/weather_forecast.py: 100%

1#!/usr/bin/env python3 

2""" 

3降雨予想を取得します。 

4 

5Usage: 

6 weather_forecast.py [-c CONFIG] [-D] 

7 

8Options: 

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

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

11""" 

12 

13import datetime 

14import functools 

15import json 

16import logging 

17 

18import my_lib.pretty 

19import my_lib.time 

20import my_lib.webapp.config 

21import requests 

22 

23YAHOO_API_ENDPOINT = "https://map.yahooapis.jp/weather/V1/place" 

24 

25 

26def get_weather_info_yahoo(config): 

27 try: 

28 params = { 

29 "appid": config["weather"]["rain_fall"]["forecast"]["yahoo"]["id"], 

30 "coordinates": ",".join( 

31 map( 

32 str, 

33 [ 

34 config["weather"]["rain_fall"]["forecast"]["point"]["lon"], 

35 config["weather"]["rain_fall"]["forecast"]["point"]["lat"], 

36 ], 

37 ) 

38 ), 

39 "output": "json", 

40 "past": 2, 

41 } 

42 

43 res = requests.get(YAHOO_API_ENDPOINT, params=params, timeout=5) 

44 

45 if res.status_code != 200: 

46 logging.warning("Failed to fetch weather info from Yahoo") 

47 return None 

48 

49 return json.loads(res.content)["Feature"][0]["Property"]["WeatherList"]["Weather"] 

50 except Exception: 

51 logging.warning("Failed to fetch weather info from Yahoo") 

52 return None 

53 

54 

55def get_rain_fall(config): 

56 weather_info = get_weather_info_yahoo(config) 

57 

58 if weather_info is None: 

59 return (False, 0) 

60 

61 logging.debug(my_lib.pretty.format(weather_info)) 

62 

63 # NOTE: YAhoo の場合、1 時間後までしか情報がとれないことに注意 

64 rainfall_list = [ 

65 x["Rainfall"] 

66 for x in filter( 

67 lambda x: ( 

68 my_lib.time.now() 

69 - datetime.datetime.strptime(x["Date"], "%Y%m%d%H%M").replace(tzinfo=my_lib.time.get_pytz()) 

70 ).total_seconds() 

71 / (60 * 60) 

72 < config["weather"]["rain_fall"]["forecast"]["threshold"]["before_hour"], 

73 weather_info, 

74 ) 

75 ] 

76 

77 rainfall_sum = functools.reduce(lambda x, y: x + y, rainfall_list) 

78 

79 logging.info( 

80 "Rain fall forecast sum: %d (%s)", rainfall_sum, ", ".join(f"{num:.1f}" for num in rainfall_list) 

81 ) 

82 

83 rainfall_judge = rainfall_sum > config["weather"]["rain_fall"]["forecast"]["threshold"]["sum"] 

84 logging.info("Rain fall forecast judge: %s", rainfall_judge) 

85 

86 return (rainfall_judge, rainfall_sum) 

87 

88 

89if __name__ == "__main__": 

90 # TEST Code 

91 import docopt 

92 import my_lib.config 

93 import my_lib.logger 

94 

95 args = docopt.docopt(__doc__) 

96 

97 config_file = args["-c"] 

98 debug_mode = args["-D"] 

99 

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

101 

102 config = my_lib.config.load(config_file) 

103 

104 logging.info(my_lib.pretty.format(get_rain_fall(config)))