forked from dala318/python_ph803w
-
Notifications
You must be signed in to change notification settings - Fork 0
/
sensor.py
155 lines (130 loc) · 4.29 KB
/
sensor.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
"""Platform for sensor integration."""
from __future__ import annotations
import logging
from homeassistant.components.sensor import (
ENTITY_ID_FORMAT,
SensorDeviceClass,
SensorEntity,
)
from homeassistant.const import (
ELECTRIC_POTENTIAL_MILLIVOLT,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import slugify
from . import UPDATE_TOPIC
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
class DeviceSensorConfig:
"""PH-803W Device Sensor configuration."""
def __init__(
self,
friendly_name,
field,
icon="mdi:gauge",
unit_of_measurement=None,
device_class=None,
):
"""Initialize configuration."""
self.device_class = device_class
self.friendly_name = friendly_name
self.field = field
self.icon = icon
self.unit_of_measurement = unit_of_measurement
SENSORS = [
DeviceSensorConfig("PH-803W pH", "ph", "mdi:water-percent", ""),
DeviceSensorConfig(
"PH-803W ORP",
"orp",
"mdi:water-opacity",
ELECTRIC_POTENTIAL_MILLIVOLT,
SensorDeviceClass.VOLTAGE,
),
]
def setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the PH-803W sensor."""
if discovery_info is None:
return
sensors = []
device_data = hass.data[DOMAIN]
for sconfig in SENSORS:
sensors.append(DeviceSensor(device_data, sconfig))
add_entities(sensors)
class DeviceSensor(SensorEntity):
"""Implementing the Waterfurnace sensor."""
def __init__(self, device_data, config):
"""Initialize the sensor."""
self.device_data = device_data
self._name = config.friendly_name
self._attr = config.field
self._state = None
if self.device_data.device_client.get_latest_measurement() is not None:
self._state = getattr(
self.device_data.device_client.get_latest_measurement(),
self._attr,
None,
)
self._icon = config.icon
self._unit_of_measurement = config.unit_of_measurement
self._attr_device_class = config.device_class
# This ensures that the sensors are isolated per waterfurnace unit
self.entity_id = ENTITY_ID_FORMAT.format(
f"wf_{slugify(self.device_data.host)}_{slugify(self._attr)}"
)
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def device_info(self):
"""Return information to link this entity with the correct device."""
return {
"identifiers": {(DOMAIN, self.device_data.device_client.passcode)},
"name": self.device_data.device_client.get_unique_name(),
}
@property
def unique_id(self):
"""Return the sensor unique id."""
return self.device_data.device_client.passcode + self._attr
@property
def native_value(self):
"""Return the state of the sensor."""
return self._state
@property
def icon(self):
"""Return icon."""
return self._icon
@property
def native_unit_of_measurement(self):
"""Return the units of measurement."""
return self._unit_of_measurement
@property
def should_poll(self):
"""Return the polling state."""
return False
async def async_added_to_hass(self):
"""Register callbacks."""
self.async_on_remove(
async_dispatcher_connect(
self.hass, UPDATE_TOPIC, self.async_update_callback
)
)
@callback
def async_update_callback(self):
"""Update state."""
if self.device_data.device_client.get_latest_measurement() is not None:
self._state = getattr(
self.device_data.device_client.get_latest_measurement(),
self._attr,
None,
)
else:
self._state = None
self.async_write_ha_state()