Coverage for flask/src/rasp_shutter/webapp_sensor.py: 100%

1#!/usr/bin/env python3 

2import datetime 

3 

4import flask_cors 

5import my_lib.flask_util 

6import my_lib.sensor_data 

7import my_lib.webapp.config 

8import pysolar.solar 

9import pytz 

10 

11import flask 

12 

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

14 

15 

16def get_solar_altitude(config): 

17 now = datetime.datetime.now(datetime.timezone.utc) 

18 return { 

19 "value": pysolar.solar.get_altitude( 

20 config["location"]["latitude"], config["location"]["longitude"], now 

21 ), 

22 "valid": True, 

23 "time": now, 

24 } 

25 

26 

27def get_sensor_data(config): 

28 timezone = pytz.timezone("Asia/Tokyo") 

29 

30 sense_data = {} 

31 for field in ["lux", "solar_rad"]: 

32 sensor = config["sensor"][field] 

33 data = my_lib.sensor_data.fetch_data( 

34 config["sensor"]["influxdb"], 

35 sensor["measure"], 

36 sensor["hostname"], 

37 field, 

38 start="-1h", 

39 last=True, 

40 ) 

41 if data["valid"]: 

42 sense_data[field] = { 

43 "value": data["value"][0], 

44 # NOTE: タイムゾーン情報を削除しておく。 

45 "time": timezone.localize(data["time"][0].replace(tzinfo=None)), 

46 "valid": True, 

47 } 

48 else: 

49 sense_data[field] = { 

50 "valid": False, 

51 } 

52 

53 sense_data["altitude"] = get_solar_altitude(config) 

54 

55 return sense_data 

56 

57 

58@blueprint.route("/api/sensor", methods=["GET"]) 

59@my_lib.flask_util.support_jsonp 

60@flask_cors.cross_origin() 

61def api_sensor_data(): 

62 return flask.jsonify(get_sensor_data(flask.current_app.config["CONFIG"]))