switch.py

Go to the documentation of this file.

"""Switch platform for FireServiceRota integration."""
 
import logging
from typing import Any
 
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
from .const import DATA_CLIENT, DATA_COORDINATOR, DOMAIN as FIRESERVICEROTA_DOMAIN
 
_LOGGER = logging.getLogger(__name__)
 
 
async def async_setup_entry(
    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
    """Set up FireServiceRota switch based on a config entry."""
    client = hass.data[FIRESERVICEROTA_DOMAIN][entry.entry_id][DATA_CLIENT]
 
    coordinator = hass.data[FIRESERVICEROTA_DOMAIN][entry.entry_id][DATA_COORDINATOR]
 
    async_add_entities([ResponseSwitch(coordinator, client, entry)])
 
 
class ResponseSwitch(SwitchEntity):
    """Representation of an FireServiceRota switch."""
 
    _attr_should_poll = False
    _attr_has_entity_name = True
    _attr_translation_key = "incident_response"
 
    def __init__(self, coordinator, client, entry):
        """Initialize."""
        self._coordinator_coordinator = coordinator
        self._client_client = client
        self._attr_unique_id_attr_unique_id = f"{entry.unique_id}_Response"
        self._entry_id_entry_id = entry.entry_id
 
        self._state_state = None
        self._state_attributes_state_attributes = {}
        self._state_icon_state_icon = None
 
    @property
    def icon(self) -> str:
        """Return the icon to use in the frontend."""
        if self._state_icon_state_icon == "acknowledged":
            return "mdi:run-fast"
        if self._state_icon_state_icon == "rejected":
            return "mdi:account-off-outline"
 
        return "mdi:forum"
 
    @property
    def is_on(self) -> bool:
        """Get the assumed state of the switch."""
        return self._state_state
 
    @property
    def available(self) -> bool:
        """Return if switch is available."""
        return self._client_client.on_duty
 
    @property
    def extra_state_attributes(self) -> dict[str, Any]:
        """Return available attributes for switch."""
        attr: dict[str, Any] = {}
        if not self._state_attributes_state_attributes:
            return attr
 
        data = self._state_attributes_state_attributes
        return {
            key: data[key]
            for key in (
                "user_name",
                "assigned_skill_ids",
                "responded_at",
                "start_time",
                "status",
                "reported_status",
                "arrived_at_station",
                "available_at_incident_creation",
                "active_duty_function_ids",
            )
            if key in data
        }
 
    async def async_turn_on(self, **kwargs: Any) -> None:
        """Send Acknowledge response status."""
        await self.async_set_responseasync_set_response(True)
 
    async def async_turn_off(self, **kwargs: Any) -> None:
        """Send Reject response status."""
        await self.async_set_responseasync_set_response(False)
 
    async def async_set_response(self, value) -> None:
        """Send response status."""
        if not self._client_client.on_duty:
            _LOGGER.debug(
                "Cannot send incident response when not on duty",
            )
            return
 
        await self._client_client.async_set_response(value)
        self.client_updateclient_update()
 
    async def async_added_to_hass(self) -> None:
        """Register update callback."""
        self.async_on_removeasync_on_remove(
            async_dispatcher_connect(
                self.hasshass,
                f"{FIRESERVICEROTA_DOMAIN}_{self._entry_id}_update",
                self.client_updateclient_update,
            )
        )
        self.async_on_removeasync_on_remove(
            self._coordinator_coordinator.async_add_listener(self.async_write_ha_stateasync_write_ha_state)
        )
 
    @callback
    def client_update(self) -> None:
        """Handle updated incident data from the client."""
        self.async_schedule_update_ha_stateasync_schedule_update_ha_state(True)
 
    async def async_update(self) -> None:
        """Update FireServiceRota response data."""
        data = await self._client_client.async_response_update()
 
        if not data or "status" not in data:
            return
 
        self._state_state = data["status"] == "acknowledged"
        self._state_attributes_state_attributes = data
        self._state_icon_state_icon = data["status"]
 
        _LOGGER.debug("Set state of entity 'Response Switch' to '%s'", self._state_state)