mirror of
https://github.com/home-assistant/core
synced 2024-10-13 14:52:10 +02:00
ad102b3840
* Use f-strings in integrations starting with F * Use f-strings in tests for integrations starting with F * Use f-strings in integrations starting with G * Use f-strings in tests for integrations starting with G * Fix pylint error * Fix broken test
248 lines
8.1 KiB
Python
248 lines
8.1 KiB
Python
"""Support to use flic buttons as a binary sensor."""
|
|
import logging
|
|
import threading
|
|
|
|
from pyflic import (
|
|
ButtonConnectionChannel,
|
|
ClickType,
|
|
ConnectionStatus,
|
|
FlicClient,
|
|
ScanWizard,
|
|
ScanWizardResult,
|
|
)
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.components.binary_sensor import PLATFORM_SCHEMA, BinarySensorDevice
|
|
from homeassistant.const import (
|
|
CONF_DISCOVERY,
|
|
CONF_HOST,
|
|
CONF_PORT,
|
|
CONF_TIMEOUT,
|
|
EVENT_HOMEASSISTANT_STOP,
|
|
)
|
|
import homeassistant.helpers.config_validation as cv
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
DEFAULT_TIMEOUT = 3
|
|
|
|
CLICK_TYPE_SINGLE = "single"
|
|
CLICK_TYPE_DOUBLE = "double"
|
|
CLICK_TYPE_HOLD = "hold"
|
|
CLICK_TYPES = [CLICK_TYPE_SINGLE, CLICK_TYPE_DOUBLE, CLICK_TYPE_HOLD]
|
|
|
|
CONF_IGNORED_CLICK_TYPES = "ignored_click_types"
|
|
|
|
DEFAULT_HOST = "localhost"
|
|
DEFAULT_PORT = 5551
|
|
|
|
EVENT_NAME = "flic_click"
|
|
EVENT_DATA_NAME = "button_name"
|
|
EVENT_DATA_ADDRESS = "button_address"
|
|
EVENT_DATA_TYPE = "click_type"
|
|
EVENT_DATA_QUEUED_TIME = "queued_time"
|
|
|
|
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
|
{
|
|
vol.Optional(CONF_HOST, default=DEFAULT_HOST): cv.string,
|
|
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
|
|
vol.Optional(CONF_DISCOVERY, default=True): cv.boolean,
|
|
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
|
|
vol.Optional(CONF_IGNORED_CLICK_TYPES): vol.All(
|
|
cv.ensure_list, [vol.In(CLICK_TYPES)]
|
|
),
|
|
}
|
|
)
|
|
|
|
|
|
def setup_platform(hass, config, add_entities, discovery_info=None):
|
|
"""Set up the flic platform."""
|
|
|
|
# Initialize flic client responsible for
|
|
# connecting to buttons and retrieving events
|
|
host = config.get(CONF_HOST)
|
|
port = config.get(CONF_PORT)
|
|
discovery = config.get(CONF_DISCOVERY)
|
|
|
|
try:
|
|
client = FlicClient(host, port)
|
|
except ConnectionRefusedError:
|
|
_LOGGER.error("Failed to connect to flic server")
|
|
return
|
|
|
|
def new_button_callback(address):
|
|
"""Set up newly verified button as device in Home Assistant."""
|
|
setup_button(hass, config, add_entities, client, address)
|
|
|
|
client.on_new_verified_button = new_button_callback
|
|
if discovery:
|
|
start_scanning(config, add_entities, client)
|
|
|
|
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, lambda event: client.close())
|
|
|
|
# Start the pyflic event handling thread
|
|
threading.Thread(target=client.handle_events).start()
|
|
|
|
def get_info_callback(items):
|
|
"""Add entities for already verified buttons."""
|
|
addresses = items["bd_addr_of_verified_buttons"] or []
|
|
for address in addresses:
|
|
setup_button(hass, config, add_entities, client, address)
|
|
|
|
# Get addresses of already verified buttons
|
|
client.get_info(get_info_callback)
|
|
|
|
|
|
def start_scanning(config, add_entities, client):
|
|
"""Start a new flic client for scanning and connecting to new buttons."""
|
|
scan_wizard = ScanWizard()
|
|
|
|
def scan_completed_callback(scan_wizard, result, address, name):
|
|
"""Restart scan wizard to constantly check for new buttons."""
|
|
if result == ScanWizardResult.WizardSuccess:
|
|
_LOGGER.info("Found new button %s", address)
|
|
elif result != ScanWizardResult.WizardFailedTimeout:
|
|
_LOGGER.warning(
|
|
"Failed to connect to button %s. Reason: %s", address, result
|
|
)
|
|
|
|
# Restart scan wizard
|
|
start_scanning(config, add_entities, client)
|
|
|
|
scan_wizard.on_completed = scan_completed_callback
|
|
client.add_scan_wizard(scan_wizard)
|
|
|
|
|
|
def setup_button(hass, config, add_entities, client, address):
|
|
"""Set up a single button device."""
|
|
timeout = config.get(CONF_TIMEOUT)
|
|
ignored_click_types = config.get(CONF_IGNORED_CLICK_TYPES)
|
|
button = FlicButton(hass, client, address, timeout, ignored_click_types)
|
|
_LOGGER.info("Connected to button %s", address)
|
|
|
|
add_entities([button])
|
|
|
|
|
|
class FlicButton(BinarySensorDevice):
|
|
"""Representation of a flic button."""
|
|
|
|
def __init__(self, hass, client, address, timeout, ignored_click_types):
|
|
"""Initialize the flic button."""
|
|
|
|
self._hass = hass
|
|
self._address = address
|
|
self._timeout = timeout
|
|
self._is_down = False
|
|
self._ignored_click_types = ignored_click_types or []
|
|
self._hass_click_types = {
|
|
ClickType.ButtonClick: CLICK_TYPE_SINGLE,
|
|
ClickType.ButtonSingleClick: CLICK_TYPE_SINGLE,
|
|
ClickType.ButtonDoubleClick: CLICK_TYPE_DOUBLE,
|
|
ClickType.ButtonHold: CLICK_TYPE_HOLD,
|
|
}
|
|
|
|
self._channel = self._create_channel()
|
|
client.add_connection_channel(self._channel)
|
|
|
|
def _create_channel(self):
|
|
"""Create a new connection channel to the button."""
|
|
channel = ButtonConnectionChannel(self._address)
|
|
channel.on_button_up_or_down = self._on_up_down
|
|
|
|
# If all types of clicks should be ignored, skip registering callbacks
|
|
if set(self._ignored_click_types) == set(CLICK_TYPES):
|
|
return channel
|
|
|
|
if CLICK_TYPE_DOUBLE in self._ignored_click_types:
|
|
# Listen to all but double click type events
|
|
channel.on_button_click_or_hold = self._on_click
|
|
elif CLICK_TYPE_HOLD in self._ignored_click_types:
|
|
# Listen to all but hold click type events
|
|
channel.on_button_single_or_double_click = self._on_click
|
|
else:
|
|
# Listen to all click type events
|
|
channel.on_button_single_or_double_click_or_hold = self._on_click
|
|
|
|
return channel
|
|
|
|
@property
|
|
def name(self):
|
|
"""Return the name of the device."""
|
|
return f"flic_{self.address.replace(':', '')}"
|
|
|
|
@property
|
|
def address(self):
|
|
"""Return the bluetooth address of the device."""
|
|
return self._address
|
|
|
|
@property
|
|
def is_on(self):
|
|
"""Return true if sensor is on."""
|
|
return self._is_down
|
|
|
|
@property
|
|
def should_poll(self):
|
|
"""No polling needed."""
|
|
return False
|
|
|
|
@property
|
|
def device_state_attributes(self):
|
|
"""Return device specific state attributes."""
|
|
return {"address": self.address}
|
|
|
|
def _queued_event_check(self, click_type, time_diff):
|
|
"""Generate a log message and returns true if timeout exceeded."""
|
|
time_string = f"{time_diff:d} {'second' if time_diff == 1 else 'seconds'}"
|
|
|
|
if time_diff > self._timeout:
|
|
_LOGGER.warning(
|
|
"Queued %s dropped for %s. Time in queue was %s",
|
|
click_type,
|
|
self.address,
|
|
time_string,
|
|
)
|
|
return True
|
|
_LOGGER.info(
|
|
"Queued %s allowed for %s. Time in queue was %s",
|
|
click_type,
|
|
self.address,
|
|
time_string,
|
|
)
|
|
return False
|
|
|
|
def _on_up_down(self, channel, click_type, was_queued, time_diff):
|
|
"""Update device state, if event was not queued."""
|
|
if was_queued and self._queued_event_check(click_type, time_diff):
|
|
return
|
|
|
|
self._is_down = click_type == ClickType.ButtonDown
|
|
self.schedule_update_ha_state()
|
|
|
|
def _on_click(self, channel, click_type, was_queued, time_diff):
|
|
"""Fire click event, if event was not queued."""
|
|
# Return if click event was queued beyond allowed timeout
|
|
if was_queued and self._queued_event_check(click_type, time_diff):
|
|
return
|
|
|
|
# Return if click event is in ignored click types
|
|
hass_click_type = self._hass_click_types[click_type]
|
|
if hass_click_type in self._ignored_click_types:
|
|
return
|
|
|
|
self._hass.bus.fire(
|
|
EVENT_NAME,
|
|
{
|
|
EVENT_DATA_NAME: self.name,
|
|
EVENT_DATA_ADDRESS: self.address,
|
|
EVENT_DATA_QUEUED_TIME: time_diff,
|
|
EVENT_DATA_TYPE: hass_click_type,
|
|
},
|
|
)
|
|
|
|
def _connection_status_changed(self, channel, connection_status, disconnect_reason):
|
|
"""Remove device, if button disconnects."""
|
|
if connection_status == ConnectionStatus.Disconnected:
|
|
_LOGGER.warning(
|
|
"Button (%s) disconnected. Reason: %s", self.address, disconnect_reason
|
|
)
|