Add rest in the inn switch to Habitica integration (#121472)

* Add rest in the inn swich  to Habitica

* Move api call execution to coordinator
This commit is contained in:
Mr. Bubbles 2024-07-08 09:58:33 +02:00 committed by GitHub
parent ad47a7b8c6
commit 6350c5479b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 147 additions and 1 deletions

View File

@ -82,7 +82,7 @@ INSTANCE_LIST_SCHEMA = vol.All(
) )
CONFIG_SCHEMA = vol.Schema({DOMAIN: INSTANCE_LIST_SCHEMA}, extra=vol.ALLOW_EXTRA) CONFIG_SCHEMA = vol.Schema({DOMAIN: INSTANCE_LIST_SCHEMA}, extra=vol.ALLOW_EXTRA)
PLATFORMS = [Platform.BUTTON, Platform.SENSOR, Platform.TODO] PLATFORMS = [Platform.BUTTON, Platform.SENSOR, Platform.SWITCH, Platform.TODO]
SERVICE_API_CALL_SCHEMA = vol.Schema( SERVICE_API_CALL_SCHEMA = vol.Schema(
{ {

View File

@ -2,8 +2,10 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
from datetime import timedelta from datetime import timedelta
from http import HTTPStatus
import logging import logging
from typing import Any from typing import Any
@ -12,6 +14,7 @@ from habitipy.aio import HabitipyAsync
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import ADDITIONAL_USER_FIELDS, DOMAIN from .const import ADDITIONAL_USER_FIELDS, DOMAIN
@ -53,3 +56,23 @@ class HabiticaDataUpdateCoordinator(DataUpdateCoordinator[HabiticaData]):
raise UpdateFailed(f"Error communicating with API: {error}") from error raise UpdateFailed(f"Error communicating with API: {error}") from error
return HabiticaData(user=user_response, tasks=tasks_response) return HabiticaData(user=user_response, tasks=tasks_response)
async def execute(
self, func: Callable[[HabiticaDataUpdateCoordinator], Any]
) -> None:
"""Execute an API call."""
try:
await func(self)
except ClientResponseError as e:
if e.status == HTTPStatus.TOO_MANY_REQUESTS:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="setup_rate_limit_exception",
) from e
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="service_call_exception",
) from e
else:
await self.async_refresh()

View File

@ -65,6 +65,14 @@
"rogue": "mdi:ninja" "rogue": "mdi:ninja"
} }
} }
},
"switch": {
"sleep": {
"default": "mdi:sleep-off",
"state": {
"on": "mdi:sleep"
}
}
} }
}, },
"services": { "services": {

View File

@ -72,6 +72,11 @@
} }
} }
}, },
"switch": {
"sleep": {
"name": "Rest in the inn"
}
},
"todo": { "todo": {
"todos": { "todos": {
"name": "To-Do's" "name": "To-Do's"

View File

@ -0,0 +1,110 @@
"""Switch platform for Habitica integration."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from enum import StrEnum
from typing import TYPE_CHECKING, Any
from homeassistant.components.switch import (
SwitchDeviceClass,
SwitchEntity,
SwitchEntityDescription,
)
from homeassistant.const import CONF_NAME, CONF_URL
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import HabiticaConfigEntry
from .const import DOMAIN, MANUFACTURER, NAME
from .coordinator import HabiticaData, HabiticaDataUpdateCoordinator
@dataclass(kw_only=True, frozen=True)
class HabiticaSwitchEntityDescription(SwitchEntityDescription):
"""Describes Habitica switch entity."""
turn_on_fn: Callable[[HabiticaDataUpdateCoordinator], Any]
turn_off_fn: Callable[[HabiticaDataUpdateCoordinator], Any]
is_on_fn: Callable[[HabiticaData], bool]
class HabiticaSwitchEntity(StrEnum):
"""Habitica switch entities."""
SLEEP = "sleep"
SWTICH_DESCRIPTIONS: tuple[HabiticaSwitchEntityDescription, ...] = (
HabiticaSwitchEntityDescription(
key=HabiticaSwitchEntity.SLEEP,
translation_key=HabiticaSwitchEntity.SLEEP,
device_class=SwitchDeviceClass.SWITCH,
turn_on_fn=lambda coordinator: coordinator.api["user"]["sleep"].post(),
turn_off_fn=lambda coordinator: coordinator.api["user"]["sleep"].post(),
is_on_fn=lambda data: data.user["preferences"]["sleep"],
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: HabiticaConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up switches from a config entry."""
coordinator = entry.runtime_data
async_add_entities(
HabiticaSwitch(coordinator, description) for description in SWTICH_DESCRIPTIONS
)
class HabiticaSwitch(CoordinatorEntity[HabiticaDataUpdateCoordinator], SwitchEntity):
"""Representation of a Habitica Switch."""
_attr_has_entity_name = True
entity_description: HabiticaSwitchEntityDescription
def __init__(
self,
coordinator: HabiticaDataUpdateCoordinator,
entity_description: HabiticaSwitchEntityDescription,
) -> None:
"""Initialize a Habitica switch."""
super().__init__(coordinator)
if TYPE_CHECKING:
assert coordinator.config_entry.unique_id
self.entity_description = entity_description
self._attr_unique_id = (
f"{coordinator.config_entry.unique_id}_{entity_description.key}"
)
self._attr_device_info = DeviceInfo(
entry_type=DeviceEntryType.SERVICE,
manufacturer=MANUFACTURER,
model=NAME,
name=coordinator.config_entry.data[CONF_NAME],
configuration_url=coordinator.config_entry.data[CONF_URL],
identifiers={(DOMAIN, coordinator.config_entry.unique_id)},
)
@property
def is_on(self) -> bool | None:
"""Return the state of the device."""
return self.entity_description.is_on_fn(
self.coordinator.data,
)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the entity on."""
await self.coordinator.execute(self.entity_description.turn_on_fn)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the entity off."""
await self.coordinator.execute(self.entity_description.turn_off_fn)