-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathbinary_sensor.py
More file actions
73 lines (63 loc) · 2.64 KB
/
binary_sensor.py
File metadata and controls
73 lines (63 loc) · 2.64 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
from __future__ import annotations
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.core import HomeAssistant, callback
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.entity import DeviceInfo
from . import DOMAIN
from .coordinator import ITagClient, SIGNAL_BTN, SIGNAL_CONN, SIGNAL_DISC
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities):
mac = entry.data["mac"].upper()
store = hass.data[DOMAIN]
clients = store.setdefault("clients", {})
client: ITagClient | None = clients.get(mac)
if client is None:
client = clients[mac] = ITagClient(hass, mac)
async_add_entities([ITagButton(hass, mac, client)])
class ITagButton(BinarySensorEntity):
_attr_should_poll = False
def __init__(self, hass: HomeAssistant, mac: str, client: ITagClient):
self.hass = hass
self._mac = mac
self._client = client
self._attr_name = f"iTag Button {mac}"
self._attr_unique_id = f"itag_btn_{mac.replace(':','_')}_v2"
self._attr_is_on = False
self._attr_available = False
self._unsub_btn = None
self._unsub_conn = None
self._unsub_disc = None
async def async_added_to_hass(self):
self._unsub_btn = self.hass.bus.async_listen(f"{SIGNAL_BTN}_{self._mac}", self._on_press)
self._unsub_conn = self.hass.bus.async_listen(f"{SIGNAL_CONN}_{self._mac}", self._on_connected)
self._unsub_disc = self.hass.bus.async_listen(f"{SIGNAL_DISC}_{self._mac}", self._on_disconnected)
try:
await self._client.connect()
if self._client.client and getattr(self._client.client, "is_connected", False):
self._attr_available = True
self.async_write_ha_state()
except Exception:
pass
async def async_will_remove_from_hass(self):
for u in (self._unsub_btn, self._unsub_conn, self._unsub_disc):
if u:
u()
@callback
def _on_connected(self, _):
self._attr_available = True
self.async_write_ha_state()
@callback
def _on_disconnected(self, _):
self._attr_available = False
self.async_write_ha_state()
@callback
def _on_press(self, _event):
self._attr_is_on = True
self.async_write_ha_state()
self.hass.loop.call_later(0.2, self._auto_off)
@callback
def _auto_off(self):
self._attr_is_on = False
self.async_write_ha_state()
@property
def device_info(self) -> DeviceInfo:
return DeviceInfo(identifiers={(DOMAIN, self._mac)}, name=f"iTag {self._mac}")