-
Notifications
You must be signed in to change notification settings - Fork 7
/
sensor_common.py
107 lines (86 loc) · 2.86 KB
/
sensor_common.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
import logging
import typing
from typing import Iterable, Iterator, List, Optional, Type, Union
from . import WeatherLinkCoordinator, WeatherLinkEntity
from .api.conditions import ConditionRecord, CurrentConditions
from .units import Measurement
logger = logging.getLogger(__name__)
class WeatherLinkSensor(WeatherLinkEntity):
_SENSORS: List[Type["WeatherLinkSensor"]] = []
@typing.overload
def __init_subclass__(
cls,
*,
sensor_name: str,
unit_of_measurement: Union[str, Type[Measurement], None],
device_class: Optional[str],
required_conditions: Iterable[Type[ConditionRecord]] = None,
**kwargs,
) -> None:
...
@typing.overload
def __init_subclass__(
cls,
*,
abc: bool,
**kwargs,
) -> None:
...
def __init_subclass__(
cls,
abc: bool = False,
**kwargs,
) -> None:
if abc:
super().__init_subclass__(**kwargs)
return
sensor_name = kwargs.pop("sensor_name")
unit_of_measurement = kwargs.pop("unit_of_measurement")
device_class = kwargs.pop("device_class")
required_conditions = kwargs.pop("required_conditions", None)
super().__init_subclass__(**kwargs)
cls._sensor_name = sensor_name
cls._unit_of_measurement = unit_of_measurement
cls._device_class = device_class
try:
requirements = cls._required_conditions
except AttributeError:
requirements = ()
cls._required_conditions = requirements + tuple(required_conditions)
cls._SENSORS.append(cls)
@classmethod
def _conditions_ok(cls, conditions: CurrentConditions) -> bool:
for req in cls._required_conditions:
if req not in conditions:
return False
return True
@classmethod
def iter_sensors_for_coordinator(
cls, coord: WeatherLinkCoordinator
) -> Iterator["WeatherLinkSensor"]:
for cls in cls._SENSORS:
if not cls._conditions_ok(coord.data):
logger.debug(
"ignoring sensor %s because requirements are not met",
cls.__qualname__,
)
continue
yield cls(coord)
@property
def name(self):
return f"{self.coordinator.device_model_name} {self._sensor_name}"
@property
def unit_of_measurement(self):
unit = self._unit_of_measurement
if unit is None or isinstance(unit, str):
return unit
return self.units.by_measurement(unit).info.unit_of_measurement
@property
def device_class(self):
return self._device_class
def round_optional(
f: Optional[Union[int, float]], ndigits: int = None
) -> Optional[Union[int, float]]:
if not f:
return f
return round(f, ndigits)