Add Duty binary_sensor platform to FireServiceRota integration (#43638)

This commit is contained in:
Ron Klinkien 2020-11-26 15:42:55 +01:00 committed by GitHub
parent eb3e5cf446
commit 316a2750df
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 139 additions and 34 deletions

View file

@ -263,6 +263,7 @@ omit =
homeassistant/components/filesize/sensor.py
homeassistant/components/fints/sensor.py
homeassistant/components/fireservicerota/__init__.py
homeassistant/components/fireservicerota/binary_sensor.py
homeassistant/components/fireservicerota/const.py
homeassistant/components/fireservicerota/sensor.py
homeassistant/components/firmata/__init__.py

View file

@ -11,6 +11,7 @@ from pyfireservicerota import (
InvalidTokenError,
)
from homeassistant.components.binary_sensor import DOMAIN as BINARYSENSOR_DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntry
from homeassistant.const import CONF_TOKEN, CONF_URL, CONF_USERNAME
@ -18,13 +19,13 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.dispatcher import dispatcher_send
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN, WSS_BWRURL
from .const import DATA_CLIENT, DATA_COORDINATOR, DOMAIN, WSS_BWRURL
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=60)
_LOGGER = logging.getLogger(__name__)
SUPPORTED_PLATFORMS = {SENSOR_DOMAIN}
SUPPORTED_PLATFORMS = {SENSOR_DOMAIN, BINARYSENSOR_DOMAIN}
async def async_setup(hass: HomeAssistant, config: dict) -> bool:
@ -37,14 +38,30 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up FireServiceRota from a config entry."""
hass.data.setdefault(DOMAIN, {})
coordinator = FireServiceRotaCoordinator(hass, entry)
await coordinator.setup()
await coordinator.async_availability_update()
if coordinator.token_refresh_failure:
client = FireServiceRotaClient(hass, entry)
await client.setup()
if client.token_refresh_failure:
return False
hass.data[DOMAIN][entry.entry_id] = coordinator
async def async_update_data():
return await client.async_update()
coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
name="duty binary sensor",
update_method=async_update_data,
update_interval=MIN_TIME_BETWEEN_UPDATES,
)
await coordinator.async_refresh()
hass.data[DOMAIN][entry.entry_id] = {
DATA_CLIENT: client,
DATA_COORDINATOR: coordinator,
}
for platform in SUPPORTED_PLATFORMS:
hass.async_create_task(
@ -161,7 +178,7 @@ class FireServiceRotaWebSocket:
self._fsr_incidents.stop()
class FireServiceRotaCoordinator(DataUpdateCoordinator):
class FireServiceRotaClient:
"""Getting the latest data from fireservicerota."""
def __init__(self, hass, entry):
@ -169,14 +186,6 @@ class FireServiceRotaCoordinator(DataUpdateCoordinator):
self._hass = hass
self._entry = entry
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_method=self.async_availability_update,
update_interval=MIN_TIME_BETWEEN_UPDATES,
)
self._url = entry.data[CONF_URL]
self._tokens = entry.data[CONF_TOKEN]
@ -194,7 +203,7 @@ class FireServiceRotaCoordinator(DataUpdateCoordinator):
self.websocket = FireServiceRotaWebSocket(self._hass, self._entry)
async def setup(self) -> None:
"""Set up the coordinator."""
"""Set up the data client."""
await self._hass.async_add_executor_job(self.websocket.start_listener)
async def update_call(self, func, *args):
@ -207,23 +216,22 @@ class FireServiceRotaCoordinator(DataUpdateCoordinator):
except (ExpiredTokenError, InvalidTokenError):
self.websocket.stop_listener()
self.token_refresh_failure = True
self.update_interval = None
if await self.oauth.async_refresh_tokens():
self.update_interval = MIN_TIME_BETWEEN_UPDATES
self.token_refresh_failure = False
self.websocket.start_listener()
return await self._hass.async_add_executor_job(func, *args)
async def async_availability_update(self) -> None:
async def async_update(self) -> object:
"""Get the latest availability data."""
_LOGGER.debug("Updating availability data")
return await self.update_call(
data = await self.update_call(
self.fsr.get_availability, str(self._hass.config.time_zone)
)
_LOGGER.debug("Updated availability data: %s", data)
return data
async def async_response_update(self) -> object:
"""Get the latest incident response data."""
data = self.websocket.incident_data()

View file

@ -0,0 +1,93 @@
"""Binary Sensor platform for FireServiceRota integration."""
import logging
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import DATA_COORDINATOR, DOMAIN as FIRESERVICEROTA_DOMAIN
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities
) -> None:
"""Set up FireServiceRota binary sensor based on a config entry."""
coordinator: DataUpdateCoordinator = hass.data[FIRESERVICEROTA_DOMAIN][
entry.entry_id
][DATA_COORDINATOR]
async_add_entities([ResponseBinarySensor(coordinator, entry)])
class ResponseBinarySensor(CoordinatorEntity, BinarySensorEntity):
"""Representation of an FireServiceRota sensor."""
def __init__(self, coordinator: DataUpdateCoordinator, entry):
"""Initialize."""
super().__init__(coordinator)
self._unique_id = f"{entry.unique_id}_Duty"
self._state = None
@property
def name(self) -> str:
"""Return the name of the sensor."""
return "Duty"
@property
def icon(self) -> str:
"""Return the icon to use in the frontend."""
return "mdi:calendar"
@property
def unique_id(self) -> str:
"""Return the unique ID for this binary sensor."""
return self._unique_id
@property
def is_on(self):
"""Return the state of the binary sensor."""
if not self.coordinator.data:
return
data = self.coordinator.data
if "available" in data and data["available"]:
self._state = True
else:
self._state = False
_LOGGER.debug("Set state of entity 'Duty Binary Sensor' to '%s'", self._state)
return self._state
@property
def device_state_attributes(self):
"""Return available attributes for binary sensor."""
attr = {}
if not self.coordinator.data:
return attr
data = self.coordinator.data
attr = {
key: data[key]
for key in (
"start_time",
"end_time",
"available",
"active",
"assigned_function_ids",
"skill_ids",
"type",
"assigned_function",
)
if key in data
}
_LOGGER.debug("Set attributes of entity 'Duty Binary Sensor' to '%s'", attr)
return attr

View file

@ -7,3 +7,6 @@ URL_LIST = {
"www.fireservicerota.co.uk": "FireServiceRota",
}
WSS_BWRURL = "wss://{0}/cable?access_token={1}"
DATA_CLIENT = "client"
DATA_COORDINATOR = "coordinator"

View file

@ -7,7 +7,7 @@ from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import HomeAssistantType
from .const import DOMAIN as FIRESERVICEROTA_DOMAIN
from .const import DATA_CLIENT, DOMAIN as FIRESERVICEROTA_DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -16,19 +16,19 @@ async def async_setup_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities
) -> None:
"""Set up FireServiceRota sensor based on a config entry."""
coordinator = hass.data[FIRESERVICEROTA_DOMAIN][entry.entry_id]
client = hass.data[FIRESERVICEROTA_DOMAIN][entry.entry_id][DATA_CLIENT]
async_add_entities([IncidentsSensor(coordinator)])
async_add_entities([IncidentsSensor(client)])
class IncidentsSensor(RestoreEntity):
"""Representation of FireServiceRota incidents sensor."""
def __init__(self, coordinator):
def __init__(self, client):
"""Initialize."""
self._coordinator = coordinator
self._entry_id = self._coordinator._entry.entry_id
self._unique_id = f"{self._coordinator._entry.unique_id}_Incidents"
self._client = client
self._entry_id = self._client._entry.entry_id
self._unique_id = f"{self._client._entry.unique_id}_Incidents"
self._state = None
self._state_attributes = {}
@ -112,14 +112,14 @@ class IncidentsSensor(RestoreEntity):
async_dispatcher_connect(
self.hass,
f"{FIRESERVICEROTA_DOMAIN}_{self._entry_id}_update",
self.coordinator_update,
self.client_update,
)
)
@callback
def coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
data = self._coordinator.websocket.incident_data()
def client_update(self) -> None:
"""Handle updated data from the data client."""
data = self._client.websocket.incident_data()
if not data or "body" not in data:
return