forked from jabesq/netatmo-api-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathpublic_data.py
More file actions
171 lines (134 loc) · 5.36 KB
/
Copy pathpublic_data.py
File metadata and controls
171 lines (134 loc) · 5.36 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
from .exceptions import NoDevice
from .helpers import _BASE_URL, toTimeString
_GETPUBLIC_DATA = _BASE_URL + "api/getpublicdata"
_LON_NE = 6.221652
_LAT_NE = 46.610870
_LON_SW = 6.217828
_LAT_SW = 46.596485
_STATION_TEMPERATURE_TYPE = "temperature"
_STATION_PRESSURE_TYPE = "pressure"
_STATION_HUMIDITY_TYPE = "humidity"
_ACCESSORY_RAIN_LIVE_TYPE = "rain_live"
_ACCESSORY_RAIN_60MIN_TYPE = "rain_60min"
_ACCESSORY_RAIN_24H_TYPE = "rain_24h"
_ACCESSORY_RAIN_TIME_TYPE = "rain_timeutc"
_ACCESSORY_WIND_STRENGTH_TYPE = "wind_strength"
_ACCESSORY_WIND_ANGLE_TYPE = "wind_angle"
_ACCESSORY_WIND_TIME_TYPE = "wind_timeutc"
_ACCESSORY_GUST_STRENGTH_TYPE = "gust_strength"
_ACCESSORY_GUST_ANGLE_TYPE = "gust_angle"
class PublicData:
def __init__(
self,
authData,
LAT_NE=_LAT_NE,
LON_NE=_LON_NE,
LAT_SW=_LAT_SW,
LON_SW=_LON_SW,
required_data_type=None, # comma-separated list from above _STATION or _ACCESSORY values
filtering=False,
):
self.authData = authData
postParams = {
"lat_ne": LAT_NE,
"lon_ne": LON_NE,
"lat_sw": LAT_SW,
"lon_sw": LON_SW,
"filter": filtering,
}
if required_data_type:
postParams["required_data"] = required_data_type
resp = self.authData.post_request(url=_GETPUBLIC_DATA, params=postParams)
try:
self.raw_data = resp["body"]
except (KeyError, TypeError):
raise NoDevice("No public weather data returned by Netatmo server")
self.status = resp["status"]
self.time_exec = toTimeString(resp["time_exec"])
self.time_server = toTimeString(resp["time_server"])
def CountStationInArea(self):
return len(self.raw_data)
# Backwards compatibility for < 1.2
def getLive(self):
return self.getLatestRain()
def getLatestRain(self):
return self.getAccessoryMeasures(_ACCESSORY_RAIN_LIVE_TYPE)
def getAverageRain(self):
return averageMeasure(self.getLatestRain())
# Backwards compatibility for < 1.2
def get60min(self):
return self.get60minRain()
def get60minRain(self):
return self.getAccessoryMeasures(_ACCESSORY_RAIN_60MIN_TYPE)
def getAverage60minRain(self):
return averageMeasure(self.get60minRain())
# Backwards compatibility for < 1.2
def get24h(self):
return self.get24hRain()
def get24hRain(self):
return self.getAccessoryMeasures(_ACCESSORY_RAIN_24H_TYPE)
def getAverage24hRain(self):
return averageMeasure(self.get24hRain())
def getLatestPressures(self):
return self.getLatestStationMeasures(_STATION_PRESSURE_TYPE)
def getAveragePressure(self):
return averageMeasure(self.getLatestPressures())
def getLatestTemperatures(self):
return self.getLatestStationMeasures(_STATION_TEMPERATURE_TYPE)
def getAverageTemperature(self):
return averageMeasure(self.getLatestTemperatures())
def getLatestHumidities(self):
return self.getLatestStationMeasures(_STATION_HUMIDITY_TYPE)
def getAverageHumidity(self):
return averageMeasure(self.getLatestHumidities())
def getLatestWindStrengths(self):
return self.getAccessoryMeasures(_ACCESSORY_WIND_STRENGTH_TYPE)
def getAverageWindStrength(self):
return averageMeasure(self.getLatestWindStrengths())
def getLatestWindAngles(self):
return self.getAccessoryMeasures(_ACCESSORY_WIND_ANGLE_TYPE)
def getLatestGustStrengths(self):
return self.getAccessoryMeasures(_ACCESSORY_GUST_STRENGTH_TYPE)
def getAverageGustStrength(self):
return averageMeasure(self.getLatestGustStrengths())
def getLatestGustAngles(self):
return self.getAccessoryMeasures(_ACCESSORY_GUST_ANGLE_TYPE)
def getLocations(self):
locations = {}
for station in self.raw_data:
locations[station["_id"]] = station["place"]["location"]
return locations
# Backwards compatibility for < 1.2
def getTimeforMeasure(self):
return self.getTimeForRainMeasures()
def getTimeForRainMeasures(self):
return self.getAccessoryMeasures(_ACCESSORY_RAIN_TIME_TYPE)
def getTimeForWindMeasures(self):
return self.getAccessoryMeasures(_ACCESSORY_WIND_TIME_TYPE)
def getLatestStationMeasures(self, data_type):
measures = {}
for station in self.raw_data:
for _, module in station["measures"].items():
if (
"type" in module
and data_type in module["type"]
and "res" in module
and module["res"]
):
measure_index = module["type"].index(data_type)
latest_timestamp = sorted(module["res"], reverse=True)[0]
measures[station["_id"]] = module["res"][latest_timestamp][
measure_index
]
return measures
def getAccessoryMeasures(self, data_type):
measures = {}
for station in self.raw_data:
for _, module in station["measures"].items():
if data_type in module:
measures[station["_id"]] = module[data_type]
return measures
def averageMeasure(measures):
if measures:
return sum(measures.values()) / len(measures)
return 0.0