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

1#!/usr/bin/env python3 

2import datetime 

3import functools 

4import json 

5import logging 

6 

7import my_lib.webapp.config 

8import requests 

9 

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

11 

12 

13def get_weather_info_yahoo(config): 

14 try: 

15 params = { 

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

17 "coordinates": ",".join( 

18 map( 

19 str, 

20 [ 

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

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

23 ], 

24 ) 

25 ), 

26 "output": "json", 

27 "past": 2, 

28 } 

29 

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

31 

32 if res.status_code != 200: 

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

34 return None 

35 

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

37 except Exception: 

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

39 return None 

40 

41 

42def get_rain_fall(config): 

43 weather_info = get_weather_info_yahoo(config) 

44 

45 if weather_info is None: 

46 return (False, 0) 

47 

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

49 rainfall_list = [ 

50 x["Rainfall"] 

51 for x in filter( 

52 lambda x: ( 

53 datetime.datetime.now(my_lib.webapp.config.TIMEZONE) 

54 - datetime.datetime.strptime(x["Date"], "%Y%m%d%H%M").astimezone( 

55 my_lib.webapp.config.TIMEZONE_PYTZ 

56 ) 

57 ).total_seconds() 

58 / (60 * 60) 

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

60 weather_info, 

61 ) 

62 ] 

63 

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

65 

66 logging.info( 

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

68 ) 

69 

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

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

72 

73 return (rainfall_judge, rainfall_sum) 

74 

75 

76if __name__ == "__main__": 

77 import my_lib.config 

78 import my_lib.logger 

79 

80 my_lib.logger.init("test", level=logging.INFO) 

81 

82 config = my_lib.load() 

83 

84 print(get_rain_fall(config)) # noqa: T201