/
air_quality.py
166 lines (137 loc) · 4.68 KB
/
air_quality.py
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
"""Support for the GIOS service."""
from homeassistant.components.air_quality import (
ATTR_CO,
ATTR_NO2,
ATTR_OZONE,
ATTR_PM_2_5,
ATTR_PM_10,
ATTR_SO2,
AirQualityEntity,
)
from homeassistant.const import CONF_NAME
from .const import ATTR_STATION, DATA_CLIENT, DOMAIN, ICONS_MAP
ATTRIBUTION = "Data provided by GIOŚ"
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Add a GIOS entities from a config_entry."""
name = config_entry.data[CONF_NAME]
data = hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id]
async_add_entities([GiosAirQuality(data, name)], True)
def round_state(func):
"""Round state."""
def _decorator(self):
res = func(self)
if isinstance(res, float):
return round(res)
return res
return _decorator
class GiosAirQuality(AirQualityEntity):
"""Define an GIOS sensor."""
def __init__(self, gios, name):
"""Initialize."""
self.gios = gios
self._name = name
self._aqi = None
self._co = None
self._no2 = None
self._o3 = None
self._pm_2_5 = None
self._pm_10 = None
self._so2 = None
self._icon = None
self._attrs = {}
@property
def name(self):
"""Return the name."""
return self._name
@property
def icon(self):
"""Return the icon."""
if self._aqi in ICONS_MAP:
self._icon = ICONS_MAP[self._aqi]
else:
self._icon = "mdi:blur"
return self._icon
@property
def air_quality_index(self):
"""Return the air quality index."""
return self._aqi
@property
@round_state
def particulate_matter_2_5(self):
"""Return the particulate matter 2.5 level."""
return self._pm_2_5
@property
@round_state
def particulate_matter_10(self):
"""Return the particulate matter 10 level."""
return self._pm_10
@property
@round_state
def ozone(self):
"""Return the O3 (ozone) level."""
return self._o3
@property
@round_state
def carbon_monoxide(self):
"""Return the CO (carbon monoxide) level."""
return self._co
@property
@round_state
def sulphur_dioxide(self):
"""Return the SO2 (sulphur dioxide) level."""
return self._so2
@property
@round_state
def nitrogen_dioxide(self):
"""Return the NO2 (nitrogen dioxide) level."""
return self._no2
@property
def attribution(self):
"""Return the attribution."""
return ATTRIBUTION
@property
def state(self):
"""Return the air quality index."""
if self.gios.available:
return self._aqi
@property
def unique_id(self):
"""Return a unique_id for this entity."""
return self.gios.station_id
@property
def available(self):
"""Return True if entity is available."""
return self.gios.available
@property
def device_state_attributes(self):
"""Return the state attributes."""
self._attrs[ATTR_STATION] = self.gios.station_name
return self._attrs
async def async_update(self):
"""Get the data from GIOS."""
await self.gios.async_update()
if self.gios.available:
# Different measuring stations have different sets of sensors. We don't know
# what data we will get.
if "AQI" in self.gios.sensors:
self._aqi = self.gios.sensors["AQI"]["value"]
if "CO" in self.gios.sensors:
self._co = self.gios.sensors["CO"]["value"]
self._attrs[f"{ATTR_CO}_index"] = self.gios.sensors["CO"]["index"]
if "NO2" in self.gios.sensors:
self._no2 = self.gios.sensors["NO2"]["value"]
self._attrs[f"{ATTR_NO2}_index"] = self.gios.sensors["NO2"]["index"]
if "O3" in self.gios.sensors:
self._o3 = self.gios.sensors["O3"]["value"]
self._attrs[f"{ATTR_OZONE}_index"] = self.gios.sensors["O3"]["index"]
if "PM2.5" in self.gios.sensors:
self._pm_2_5 = self.gios.sensors["PM2.5"]["value"]
self._attrs[f"{ATTR_PM_2_5}_index"] = self.gios.sensors["PM2.5"][
"index"
]
if "PM10" in self.gios.sensors:
self._pm_10 = self.gios.sensors["PM10"]["value"]
self._attrs[f"{ATTR_PM_10}_index"] = self.gios.sensors["PM10"]["index"]
if "SO2" in self.gios.sensors:
self._so2 = self.gios.sensors["SO2"]["value"]
self._attrs[f"{ATTR_SO2}_index"] = self.gios.sensors["SO2"]["index"]