"""Implement the Google Smart Home traits."""
from __future__ import annotations
from abc import ABC, abstractmethod
from datetime import datetime, timedelta
import logging
from typing import Any
from homeassistant.components import (
alarm_control_panel,
binary_sensor,
button,
camera,
climate,
cover,
event,
fan,
group,
humidifier,
input_boolean,
input_button,
input_select,
lawn_mower,
light,
lock,
media_player,
scene,
script,
select,
sensor,
switch,
vacuum,
valve,
water_heater,
)
from homeassistant.components.alarm_control_panel import (
AlarmControlPanelEntityFeature,
AlarmControlPanelState,
)
from homeassistant.components.camera import CameraEntityFeature
from homeassistant.components.climate import ClimateEntityFeature
from homeassistant.components.cover import CoverEntityFeature
from homeassistant.components.fan import FanEntityFeature
from homeassistant.components.humidifier import HumidifierEntityFeature
from homeassistant.components.lawn_mower import LawnMowerEntityFeature
from homeassistant.components.light import LightEntityFeature
from homeassistant.components.lock import LockState
from homeassistant.components.media_player import MediaPlayerEntityFeature, MediaType
from homeassistant.components.vacuum import VacuumEntityFeature
from homeassistant.components.valve import ValveEntityFeature
from homeassistant.components.water_heater import WaterHeaterEntityFeature
from homeassistant.const import (
ATTR_ASSUMED_STATE,
ATTR_BATTERY_LEVEL,
ATTR_CODE,
ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
ATTR_MODE,
ATTR_SUPPORTED_FEATURES,
ATTR_TEMPERATURE,
CAST_APP_ID_HOMEASSISTANT_MEDIA,
SERVICE_ALARM_ARM_AWAY,
SERVICE_ALARM_ARM_CUSTOM_BYPASS,
SERVICE_ALARM_ARM_HOME,
SERVICE_ALARM_ARM_NIGHT,
SERVICE_ALARM_DISARM,
SERVICE_ALARM_TRIGGER,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_IDLE,
STATE_OFF,
STATE_ON,
STATE_PAUSED,
STATE_PLAYING,
STATE_STANDBY,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
UnitOfTemperature,
)
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from homeassistant.helpers.network import get_url
from homeassistant.util import color as color_util, dt as dt_util
from homeassistant.util.dt import utcnow
from homeassistant.util.percentage import (
ordered_list_item_to_percentage,
percentage_to_ordered_list_item,
)
from homeassistant.util.unit_conversion import TemperatureConverter
from .const import (
CHALLENGE_FAILED_PIN_NEEDED,
CHALLENGE_PIN_NEEDED,
ERR_ALREADY_ARMED,
ERR_ALREADY_DISARMED,
ERR_ALREADY_STOPPED,
ERR_CHALLENGE_NOT_SETUP,
ERR_FUNCTION_NOT_SUPPORTED,
ERR_NO_AVAILABLE_CHANNEL,
ERR_NOT_SUPPORTED,
ERR_UNSUPPORTED_INPUT,
ERR_VALUE_OUT_OF_RANGE,
FAN_SPEEDS,
)
from .error import ChallengeNeeded, SmartHomeError
_LOGGER = logging.getLogger(__name__)
PREFIX_TRAITS = "action.devices.traits."
TRAIT_ARM_DISARM = f"{PREFIX_TRAITS}ArmDisarm"
TRAIT_BRIGHTNESS = f"{PREFIX_TRAITS}Brightness"
TRAIT_CAMERA_STREAM = f"{PREFIX_TRAITS}CameraStream"
TRAIT_CHANNEL = f"{PREFIX_TRAITS}Channel"
TRAIT_COLOR_SETTING = f"{PREFIX_TRAITS}ColorSetting"
TRAIT_DOCK = f"{PREFIX_TRAITS}Dock"
TRAIT_ENERGY_STORAGE = f"{PREFIX_TRAITS}EnergyStorage"
TRAIT_FAN_SPEED = f"{PREFIX_TRAITS}FanSpeed"
TRAIT_HUMIDITY_SETTING = f"{PREFIX_TRAITS}HumiditySetting"
TRAIT_INPUT_SELECTOR = f"{PREFIX_TRAITS}InputSelector"
TRAIT_LOCATOR = f"{PREFIX_TRAITS}Locator"
TRAIT_LOCK_UNLOCK = f"{PREFIX_TRAITS}LockUnlock"
TRAIT_MEDIA_STATE = f"{PREFIX_TRAITS}MediaState"
TRAIT_MODES = f"{PREFIX_TRAITS}Modes"
TRAIT_OBJECT_DETECTION = f"{PREFIX_TRAITS}ObjectDetection"
TRAIT_ON_OFF = f"{PREFIX_TRAITS}OnOff"
TRAIT_OPEN_CLOSE = f"{PREFIX_TRAITS}OpenClose"
TRAIT_SCENE = f"{PREFIX_TRAITS}Scene"
TRAIT_SENSOR_STATE = f"{PREFIX_TRAITS}SensorState"
TRAIT_START_STOP = f"{PREFIX_TRAITS}StartStop"
TRAIT_TEMPERATURE_CONTROL = f"{PREFIX_TRAITS}TemperatureControl"
TRAIT_TEMPERATURE_SETTING = f"{PREFIX_TRAITS}TemperatureSetting"
TRAIT_TRANSPORT_CONTROL = f"{PREFIX_TRAITS}TransportControl"
TRAIT_VOLUME = f"{PREFIX_TRAITS}Volume"
PREFIX_COMMANDS = "action.devices.commands."
COMMAND_ACTIVATE_SCENE = f"{PREFIX_COMMANDS}ActivateScene"
COMMAND_ARM_DISARM = f"{PREFIX_COMMANDS}ArmDisarm"
COMMAND_BRIGHTNESS_ABSOLUTE = f"{PREFIX_COMMANDS}BrightnessAbsolute"
COMMAND_CHARGE = f"{PREFIX_COMMANDS}Charge"
COMMAND_COLOR_ABSOLUTE = f"{PREFIX_COMMANDS}ColorAbsolute"
COMMAND_DOCK = f"{PREFIX_COMMANDS}Dock"
COMMAND_GET_CAMERA_STREAM = f"{PREFIX_COMMANDS}GetCameraStream"
COMMAND_LOCK_UNLOCK = f"{PREFIX_COMMANDS}LockUnlock"
COMMAND_LOCATE = f"{PREFIX_COMMANDS}Locate"
COMMAND_NEXT_INPUT = f"{PREFIX_COMMANDS}NextInput"
COMMAND_MEDIA_NEXT = f"{PREFIX_COMMANDS}mediaNext"
COMMAND_MEDIA_PAUSE = f"{PREFIX_COMMANDS}mediaPause"
COMMAND_MEDIA_PREVIOUS = f"{PREFIX_COMMANDS}mediaPrevious"
COMMAND_MEDIA_RESUME = f"{PREFIX_COMMANDS}mediaResume"
COMMAND_MEDIA_SEEK_RELATIVE = f"{PREFIX_COMMANDS}mediaSeekRelative"
COMMAND_MEDIA_SEEK_TO_POSITION = f"{PREFIX_COMMANDS}mediaSeekToPosition"
COMMAND_MEDIA_SHUFFLE = f"{PREFIX_COMMANDS}mediaShuffle"
COMMAND_MEDIA_STOP = f"{PREFIX_COMMANDS}mediaStop"
COMMAND_MUTE = f"{PREFIX_COMMANDS}mute"
COMMAND_OPEN_CLOSE = f"{PREFIX_COMMANDS}OpenClose"
COMMAND_ON_OFF = f"{PREFIX_COMMANDS}OnOff"
COMMAND_OPEN_CLOSE_RELATIVE = f"{PREFIX_COMMANDS}OpenCloseRelative"
COMMAND_PAUSE_UNPAUSE = f"{PREFIX_COMMANDS}PauseUnpause"
COMMAND_REVERSE = f"{PREFIX_COMMANDS}Reverse"
COMMAND_PREVIOUS_INPUT = f"{PREFIX_COMMANDS}PreviousInput"
COMMAND_SELECT_CHANNEL = f"{PREFIX_COMMANDS}selectChannel"
COMMAND_SET_TEMPERATURE = f"{PREFIX_COMMANDS}SetTemperature"
COMMAND_SET_FAN_SPEED = f"{PREFIX_COMMANDS}SetFanSpeed"
COMMAND_SET_FAN_SPEED_RELATIVE = f"{PREFIX_COMMANDS}SetFanSpeedRelative"
COMMAND_SET_HUMIDITY = f"{PREFIX_COMMANDS}SetHumidity"
COMMAND_SET_INPUT = f"{PREFIX_COMMANDS}SetInput"
COMMAND_SET_MODES = f"{PREFIX_COMMANDS}SetModes"
COMMAND_SET_VOLUME = f"{PREFIX_COMMANDS}setVolume"
COMMAND_START_STOP = f"{PREFIX_COMMANDS}StartStop"
COMMAND_THERMOSTAT_SET_MODE = f"{PREFIX_COMMANDS}ThermostatSetMode"
COMMAND_THERMOSTAT_TEMPERATURE_SETPOINT = (
f"{PREFIX_COMMANDS}ThermostatTemperatureSetpoint"
)
COMMAND_THERMOSTAT_TEMPERATURE_SET_RANGE = (
f"{PREFIX_COMMANDS}ThermostatTemperatureSetRange"
)
COMMAND_VOLUME_RELATIVE = f"{PREFIX_COMMANDS}volumeRelative"
TRAITS: list[type[_Trait]] = []
FAN_SPEED_MAX_SPEED_COUNT = 5
COVER_VALVE_STATES = {
cover.DOMAIN: {
"closed": cover.STATE_CLOSED,
"closing": cover.STATE_CLOSING,
"open": cover.STATE_OPEN,
"opening": cover.STATE_OPENING,
},
valve.DOMAIN: {
"closed": valve.STATE_CLOSED,
"closing": valve.STATE_CLOSING,
"open": valve.STATE_OPEN,
"opening": valve.STATE_OPENING,
},
}
SERVICE_STOP_COVER_VALVE = {
cover.DOMAIN: cover.SERVICE_STOP_COVER,
valve.DOMAIN: valve.SERVICE_STOP_VALVE,
}
SERVICE_OPEN_COVER_VALVE = {
cover.DOMAIN: cover.SERVICE_OPEN_COVER,
valve.DOMAIN: valve.SERVICE_OPEN_VALVE,
}
SERVICE_CLOSE_COVER_VALVE = {
cover.DOMAIN: cover.SERVICE_CLOSE_COVER,
valve.DOMAIN: valve.SERVICE_CLOSE_VALVE,
}
SERVICE_TOGGLE_COVER_VALVE = {
cover.DOMAIN: cover.SERVICE_TOGGLE,
valve.DOMAIN: valve.SERVICE_TOGGLE,
}
SERVICE_SET_POSITION_COVER_VALVE = {
cover.DOMAIN: cover.SERVICE_SET_COVER_POSITION,
valve.DOMAIN: valve.SERVICE_SET_VALVE_POSITION,
}
COVER_VALVE_CURRENT_POSITION = {
cover.DOMAIN: cover.ATTR_CURRENT_POSITION,
valve.DOMAIN: valve.ATTR_CURRENT_POSITION,
}
COVER_VALVE_POSITION = {
cover.DOMAIN: cover.ATTR_POSITION,
valve.DOMAIN: valve.ATTR_POSITION,
}
COVER_VALVE_SET_POSITION_FEATURE = {
cover.DOMAIN: CoverEntityFeature.SET_POSITION,
valve.DOMAIN: ValveEntityFeature.SET_POSITION,
}
COVER_VALVE_STOP_FEATURE = {
cover.DOMAIN: CoverEntityFeature.STOP,
valve.DOMAIN: ValveEntityFeature.STOP,
}
COVER_VALVE_DOMAINS = {cover.DOMAIN, valve.DOMAIN}
FRIENDLY_DOMAIN = {cover.DOMAIN: "Cover", valve.DOMAIN: "Valve"}
def register_trait[_TraitT: _Trait](trait: type[_TraitT]) -> type[_TraitT]:
"""Decorate a class to register a trait."""
TRAITS.append(trait)
return trait
def _google_temp_unit(units):
"""Return Google temperature unit."""
if units == UnitOfTemperature.FAHRENHEIT:
return "F"
return "C"
def _next_selected(items: list[str], selected: str | None) -> str | None:
"""Return the next item in an item list starting at given value.
If selected is missing in items, None is returned
"""
if selected is None:
return None
try:
index = items.index(selected)
except ValueError:
return None
next_item = 0 if index == len(items) - 1 else index + 1
return items[next_item]
class _Trait(ABC):
"""Represents a Trait inside Google Assistant skill."""
name: str
commands: list[str] = []
@staticmethod
def might_2fa(domain, features, device_class):
"""Return if the trait might ask for 2FA."""
return False
@staticmethod
@abstractmethod
def supported(domain, features, device_class, attributes):
"""Test if state is supported."""
def __init__(self, hass: HomeAssistant, state, config) -> None:
"""Initialize a trait for a state."""
self.hass = hass
self.state = state
self.config = config
def sync_attributes(self) -> dict[str, Any]:
"""Return attributes for a sync request."""
raise NotImplementedError
def sync_options(self) -> dict[str, Any]:
"""Add options for the sync request."""
return {}
def query_attributes(self) -> dict[str, Any]:
"""Return the attributes of this trait for this entity."""
raise NotImplementedError
def query_notifications(self) -> dict[str, Any] | None:
"""Return notifications payload."""
def can_execute(self, command, params):
"""Test if command can be executed."""
return command in self.commands
async def execute(self, command, data, params, challenge):
"""Execute a trait command."""
raise NotImplementedError
@register_trait
class BrightnessTrait(_Trait):
"""Trait to control brightness of a device.
https://developers.google.com/actions/smarthome/traits/brightness
"""
name = TRAIT_BRIGHTNESS
commands = [COMMAND_BRIGHTNESS_ABSOLUTE]
@staticmethod
def supported(domain, features, device_class, attributes):
"""Test if state is supported."""
if domain == light.DOMAIN:
color_modes = attributes.get(light.ATTR_SUPPORTED_COLOR_MODES)
return light.brightness_supported(color_modes)
return False
def sync_attributes(self) -> dict[str, Any]:
"""Return brightness attributes for a sync request."""
return {}
def query_attributes(self) -> dict[str, Any]:
"""Return brightness query attributes."""
domain = self.state.domain
response = {}
if domain == light.DOMAIN:
brightness = self.state.attributes.get(light.ATTR_BRIGHTNESS)
if brightness is not None:
response["brightness"] = round(100 * (brightness / 255))
return response
async def execute(self, command, data, params, challenge):
"""Execute a brightness command."""
if self.state.domain == light.DOMAIN:
await self.hass.services.async_call(
light.DOMAIN,
light.SERVICE_TURN_ON,
{
ATTR_ENTITY_ID: self.state.entity_id,
light.ATTR_BRIGHTNESS_PCT: params["brightness"],
},
blocking=not self.config.should_report_state,
context=data.context,
)
@register_trait
class CameraStreamTrait(_Trait):
"""Trait to stream from cameras.
https://developers.google.com/actions/smarthome/traits/camerastream
"""
name = TRAIT_CAMERA_STREAM
commands = [COMMAND_GET_CAMERA_STREAM]
stream_info: dict[str, str] | None = None
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
if domain == camera.DOMAIN:
return features & CameraEntityFeature.STREAM
return False
def sync_attributes(self) -> dict[str, Any]:
"""Return stream attributes for a sync request."""
return {
"cameraStreamSupportedProtocols": ["hls"],
"cameraStreamNeedAuthToken": False,
"cameraStreamNeedDrmEncryption": False,
}
def query_attributes(self) -> dict[str, Any]:
"""Return camera stream attributes."""
return self.stream_info or {}
async def execute(self, command, data, params, challenge):
"""Execute a get camera stream command."""
url = await camera.async_request_stream(self.hass, self.state.entity_id, "hls")
self.stream_info = {
"cameraStreamAccessUrl": f"{get_url(self.hass)}{url}",
"cameraStreamReceiverAppId": CAST_APP_ID_HOMEASSISTANT_MEDIA,
}
@register_trait
class ObjectDetection(_Trait):
"""Trait to object detection.
https://developers.google.com/actions/smarthome/traits/objectdetection
"""
name = TRAIT_OBJECT_DETECTION
commands = []
@staticmethod
def supported(domain, features, device_class, _) -> bool:
"""Test if state is supported."""
return (
domain == event.DOMAIN and device_class == event.EventDeviceClass.DOORBELL
)
def sync_attributes(self) -> dict[str, Any]:
"""Return ObjectDetection attributes for a sync request."""
return {}
def sync_options(self) -> dict[str, Any]:
"""Add options for the sync request."""
return {"notificationSupportedByAgent": True}
def query_attributes(self) -> dict[str, Any]:
"""Return ObjectDetection query attributes."""
return {}
def query_notifications(self) -> dict[str, Any] | None:
"""Return notifications payload."""
if self.state.state in {STATE_UNKNOWN, STATE_UNAVAILABLE}:
return None
# Only notify if last event was less then 30 seconds ago
time_stamp: datetime = datetime.fromisoformat(self.state.state)
if (utcnow() - time_stamp) > timedelta(seconds=30):
return None
# A doorbell event is treated as an object detection of 1 unclassified object.
# The implementation follows the pattern from the Smart Home Doorbell Guide:
# https://developers.home.google.com/cloud-to-cloud/guides/doorbell
# The detectionTimestamp is the time in ms from January 1, 1970, 00:00:00 (UTC)
return {
"ObjectDetection": {
"objects": {
"unclassified": 1,
},
"priority": 0,
"detectionTimestamp": int(time_stamp.timestamp() * 1000),
},
}
async def execute(self, command, data, params, challenge):
"""Execute an ObjectDetection command."""
@register_trait
class OnOffTrait(_Trait):
"""Trait to offer basic on and off functionality.
https://developers.google.com/actions/smarthome/traits/onoff
"""
name = TRAIT_ON_OFF
commands = [COMMAND_ON_OFF]
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
if domain == water_heater.DOMAIN and features & WaterHeaterEntityFeature.ON_OFF:
return True
if domain == climate.DOMAIN and features & (
ClimateEntityFeature.TURN_OFF | ClimateEntityFeature.TURN_ON
):
return True
return domain in (
group.DOMAIN,
input_boolean.DOMAIN,
switch.DOMAIN,
fan.DOMAIN,
light.DOMAIN,
media_player.DOMAIN,
humidifier.DOMAIN,
)
def sync_attributes(self) -> dict[str, Any]:
"""Return OnOff attributes for a sync request."""
if self.state.attributes.get(ATTR_ASSUMED_STATE, False):
return {"commandOnlyOnOff": True}
return {}
def query_attributes(self) -> dict[str, Any]:
"""Return OnOff query attributes."""
return {"on": self.state.state not in (STATE_OFF, STATE_UNKNOWN)}
async def execute(self, command, data, params, challenge):
"""Execute an OnOff command."""
if (domain := self.state.domain) == group.DOMAIN:
service_domain = HOMEASSISTANT_DOMAIN
service = SERVICE_TURN_ON if params["on"] else SERVICE_TURN_OFF
else:
service_domain = domain
service = SERVICE_TURN_ON if params["on"] else SERVICE_TURN_OFF
await self.hass.services.async_call(
service_domain,
service,
{ATTR_ENTITY_ID: self.state.entity_id},
blocking=not self.config.should_report_state,
context=data.context,
)
@register_trait
class ColorSettingTrait(_Trait):
"""Trait to offer color temperature functionality.
https://developers.google.com/actions/smarthome/traits/colortemperature
"""
name = TRAIT_COLOR_SETTING
commands = [COMMAND_COLOR_ABSOLUTE]
@staticmethod
def supported(domain, features, device_class, attributes):
"""Test if state is supported."""
if domain != light.DOMAIN:
return False
color_modes = attributes.get(light.ATTR_SUPPORTED_COLOR_MODES)
return light.color_temp_supported(color_modes) or light.color_supported(
color_modes
)
def sync_attributes(self) -> dict[str, Any]:
"""Return color temperature attributes for a sync request."""
attrs = self.state.attributes
color_modes = attrs.get(light.ATTR_SUPPORTED_COLOR_MODES)
response: dict[str, Any] = {}
if light.color_supported(color_modes):
response["colorModel"] = "hsv"
if light.color_temp_supported(color_modes):
response["colorTemperatureRange"] = {
"temperatureMaxK": int(attrs.get(light.ATTR_MAX_COLOR_TEMP_KELVIN)),
"temperatureMinK": int(attrs.get(light.ATTR_MIN_COLOR_TEMP_KELVIN)),
}
return response
def query_attributes(self) -> dict[str, Any]:
"""Return color temperature query attributes."""
color_mode = self.state.attributes.get(light.ATTR_COLOR_MODE)
color: dict[str, Any] = {}
if light.color_supported([color_mode]):
color_hs = self.state.attributes.get(light.ATTR_HS_COLOR)
brightness = self.state.attributes.get(light.ATTR_BRIGHTNESS, 1)
if color_hs is not None:
color["spectrumHsv"] = {
"hue": color_hs[0],
"saturation": color_hs[1] / 100,
"value": brightness / 255,
}
if light.color_temp_supported([color_mode]):
temp = self.state.attributes.get(light.ATTR_COLOR_TEMP_KELVIN)
# Some faulty integrations might put 0 in here, raising exception.
if temp == 0:
_LOGGER.warning(
"Entity %s has incorrect color temperature %s",
self.state.entity_id,
temp,
)
elif temp is not None:
color["temperatureK"] = temp
response = {}
if color:
response["color"] = color
return response
async def execute(self, command, data, params, challenge):
"""Execute a color temperature command."""
if "temperature" in params["color"]:
temp = params["color"]["temperature"]
max_temp = self.state.attributes[light.ATTR_MAX_COLOR_TEMP_KELVIN]
min_temp = self.state.attributes[light.ATTR_MIN_COLOR_TEMP_KELVIN]
if temp < min_temp or temp > max_temp:
raise SmartHomeError(
ERR_VALUE_OUT_OF_RANGE,
f"Temperature should be between {min_temp} and {max_temp}",
)
await self.hass.services.async_call(
light.DOMAIN,
SERVICE_TURN_ON,
{
ATTR_ENTITY_ID: self.state.entity_id,
light.ATTR_COLOR_TEMP_KELVIN: temp,
},
blocking=not self.config.should_report_state,
context=data.context,
)
elif "spectrumRGB" in params["color"]:
# Convert integer to hex format and left pad with 0's till length 6
hex_value = f"{params['color']['spectrumRGB']:06x}"
color = color_util.color_RGB_to_hs(
*color_util.rgb_hex_to_rgb_list(hex_value)
)
await self.hass.services.async_call(
light.DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: self.state.entity_id, light.ATTR_HS_COLOR: color},
blocking=not self.config.should_report_state,
context=data.context,
)
elif "spectrumHSV" in params["color"]:
color = params["color"]["spectrumHSV"]
saturation = color["saturation"] * 100
brightness = color["value"] * 255
await self.hass.services.async_call(
light.DOMAIN,
SERVICE_TURN_ON,
{
ATTR_ENTITY_ID: self.state.entity_id,
light.ATTR_HS_COLOR: [color["hue"], saturation],
light.ATTR_BRIGHTNESS: brightness,
},
blocking=not self.config.should_report_state,
context=data.context,
)
@register_trait
class SceneTrait(_Trait):
"""Trait to offer scene functionality.
https://developers.google.com/actions/smarthome/traits/scene
"""
name = TRAIT_SCENE
commands = [COMMAND_ACTIVATE_SCENE]
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
return domain in (
button.DOMAIN,
input_button.DOMAIN,
scene.DOMAIN,
script.DOMAIN,
)
def sync_attributes(self) -> dict[str, Any]:
"""Return scene attributes for a sync request."""
# None of the supported domains can support sceneReversible
return {}
def query_attributes(self) -> dict[str, Any]:
"""Return scene query attributes."""
return {}
async def execute(self, command, data, params, challenge):
"""Execute a scene command."""
service = SERVICE_TURN_ON
if self.state.domain == button.DOMAIN:
service = button.SERVICE_PRESS
elif self.state.domain == input_button.DOMAIN:
service = input_button.SERVICE_PRESS
# Don't block for scripts or buttons, as they can be slow.
await self.hass.services.async_call(
self.state.domain,
service,
{ATTR_ENTITY_ID: self.state.entity_id},
blocking=(not self.config.should_report_state)
and self.state.domain
not in (button.DOMAIN, input_button.DOMAIN, script.DOMAIN),
context=data.context,
)
@register_trait
class DockTrait(_Trait):
"""Trait to offer dock functionality.
https://developers.google.com/actions/smarthome/traits/dock
"""
name = TRAIT_DOCK
commands = [COMMAND_DOCK]
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
return domain in (vacuum.DOMAIN, lawn_mower.DOMAIN)
def sync_attributes(self) -> dict[str, Any]:
"""Return dock attributes for a sync request."""
return {}
def query_attributes(self) -> dict[str, Any]:
"""Return dock query attributes."""
domain = self.state.domain
state = self.state.state
if domain == vacuum.DOMAIN:
return {"isDocked": state == vacuum.VacuumActivity.DOCKED}
if domain == lawn_mower.DOMAIN:
return {"isDocked": state == lawn_mower.LawnMowerActivity.DOCKED}
raise NotImplementedError(f"Unsupported domain {domain}")
async def execute(self, command, data, params, challenge):
"""Execute a dock command."""
domain = self.state.domain
service: str | None = None
if domain == vacuum.DOMAIN:
service = vacuum.SERVICE_RETURN_TO_BASE
elif domain == lawn_mower.DOMAIN:
service = lawn_mower.SERVICE_DOCK
if service:
await self.hass.services.async_call(
self.state.domain,
service,
{ATTR_ENTITY_ID: self.state.entity_id},
blocking=not self.config.should_report_state,
context=data.context,
)
@register_trait
class LocatorTrait(_Trait):
"""Trait to offer locate functionality.
https://developers.google.com/actions/smarthome/traits/locator
"""
name = TRAIT_LOCATOR
commands = [COMMAND_LOCATE]
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
return domain == vacuum.DOMAIN and features & VacuumEntityFeature.LOCATE
def sync_attributes(self) -> dict[str, Any]:
"""Return locator attributes for a sync request."""
return {}
def query_attributes(self) -> dict[str, Any]:
"""Return locator query attributes."""
return {}
async def execute(self, command, data, params, challenge):
"""Execute a locate command."""
if params.get("silence", False):
raise SmartHomeError(
ERR_FUNCTION_NOT_SUPPORTED,
"Silencing a Locate request is not yet supported",
)
await self.hass.services.async_call(
self.state.domain,
vacuum.SERVICE_LOCATE,
{ATTR_ENTITY_ID: self.state.entity_id},
blocking=not self.config.should_report_state,
context=data.context,
)
@register_trait
class EnergyStorageTrait(_Trait):
"""Trait to offer EnergyStorage functionality.
https://developers.google.com/actions/smarthome/traits/energystorage
"""
name = TRAIT_ENERGY_STORAGE
commands = [COMMAND_CHARGE]
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
return domain == vacuum.DOMAIN and features & VacuumEntityFeature.BATTERY
def sync_attributes(self) -> dict[str, Any]:
"""Return EnergyStorage attributes for a sync request."""
return {
"isRechargeable": True,
"queryOnlyEnergyStorage": True,
}
def query_attributes(self) -> dict[str, Any]:
"""Return EnergyStorage query attributes."""
battery_level = self.state.attributes.get(ATTR_BATTERY_LEVEL)
if battery_level is None:
return {}
if battery_level == 100:
descriptive_capacity_remaining = "FULL"
elif 75 <= battery_level < 100:
descriptive_capacity_remaining = "HIGH"
elif 50 <= battery_level < 75:
descriptive_capacity_remaining = "MEDIUM"
elif 25 <= battery_level < 50:
descriptive_capacity_remaining = "LOW"
elif 0 <= battery_level < 25:
descriptive_capacity_remaining = "CRITICALLY_LOW"
return {
"descriptiveCapacityRemaining": descriptive_capacity_remaining,
"capacityRemaining": [{"rawValue": battery_level, "unit": "PERCENTAGE"}],
"capacityUntilFull": [
{"rawValue": 100 - battery_level, "unit": "PERCENTAGE"}
],
"isCharging": self.state.state == vacuum.VacuumActivity.DOCKED,
"isPluggedIn": self.state.state == vacuum.VacuumActivity.DOCKED,
}
async def execute(self, command, data, params, challenge):
"""Execute a dock command."""
raise SmartHomeError(
ERR_FUNCTION_NOT_SUPPORTED,
"Controlling charging of a vacuum is not yet supported",
)
@register_trait
class StartStopTrait(_Trait):
"""Trait to offer StartStop functionality.
https://developers.google.com/actions/smarthome/traits/startstop
"""
name = TRAIT_START_STOP
commands = [COMMAND_START_STOP, COMMAND_PAUSE_UNPAUSE]
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
if domain in (vacuum.DOMAIN, lawn_mower.DOMAIN):
return True
if (
domain in COVER_VALVE_DOMAINS
and features & COVER_VALVE_STOP_FEATURE[domain]
):
return True
return False
def sync_attributes(self) -> dict[str, Any]:
"""Return StartStop attributes for a sync request."""
domain = self.state.domain
if domain == vacuum.DOMAIN:
return {
"pausable": self.state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
& VacuumEntityFeature.PAUSE
!= 0
}
if domain == lawn_mower.DOMAIN:
return {
"pausable": self.state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
& LawnMowerEntityFeature.PAUSE
!= 0
}
if domain in COVER_VALVE_DOMAINS:
return {}
raise NotImplementedError(f"Unsupported domain {domain}")
def query_attributes(self) -> dict[str, Any]:
"""Return StartStop query attributes."""
domain = self.state.domain
state = self.state.state
if domain == vacuum.DOMAIN:
return {
"isRunning": state == vacuum.VacuumActivity.CLEANING,
"isPaused": state == vacuum.VacuumActivity.PAUSED,
}
if domain == lawn_mower.DOMAIN:
return {
"isRunning": state == lawn_mower.LawnMowerActivity.MOWING,
"isPaused": state == lawn_mower.LawnMowerActivity.PAUSED,
}
if domain in COVER_VALVE_DOMAINS:
return {
"isRunning": state
in (
COVER_VALVE_STATES[domain]["closing"],
COVER_VALVE_STATES[domain]["opening"],
)
}
raise NotImplementedError(f"Unsupported domain {domain}")
async def execute(self, command, data, params, challenge):
"""Execute a StartStop command."""
domain = self.state.domain
if domain == vacuum.DOMAIN:
await self._execute_vacuum(command, data, params, challenge)
return
if domain == lawn_mower.DOMAIN:
await self._execute_lawn_mower(command, data, params, challenge)
return
if domain in COVER_VALVE_DOMAINS:
await self._execute_cover_or_valve(command, data, params, challenge)
return
async def _execute_vacuum(self, command, data, params, challenge):
"""Execute a StartStop command."""
service: str | None = None
if command == COMMAND_START_STOP:
service = vacuum.SERVICE_START if params["start"] else vacuum.SERVICE_STOP
elif command == COMMAND_PAUSE_UNPAUSE:
service = vacuum.SERVICE_PAUSE if params["pause"] else vacuum.SERVICE_START
if service:
await self.hass.services.async_call(
self.state.domain,
service,
{ATTR_ENTITY_ID: self.state.entity_id},
blocking=not self.config.should_report_state,
context=data.context,
)
async def _execute_lawn_mower(self, command, data, params, challenge):
"""Execute a StartStop command."""
service: str | None = None
if command == COMMAND_START_STOP:
service = (
lawn_mower.SERVICE_START_MOWING
if params["start"]
else lawn_mower.SERVICE_DOCK
)
elif command == COMMAND_PAUSE_UNPAUSE:
service = (
lawn_mower.SERVICE_PAUSE
if params["pause"]
else lawn_mower.SERVICE_START_MOWING
)
if service:
await self.hass.services.async_call(
self.state.domain,
service,
{ATTR_ENTITY_ID: self.state.entity_id},
blocking=not self.config.should_report_state,
context=data.context,
)
async def _execute_cover_or_valve(self, command, data, params, challenge):
"""Execute a StartStop command."""
domain = self.state.domain
if command == COMMAND_START_STOP:
if params["start"] is False:
if self.state.state in (
COVER_VALVE_STATES[domain]["closing"],
COVER_VALVE_STATES[domain]["opening"],
) or self.state.attributes.get(ATTR_ASSUMED_STATE):
await self.hass.services.async_call(
domain,
SERVICE_STOP_COVER_VALVE[domain],
{ATTR_ENTITY_ID: self.state.entity_id},
blocking=not self.config.should_report_state,
context=data.context,
)
else:
raise SmartHomeError(
ERR_ALREADY_STOPPED,
f"{FRIENDLY_DOMAIN[domain]} is already stopped",
)
else:
await self.hass.services.async_call(
domain,
SERVICE_TOGGLE_COVER_VALVE[domain],
{ATTR_ENTITY_ID: self.state.entity_id},
blocking=not self.config.should_report_state,
context=data.context,
)
else:
raise SmartHomeError(
ERR_NOT_SUPPORTED, f"Command {command} is not supported"
)
@register_trait
class TemperatureControlTrait(_Trait):
"""Trait for devices (other than thermostats) that support controlling temperature.
Control the target temperature of water heaters.
Offers a workaround for Temperature sensors by setting queryOnlyTemperatureControl
in the response.
https://developers.google.com/assistant/smarthome/traits/temperaturecontrol
"""
name = TRAIT_TEMPERATURE_CONTROL
commands = [
COMMAND_SET_TEMPERATURE,
]
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
return (
domain == water_heater.DOMAIN
and features & WaterHeaterEntityFeature.TARGET_TEMPERATURE
) or (
domain == sensor.DOMAIN
and device_class == sensor.SensorDeviceClass.TEMPERATURE
)
def sync_attributes(self) -> dict[str, Any]:
"""Return temperature attributes for a sync request."""
response = {}
domain = self.state.domain
attrs = self.state.attributes
unit = self.hass.config.units.temperature_unit
response["temperatureUnitForUX"] = _google_temp_unit(unit)
if domain == water_heater.DOMAIN:
min_temp = round(
TemperatureConverter.convert(
float(attrs[water_heater.ATTR_MIN_TEMP]),
unit,
UnitOfTemperature.CELSIUS,
)
)
max_temp = round(
TemperatureConverter.convert(
float(attrs[water_heater.ATTR_MAX_TEMP]),
unit,
UnitOfTemperature.CELSIUS,
)
)
response["temperatureRange"] = {
"minThresholdCelsius": min_temp,
"maxThresholdCelsius": max_temp,
}
else:
response["queryOnlyTemperatureControl"] = True
response["temperatureRange"] = {
"minThresholdCelsius": -100,
"maxThresholdCelsius": 100,
}
return response
def query_attributes(self) -> dict[str, Any]:
"""Return temperature states."""
response = {}
domain = self.state.domain
unit = self.hass.config.units.temperature_unit
if domain == water_heater.DOMAIN:
target_temp = self.state.attributes[water_heater.ATTR_TEMPERATURE]
current_temp = self.state.attributes[water_heater.ATTR_CURRENT_TEMPERATURE]
if target_temp not in (STATE_UNKNOWN, STATE_UNAVAILABLE):
response["temperatureSetpointCelsius"] = round(
TemperatureConverter.convert(
float(target_temp),
unit,
UnitOfTemperature.CELSIUS,
),
1,
)
if current_temp is not None:
response["temperatureAmbientCelsius"] = round(
TemperatureConverter.convert(
float(current_temp),
unit,
UnitOfTemperature.CELSIUS,
),
1,
)
return response
# domain == sensor.DOMAIN
current_temp = self.state.state
if current_temp not in (STATE_UNKNOWN, STATE_UNAVAILABLE):
temp = round(
TemperatureConverter.convert(
float(current_temp), unit, UnitOfTemperature.CELSIUS
),
1,
)
response["temperatureSetpointCelsius"] = temp
response["temperatureAmbientCelsius"] = temp
return response
async def execute(self, command, data, params, challenge):
"""Execute a temperature point or mode command."""
# All sent in temperatures are always in Celsius
domain = self.state.domain
unit = self.hass.config.units.temperature_unit
if domain == water_heater.DOMAIN and command == COMMAND_SET_TEMPERATURE:
min_temp = self.state.attributes[water_heater.ATTR_MIN_TEMP]
max_temp = self.state.attributes[water_heater.ATTR_MAX_TEMP]
temp = TemperatureConverter.convert(
params["temperature"], UnitOfTemperature.CELSIUS, unit
)
if unit == UnitOfTemperature.FAHRENHEIT:
temp = round(temp)
if temp < min_temp or temp > max_temp:
raise SmartHomeError(
ERR_VALUE_OUT_OF_RANGE,
f"Temperature should be between {min_temp} and {max_temp}",
)
await self.hass.services.async_call(
water_heater.DOMAIN,
water_heater.SERVICE_SET_TEMPERATURE,
{ATTR_ENTITY_ID: self.state.entity_id, ATTR_TEMPERATURE: temp},
blocking=not self.config.should_report_state,
context=data.context,
)
return
raise SmartHomeError(ERR_NOT_SUPPORTED, f"Execute is not supported by {domain}")
@register_trait
class TemperatureSettingTrait(_Trait):
"""Trait to offer handling both temperature point and modes functionality.
https://developers.google.com/actions/smarthome/traits/temperaturesetting
"""
name = TRAIT_TEMPERATURE_SETTING
commands = [
COMMAND_THERMOSTAT_TEMPERATURE_SETPOINT,
COMMAND_THERMOSTAT_TEMPERATURE_SET_RANGE,
COMMAND_THERMOSTAT_SET_MODE,
]
# We do not support "on" as we are unable to know how to restore
# the last mode.
hvac_to_google = {
climate.HVACMode.HEAT: "heat",
climate.HVACMode.COOL: "cool",
climate.HVACMode.OFF: "off",
climate.HVACMode.AUTO: "auto",
climate.HVACMode.HEAT_COOL: "heatcool",
climate.HVACMode.FAN_ONLY: "fan-only",
climate.HVACMode.DRY: "dry",
}
google_to_hvac = {value: key for key, value in hvac_to_google.items()}
preset_to_google = {climate.PRESET_ECO: "eco"}
google_to_preset = {value: key for key, value in preset_to_google.items()}
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
return domain == climate.DOMAIN
@property
def climate_google_modes(self):
"""Return supported Google modes."""
modes = []
attrs = self.state.attributes
for mode in attrs.get(climate.ATTR_HVAC_MODES) or []:
google_mode = self.hvac_to_google.get(mode)
if google_mode and google_mode not in modes:
modes.append(google_mode)
for preset in attrs.get(climate.ATTR_PRESET_MODES) or []:
google_mode = self.preset_to_google.get(preset)
if google_mode and google_mode not in modes:
modes.append(google_mode)
return modes
def sync_attributes(self) -> dict[str, Any]:
"""Return temperature point and modes attributes for a sync request."""
response = {}
attrs = self.state.attributes
unit = self.hass.config.units.temperature_unit
response["thermostatTemperatureUnit"] = _google_temp_unit(unit)
min_temp = round(
TemperatureConverter.convert(
float(attrs[climate.ATTR_MIN_TEMP]),
unit,
UnitOfTemperature.CELSIUS,
)
)
max_temp = round(
TemperatureConverter.convert(
float(attrs[climate.ATTR_MAX_TEMP]),
unit,
UnitOfTemperature.CELSIUS,
)
)
response["thermostatTemperatureRange"] = {
"minThresholdCelsius": min_temp,
"maxThresholdCelsius": max_temp,
}
modes = self.climate_google_modes
# Some integrations don't support modes (e.g. opentherm), but Google doesn't
# support changing the temperature if we don't have any modes. If there's
# only one Google doesn't support changing it, so the default mode here is
# only cosmetic.
if len(modes) == 0:
modes.append("heat")
if "off" in modes and any(
mode in modes for mode in ("heatcool", "heat", "cool")
):
modes.append("on")
response["availableThermostatModes"] = modes
return response
def query_attributes(self) -> dict[str, Any]:
"""Return temperature point and modes query attributes."""
response: dict[str, Any] = {}
attrs = self.state.attributes
unit = self.hass.config.units.temperature_unit
operation = self.state.state
preset = attrs.get(climate.ATTR_PRESET_MODE)
supported = attrs.get(ATTR_SUPPORTED_FEATURES, 0)
if preset in self.preset_to_google:
response["thermostatMode"] = self.preset_to_google[preset]
else:
response["thermostatMode"] = self.hvac_to_google.get(operation, "none")
current_temp = attrs.get(climate.ATTR_CURRENT_TEMPERATURE)
if current_temp is not None:
response["thermostatTemperatureAmbient"] = round(
TemperatureConverter.convert(
current_temp, unit, UnitOfTemperature.CELSIUS
),
1,
)
current_humidity = attrs.get(climate.ATTR_CURRENT_HUMIDITY)
if current_humidity is not None:
response["thermostatHumidityAmbient"] = current_humidity
if operation in (climate.HVACMode.AUTO, climate.HVACMode.HEAT_COOL):
if supported & ClimateEntityFeature.TARGET_TEMPERATURE_RANGE:
response["thermostatTemperatureSetpointHigh"] = round(
TemperatureConverter.convert(
attrs[climate.ATTR_TARGET_TEMP_HIGH],
unit,
UnitOfTemperature.CELSIUS,
),
1,
)
response["thermostatTemperatureSetpointLow"] = round(
TemperatureConverter.convert(
attrs[climate.ATTR_TARGET_TEMP_LOW],
unit,
UnitOfTemperature.CELSIUS,
),
1,
)
elif (target_temp := attrs.get(ATTR_TEMPERATURE)) is not None:
target_temp = round(
TemperatureConverter.convert(
target_temp, unit, UnitOfTemperature.CELSIUS
),
1,
)
response["thermostatTemperatureSetpointHigh"] = target_temp
response["thermostatTemperatureSetpointLow"] = target_temp
elif (target_temp := attrs.get(ATTR_TEMPERATURE)) is not None:
response["thermostatTemperatureSetpoint"] = round(
TemperatureConverter.convert(
target_temp, unit, UnitOfTemperature.CELSIUS
),
1,
)
return response
async def execute(self, command, data, params, challenge):
"""Execute a temperature point or mode command."""
# All sent in temperatures are always in Celsius
unit = self.hass.config.units.temperature_unit
min_temp = self.state.attributes[climate.ATTR_MIN_TEMP]
max_temp = self.state.attributes[climate.ATTR_MAX_TEMP]
if command == COMMAND_THERMOSTAT_TEMPERATURE_SETPOINT:
temp = TemperatureConverter.convert(
params["thermostatTemperatureSetpoint"], UnitOfTemperature.CELSIUS, unit
)
if unit == UnitOfTemperature.FAHRENHEIT:
temp = round(temp)
if temp < min_temp or temp > max_temp:
raise SmartHomeError(
ERR_VALUE_OUT_OF_RANGE,
f"Temperature should be between {min_temp} and {max_temp}",
)
await self.hass.services.async_call(
climate.DOMAIN,
climate.SERVICE_SET_TEMPERATURE,
{ATTR_ENTITY_ID: self.state.entity_id, ATTR_TEMPERATURE: temp},
blocking=not self.config.should_report_state,
context=data.context,
)
elif command == COMMAND_THERMOSTAT_TEMPERATURE_SET_RANGE:
temp_high = TemperatureConverter.convert(
params["thermostatTemperatureSetpointHigh"],
UnitOfTemperature.CELSIUS,
unit,
)
if unit == UnitOfTemperature.FAHRENHEIT:
temp_high = round(temp_high)
if temp_high < min_temp or temp_high > max_temp:
raise SmartHomeError(
ERR_VALUE_OUT_OF_RANGE,
(
"Upper bound for temperature range should be between "
f"{min_temp} and {max_temp}"
),
)
temp_low = TemperatureConverter.convert(
params["thermostatTemperatureSetpointLow"],
UnitOfTemperature.CELSIUS,
unit,
)
if unit == UnitOfTemperature.FAHRENHEIT:
temp_low = round(temp_low)
if temp_low < min_temp or temp_low > max_temp:
raise SmartHomeError(
ERR_VALUE_OUT_OF_RANGE,
(
"Lower bound for temperature range should be between "
f"{min_temp} and {max_temp}"
),
)
supported = self.state.attributes.get(ATTR_SUPPORTED_FEATURES)
svc_data = {ATTR_ENTITY_ID: self.state.entity_id}
if supported & ClimateEntityFeature.TARGET_TEMPERATURE_RANGE:
svc_data[climate.ATTR_TARGET_TEMP_HIGH] = temp_high
svc_data[climate.ATTR_TARGET_TEMP_LOW] = temp_low
else:
svc_data[ATTR_TEMPERATURE] = (temp_high + temp_low) / 2
await self.hass.services.async_call(
climate.DOMAIN,
climate.SERVICE_SET_TEMPERATURE,
svc_data,
blocking=not self.config.should_report_state,
context=data.context,
)
elif command == COMMAND_THERMOSTAT_SET_MODE:
target_mode = params["thermostatMode"]
supported = self.state.attributes.get(ATTR_SUPPORTED_FEATURES)
if target_mode == "on":
await self.hass.services.async_call(
climate.DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: self.state.entity_id},
blocking=not self.config.should_report_state,
context=data.context,
)
return
if target_mode == "off":
await self.hass.services.async_call(
climate.DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: self.state.entity_id},
blocking=not self.config.should_report_state,
context=data.context,
)
return
if target_mode in self.google_to_preset:
await self.hass.services.async_call(
climate.DOMAIN,
climate.SERVICE_SET_PRESET_MODE,
{
climate.ATTR_PRESET_MODE: self.google_to_preset[target_mode],
ATTR_ENTITY_ID: self.state.entity_id,
},
blocking=not self.config.should_report_state,
context=data.context,
)
return
await self.hass.services.async_call(
climate.DOMAIN,
climate.SERVICE_SET_HVAC_MODE,
{
ATTR_ENTITY_ID: self.state.entity_id,
climate.ATTR_HVAC_MODE: self.google_to_hvac[target_mode],
},
blocking=not self.config.should_report_state,
context=data.context,
)
@register_trait
class HumiditySettingTrait(_Trait):
"""Trait to offer humidity setting functionality.
https://developers.google.com/actions/smarthome/traits/humiditysetting
"""
name = TRAIT_HUMIDITY_SETTING
commands = [COMMAND_SET_HUMIDITY]
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
if domain == humidifier.DOMAIN:
return True
return (
domain == sensor.DOMAIN
and device_class == sensor.SensorDeviceClass.HUMIDITY
)
def sync_attributes(self) -> dict[str, Any]:
"""Return humidity attributes for a sync request."""
response: dict[str, Any] = {}
attrs = self.state.attributes
domain = self.state.domain
if domain == sensor.DOMAIN:
device_class = attrs.get(ATTR_DEVICE_CLASS)
if device_class == sensor.SensorDeviceClass.HUMIDITY:
response["queryOnlyHumiditySetting"] = True
elif domain == humidifier.DOMAIN:
response["humiditySetpointRange"] = {
"minPercent": round(
float(self.state.attributes[humidifier.ATTR_MIN_HUMIDITY])
),
"maxPercent": round(
float(self.state.attributes[humidifier.ATTR_MAX_HUMIDITY])
),
}
return response
def query_attributes(self) -> dict[str, Any]:
"""Return humidity query attributes."""
response = {}
attrs = self.state.attributes
domain = self.state.domain
if domain == sensor.DOMAIN:
device_class = attrs.get(ATTR_DEVICE_CLASS)
if device_class == sensor.SensorDeviceClass.HUMIDITY:
humidity_state = self.state.state
if humidity_state not in (STATE_UNKNOWN, STATE_UNAVAILABLE):
response["humidityAmbientPercent"] = round(float(humidity_state))
elif domain == humidifier.DOMAIN:
target_humidity: int | None = attrs.get(humidifier.ATTR_HUMIDITY)
if target_humidity is not None:
response["humiditySetpointPercent"] = target_humidity
current_humidity: int | None = attrs.get(humidifier.ATTR_CURRENT_HUMIDITY)
if current_humidity is not None:
response["humidityAmbientPercent"] = current_humidity
return response
async def execute(self, command, data, params, challenge):
"""Execute a humidity command."""
if self.state.domain == sensor.DOMAIN:
raise SmartHomeError(
ERR_NOT_SUPPORTED, "Execute is not supported by sensor"
)
if command == COMMAND_SET_HUMIDITY:
await self.hass.services.async_call(
humidifier.DOMAIN,
humidifier.SERVICE_SET_HUMIDITY,
{
ATTR_ENTITY_ID: self.state.entity_id,
humidifier.ATTR_HUMIDITY: params["humidity"],
},
blocking=not self.config.should_report_state,
context=data.context,
)
@register_trait
class LockUnlockTrait(_Trait):
"""Trait to lock or unlock a lock.
https://developers.google.com/actions/smarthome/traits/lockunlock
"""
name = TRAIT_LOCK_UNLOCK
commands = [COMMAND_LOCK_UNLOCK]
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
return domain == lock.DOMAIN
@staticmethod
def might_2fa(domain, features, device_class):
"""Return if the trait might ask for 2FA."""
return True
def sync_attributes(self) -> dict[str, Any]:
"""Return LockUnlock attributes for a sync request."""
return {}
def query_attributes(self) -> dict[str, Any]:
"""Return LockUnlock query attributes."""
if self.state.state == LockState.JAMMED:
return {"isJammed": True}
# If its unlocking its not yet unlocked so we consider is locked
return {"isLocked": self.state.state in (LockState.UNLOCKING, LockState.LOCKED)}
async def execute(self, command, data, params, challenge):
"""Execute an LockUnlock command."""
if params["lock"]:
service = lock.SERVICE_LOCK
else:
_verify_pin_challenge(data, self.state, challenge)
service = lock.SERVICE_UNLOCK
await self.hass.services.async_call(
lock.DOMAIN,
service,
{ATTR_ENTITY_ID: self.state.entity_id},
blocking=not self.config.should_report_state,
context=data.context,
)
@register_trait
class ArmDisArmTrait(_Trait):
"""Trait to Arm or Disarm a Security System.
https://developers.google.com/actions/smarthome/traits/armdisarm
"""
name = TRAIT_ARM_DISARM
commands = [COMMAND_ARM_DISARM]
state_to_service = {
AlarmControlPanelState.ARMED_HOME: SERVICE_ALARM_ARM_HOME,
AlarmControlPanelState.ARMED_NIGHT: SERVICE_ALARM_ARM_NIGHT,
AlarmControlPanelState.ARMED_AWAY: SERVICE_ALARM_ARM_AWAY,
AlarmControlPanelState.ARMED_CUSTOM_BYPASS: SERVICE_ALARM_ARM_CUSTOM_BYPASS,
AlarmControlPanelState.TRIGGERED: SERVICE_ALARM_TRIGGER,
}
state_to_support = {
AlarmControlPanelState.ARMED_HOME: AlarmControlPanelEntityFeature.ARM_HOME,
AlarmControlPanelState.ARMED_NIGHT: AlarmControlPanelEntityFeature.ARM_NIGHT,
AlarmControlPanelState.ARMED_AWAY: AlarmControlPanelEntityFeature.ARM_AWAY,
AlarmControlPanelState.ARMED_CUSTOM_BYPASS: AlarmControlPanelEntityFeature.ARM_CUSTOM_BYPASS,
AlarmControlPanelState.TRIGGERED: AlarmControlPanelEntityFeature.TRIGGER,
}
"""The list of states to support in increasing security state."""
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
return domain == alarm_control_panel.DOMAIN
@staticmethod
def might_2fa(domain, features, device_class):
"""Return if the trait might ask for 2FA."""
return True
def _supported_states(self):
"""Return supported states."""
features = self.state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
return [
state
for state, required_feature in self.state_to_support.items()
if features & required_feature != 0
]
def _default_arm_state(self):
states = self._supported_states()
if AlarmControlPanelState.TRIGGERED in states:
states.remove(AlarmControlPanelState.TRIGGERED)
if not states:
raise SmartHomeError(ERR_NOT_SUPPORTED, "ArmLevel missing")
return states[0]
def sync_attributes(self) -> dict[str, Any]:
"""Return ArmDisarm attributes for a sync request."""
response = {}
levels = []
for state in self._supported_states():
# level synonyms are generated from state names
# 'armed_away' becomes 'armed away' or 'away'
level_synonym = [state.replace("_", " ")]
if state != AlarmControlPanelState.TRIGGERED:
level_synonym.append(state.split("_")[1])
level = {
"level_name": state,
"level_values": [{"level_synonym": level_synonym, "lang": "en"}],
}
levels.append(level)
response["availableArmLevels"] = {"levels": levels, "ordered": True}
return response
def query_attributes(self) -> dict[str, Any]:
"""Return ArmDisarm query attributes."""
armed_state = self.state.attributes.get("next_state", self.state.state)
if armed_state in self.state_to_service:
return {"isArmed": True, "currentArmLevel": armed_state}
return {
"isArmed": False,
"currentArmLevel": self._default_arm_state(),
}
async def execute(self, command, data, params, challenge):
"""Execute an ArmDisarm command."""
if params["arm"] and not params.get("cancel"):
# If no arm level given, we we arm the first supported
# level in state_to_support.
if not (arm_level := params.get("armLevel")):
arm_level = self._default_arm_state()
if self.state.state == arm_level:
raise SmartHomeError(ERR_ALREADY_ARMED, "System is already armed")
if self.state.attributes["code_arm_required"]:
_verify_pin_challenge(data, self.state, challenge)
service = self.state_to_service[arm_level]
# disarm the system without asking for code when
# 'cancel' arming action is received while current status is pending
elif (
params["arm"]
and params.get("cancel")
and self.state.state == AlarmControlPanelState.PENDING
):
service = SERVICE_ALARM_DISARM
else:
if self.state.state == AlarmControlPanelState.DISARMED:
raise SmartHomeError(ERR_ALREADY_DISARMED, "System is already disarmed")
_verify_pin_challenge(data, self.state, challenge)
service = SERVICE_ALARM_DISARM
await self.hass.services.async_call(
alarm_control_panel.DOMAIN,
service,
{
ATTR_ENTITY_ID: self.state.entity_id,
ATTR_CODE: data.config.secure_devices_pin,
},
blocking=not self.config.should_report_state,
context=data.context,
)
def _get_fan_speed(speed_name: str) -> dict[str, Any]:
"""Return a fan speed synonyms for a speed name."""
speed_synonyms = FAN_SPEEDS.get(speed_name, [f"{speed_name}"])
return {
"speed_name": speed_name,
"speed_values": [
{
"speed_synonym": speed_synonyms,
"lang": "en",
}
],
}
@register_trait
class FanSpeedTrait(_Trait):
"""Trait to control speed of Fan.
https://developers.google.com/actions/smarthome/traits/fanspeed
"""
name = TRAIT_FAN_SPEED
commands = [COMMAND_SET_FAN_SPEED, COMMAND_REVERSE]
def __init__(self, hass, state, config):
"""Initialize a trait for a state."""
super().__init__(hass, state, config)
if state.domain == fan.DOMAIN:
speed_count = min(
FAN_SPEED_MAX_SPEED_COUNT,
round(
100 / (self.state.attributes.get(fan.ATTR_PERCENTAGE_STEP) or 1.0)
),
)
self._ordered_speed = [
f"{speed}/{speed_count}" for speed in range(1, speed_count + 1)
]
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
if domain == fan.DOMAIN:
return features & FanEntityFeature.SET_SPEED
if domain == climate.DOMAIN:
return features & ClimateEntityFeature.FAN_MODE
return False
def sync_attributes(self) -> dict[str, Any]:
"""Return speed point and modes attributes for a sync request."""
domain = self.state.domain
speeds = []
result: dict[str, Any] = {}
if domain == fan.DOMAIN:
reversible = bool(
self.state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
& FanEntityFeature.DIRECTION
)
result.update(
{
"reversible": reversible,
"supportsFanSpeedPercent": True,
}
)
if self._ordered_speed:
result.update(
{
"availableFanSpeeds": {
"speeds": [
_get_fan_speed(speed) for speed in self._ordered_speed
],
"ordered": True,
},
}
)
elif domain == climate.DOMAIN:
modes = self.state.attributes.get(climate.ATTR_FAN_MODES) or []
for mode in modes:
speed = {
"speed_name": mode,
"speed_values": [{"speed_synonym": [mode], "lang": "en"}],
}
speeds.append(speed)
result.update(
{
"reversible": False,
"availableFanSpeeds": {"speeds": speeds, "ordered": True},
}
)
return result
def query_attributes(self) -> dict[str, Any]:
"""Return speed point and modes query attributes."""
attrs = self.state.attributes
domain = self.state.domain
response = {}
if domain == climate.DOMAIN:
speed = attrs.get(climate.ATTR_FAN_MODE) or "off"
response["currentFanSpeedSetting"] = speed
if domain == fan.DOMAIN:
percent = attrs.get(fan.ATTR_PERCENTAGE) or 0
response["currentFanSpeedPercent"] = percent
response["currentFanSpeedSetting"] = percentage_to_ordered_list_item(
self._ordered_speed, percent
)
return response
async def execute_fanspeed(self, data, params):
"""Execute an SetFanSpeed command."""
domain = self.state.domain
if domain == climate.DOMAIN:
await self.hass.services.async_call(
climate.DOMAIN,
climate.SERVICE_SET_FAN_MODE,
{
ATTR_ENTITY_ID: self.state.entity_id,
climate.ATTR_FAN_MODE: params["fanSpeed"],
},
blocking=not self.config.should_report_state,
context=data.context,
)
if domain == fan.DOMAIN:
if fan_speed := params.get("fanSpeed"):
fan_speed_percent = ordered_list_item_to_percentage(
self._ordered_speed, fan_speed
)
else:
fan_speed_percent = params.get("fanSpeedPercent")
await self.hass.services.async_call(
fan.DOMAIN,
fan.SERVICE_SET_PERCENTAGE,
{
ATTR_ENTITY_ID: self.state.entity_id,
fan.ATTR_PERCENTAGE: fan_speed_percent,
},
blocking=not self.config.should_report_state,
context=data.context,
)
async def execute_reverse(self, data, params):
"""Execute a Reverse command."""
if self.state.domain == fan.DOMAIN:
if self.state.attributes.get(fan.ATTR_DIRECTION) == fan.DIRECTION_FORWARD:
direction = fan.DIRECTION_REVERSE
else:
direction = fan.DIRECTION_FORWARD
await self.hass.services.async_call(
fan.DOMAIN,
fan.SERVICE_SET_DIRECTION,
{ATTR_ENTITY_ID: self.state.entity_id, fan.ATTR_DIRECTION: direction},
blocking=not self.config.should_report_state,
context=data.context,
)
async def execute(self, command, data, params, challenge):
"""Execute a smart home command."""
if command == COMMAND_SET_FAN_SPEED:
await self.execute_fanspeed(data, params)
elif command == COMMAND_REVERSE:
await self.execute_reverse(data, params)
@register_trait
class ModesTrait(_Trait):
"""Trait to set modes.
https://developers.google.com/actions/smarthome/traits/modes
"""
name = TRAIT_MODES
commands = [COMMAND_SET_MODES]
SYNONYMS = {
"preset mode": ["preset mode", "mode", "preset"],
"sound mode": ["sound mode", "effects"],
"option": ["option", "setting", "mode", "value"],
}
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
if domain == fan.DOMAIN and features & FanEntityFeature.PRESET_MODE:
return True
if domain == input_select.DOMAIN:
return True
if domain == select.DOMAIN:
return True
if domain == humidifier.DOMAIN and features & HumidifierEntityFeature.MODES:
return True
if domain == light.DOMAIN and features & LightEntityFeature.EFFECT:
return True
if (
domain == water_heater.DOMAIN
and features & WaterHeaterEntityFeature.OPERATION_MODE
):
return True
if domain != media_player.DOMAIN:
return False
return features & MediaPlayerEntityFeature.SELECT_SOUND_MODE
def _generate(self, name, settings):
"""Generate a list of modes."""
mode = {
"name": name,
"name_values": [
{"name_synonym": self.SYNONYMS.get(name, [name]), "lang": "en"}
],
"settings": [],
"ordered": False,
}
for setting in settings:
mode["settings"].append(
{
"setting_name": setting,
"setting_values": [
{
"setting_synonym": self.SYNONYMS.get(setting, [setting]),
"lang": "en",
}
],
}
)
return mode
def sync_attributes(self) -> dict[str, Any]:
"""Return mode attributes for a sync request."""
modes = []
for domain, attr, name in (
(fan.DOMAIN, fan.ATTR_PRESET_MODES, "preset mode"),
(media_player.DOMAIN, media_player.ATTR_SOUND_MODE_LIST, "sound mode"),
(input_select.DOMAIN, input_select.ATTR_OPTIONS, "option"),
(select.DOMAIN, select.ATTR_OPTIONS, "option"),
(humidifier.DOMAIN, humidifier.ATTR_AVAILABLE_MODES, "mode"),
(light.DOMAIN, light.ATTR_EFFECT_LIST, "effect"),
(water_heater.DOMAIN, water_heater.ATTR_OPERATION_LIST, "operation mode"),
):
if self.state.domain != domain:
continue
if (items := self.state.attributes.get(attr)) is not None:
modes.append(self._generate(name, items))
# Shortcut since all domains are currently unique
break
return {"availableModes": modes}
def query_attributes(self) -> dict[str, Any]:
"""Return current modes."""
attrs = self.state.attributes
response: dict[str, Any] = {}
mode_settings = {}
if self.state.domain == fan.DOMAIN:
if fan.ATTR_PRESET_MODES in attrs:
mode_settings["preset mode"] = attrs.get(fan.ATTR_PRESET_MODE)
elif self.state.domain == media_player.DOMAIN:
if media_player.ATTR_SOUND_MODE_LIST in attrs:
mode_settings["sound mode"] = attrs.get(media_player.ATTR_SOUND_MODE)
elif self.state.domain in (input_select.DOMAIN, select.DOMAIN):
mode_settings["option"] = self.state.state
elif self.state.domain == humidifier.DOMAIN:
if ATTR_MODE in attrs:
mode_settings["mode"] = attrs.get(ATTR_MODE)
elif self.state.domain == water_heater.DOMAIN:
if water_heater.ATTR_OPERATION_MODE in attrs:
mode_settings["operation mode"] = attrs.get(
water_heater.ATTR_OPERATION_MODE
)
elif self.state.domain == light.DOMAIN and (
effect := attrs.get(light.ATTR_EFFECT)
):
mode_settings["effect"] = effect
if mode_settings:
response["on"] = self.state.state not in (STATE_OFF, STATE_UNKNOWN)
response["currentModeSettings"] = mode_settings
return response
async def execute(self, command, data, params, challenge):
"""Execute a SetModes command."""
settings = params.get("updateModeSettings")
if self.state.domain == fan.DOMAIN:
preset_mode = settings["preset mode"]
await self.hass.services.async_call(
fan.DOMAIN,
fan.SERVICE_SET_PRESET_MODE,
{
ATTR_ENTITY_ID: self.state.entity_id,
fan.ATTR_PRESET_MODE: preset_mode,
},
blocking=not self.config.should_report_state,
context=data.context,
)
return
if self.state.domain == input_select.DOMAIN:
option = settings["option"]
await self.hass.services.async_call(
input_select.DOMAIN,
input_select.SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: self.state.entity_id,
input_select.ATTR_OPTION: option,
},
blocking=not self.config.should_report_state,
context=data.context,
)
return
if self.state.domain == select.DOMAIN:
option = settings["option"]
await self.hass.services.async_call(
select.DOMAIN,
select.SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: self.state.entity_id,
select.ATTR_OPTION: option,
},
blocking=not self.config.should_report_state,
context=data.context,
)
return
if self.state.domain == humidifier.DOMAIN:
requested_mode = settings["mode"]
await self.hass.services.async_call(
humidifier.DOMAIN,
humidifier.SERVICE_SET_MODE,
{
ATTR_MODE: requested_mode,
ATTR_ENTITY_ID: self.state.entity_id,
},
blocking=not self.config.should_report_state,
context=data.context,
)
return
if self.state.domain == water_heater.DOMAIN:
requested_mode = settings["operation mode"]
await self.hass.services.async_call(
water_heater.DOMAIN,
water_heater.SERVICE_SET_OPERATION_MODE,
{
water_heater.ATTR_OPERATION_MODE: requested_mode,
ATTR_ENTITY_ID: self.state.entity_id,
},
blocking=not self.config.should_report_state,
context=data.context,
)
return
if self.state.domain == light.DOMAIN:
requested_effect = settings["effect"]
await self.hass.services.async_call(
light.DOMAIN,
SERVICE_TURN_ON,
{
ATTR_ENTITY_ID: self.state.entity_id,
light.ATTR_EFFECT: requested_effect,
},
blocking=not self.config.should_report_state,
context=data.context,
)
return
if self.state.domain == media_player.DOMAIN and (
sound_mode := settings.get("sound mode")
):
await self.hass.services.async_call(
media_player.DOMAIN,
media_player.SERVICE_SELECT_SOUND_MODE,
{
ATTR_ENTITY_ID: self.state.entity_id,
media_player.ATTR_SOUND_MODE: sound_mode,
},
blocking=not self.config.should_report_state,
context=data.context,
)
_LOGGER.info(
"Received an Options command for unrecognised domain %s",
self.state.domain,
)
return
@register_trait
class InputSelectorTrait(_Trait):
"""Trait to set modes.
https://developers.google.com/assistant/smarthome/traits/inputselector
"""
name = TRAIT_INPUT_SELECTOR
commands = [COMMAND_SET_INPUT, COMMAND_NEXT_INPUT, COMMAND_PREVIOUS_INPUT]
SYNONYMS: dict[str, list[str]] = {}
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
if domain == media_player.DOMAIN and (
features & MediaPlayerEntityFeature.SELECT_SOURCE
):
return True
return False
def sync_attributes(self) -> dict[str, Any]:
"""Return mode attributes for a sync request."""
attrs = self.state.attributes
sourcelist: list[str] = attrs.get(media_player.ATTR_INPUT_SOURCE_LIST) or []
inputs = [
{"key": source, "names": [{"name_synonym": [source], "lang": "en"}]}
for source in sourcelist
]
return {"availableInputs": inputs, "orderedInputs": True}
def query_attributes(self) -> dict[str, Any]:
"""Return current modes."""
attrs = self.state.attributes
return {"currentInput": attrs.get(media_player.ATTR_INPUT_SOURCE, "")}
async def execute(self, command, data, params, challenge):
"""Execute an SetInputSource command."""
sources = self.state.attributes.get(media_player.ATTR_INPUT_SOURCE_LIST) or []
source = self.state.attributes.get(media_player.ATTR_INPUT_SOURCE)
if command == COMMAND_SET_INPUT:
requested_source = params.get("newInput")
elif command == COMMAND_NEXT_INPUT:
requested_source = _next_selected(sources, source)
elif command == COMMAND_PREVIOUS_INPUT:
requested_source = _next_selected(list(reversed(sources)), source)
else:
raise SmartHomeError(ERR_NOT_SUPPORTED, "Unsupported command")
if requested_source not in sources:
raise SmartHomeError(ERR_UNSUPPORTED_INPUT, "Unsupported input")
await self.hass.services.async_call(
media_player.DOMAIN,
media_player.SERVICE_SELECT_SOURCE,
{
ATTR_ENTITY_ID: self.state.entity_id,
media_player.ATTR_INPUT_SOURCE: requested_source,
},
blocking=not self.config.should_report_state,
context=data.context,
)
@register_trait
class OpenCloseTrait(_Trait):
"""Trait to open and close a cover.
https://developers.google.com/actions/smarthome/traits/openclose
"""
# Cover device classes that require 2FA
COVER_2FA = (
cover.CoverDeviceClass.DOOR,
cover.CoverDeviceClass.GARAGE,
cover.CoverDeviceClass.GATE,
)
name = TRAIT_OPEN_CLOSE
commands = [COMMAND_OPEN_CLOSE, COMMAND_OPEN_CLOSE_RELATIVE]
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
if domain in COVER_VALVE_DOMAINS:
return True
return domain == binary_sensor.DOMAIN and device_class in (
binary_sensor.BinarySensorDeviceClass.DOOR,
binary_sensor.BinarySensorDeviceClass.GARAGE_DOOR,
binary_sensor.BinarySensorDeviceClass.LOCK,
binary_sensor.BinarySensorDeviceClass.OPENING,
binary_sensor.BinarySensorDeviceClass.WINDOW,
)
@staticmethod
def might_2fa(domain, features, device_class):
"""Return if the trait might ask for 2FA."""
return domain == cover.DOMAIN and device_class in OpenCloseTrait.COVER_2FA
def sync_attributes(self) -> dict[str, Any]:
"""Return opening direction."""
response = {}
features = self.state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
if self.state.domain == binary_sensor.DOMAIN:
response["queryOnlyOpenClose"] = True
response["discreteOnlyOpenClose"] = True
elif (
self.state.domain == cover.DOMAIN
and features & CoverEntityFeature.SET_POSITION == 0
):
response["discreteOnlyOpenClose"] = True
if (
features & CoverEntityFeature.OPEN == 0
and features & CoverEntityFeature.CLOSE == 0
):
response["queryOnlyOpenClose"] = True
elif (
self.state.domain == valve.DOMAIN
and features & ValveEntityFeature.SET_POSITION == 0
):
response["discreteOnlyOpenClose"] = True
if (
features & ValveEntityFeature.OPEN == 0
and features & ValveEntityFeature.CLOSE == 0
):
response["queryOnlyOpenClose"] = True
if self.state.attributes.get(ATTR_ASSUMED_STATE):
response["commandOnlyOpenClose"] = True
return response
def query_attributes(self) -> dict[str, Any]:
"""Return state query attributes."""
domain = self.state.domain
response: dict[str, Any] = {}
# When it's an assumed state, we will return empty state
# This shouldn't happen because we set `commandOnlyOpenClose`
# but Google still queries. Erroring here will cause device
# to show up offline.
if self.state.attributes.get(ATTR_ASSUMED_STATE):
return response
if domain in COVER_VALVE_DOMAINS:
if self.state.state == STATE_UNKNOWN:
raise SmartHomeError(
ERR_NOT_SUPPORTED, "Querying state is not supported"
)
position = self.state.attributes.get(COVER_VALVE_CURRENT_POSITION[domain])
if position is not None:
response["openPercent"] = position
elif self.state.state != COVER_VALVE_STATES[domain]["closed"]:
response["openPercent"] = 100
else:
response["openPercent"] = 0
elif domain == binary_sensor.DOMAIN:
if self.state.state == STATE_ON:
response["openPercent"] = 100
else:
response["openPercent"] = 0
return response
async def execute(self, command, data, params, challenge):
"""Execute an Open, close, Set position command."""
domain = self.state.domain
features = self.state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
if domain in COVER_VALVE_DOMAINS:
svc_params = {ATTR_ENTITY_ID: self.state.entity_id}
should_verify = False
if command == COMMAND_OPEN_CLOSE_RELATIVE:
position = self.state.attributes.get(
COVER_VALVE_CURRENT_POSITION[domain]
)
if position is None:
raise SmartHomeError(
ERR_NOT_SUPPORTED,
"Current position not know for relative command",
)
position = max(0, min(100, position + params["openRelativePercent"]))
else:
position = params["openPercent"]
if position == 0:
service = SERVICE_CLOSE_COVER_VALVE[domain]
should_verify = False
elif position == 100:
service = SERVICE_OPEN_COVER_VALVE[domain]
should_verify = True
elif features & COVER_VALVE_SET_POSITION_FEATURE[domain]:
service = SERVICE_SET_POSITION_COVER_VALVE[domain]
if position > 0:
should_verify = True
svc_params[COVER_VALVE_POSITION[domain]] = position
else:
raise SmartHomeError(
ERR_NOT_SUPPORTED, "No support for partial open close"
)
if (
should_verify
and self.state.attributes.get(ATTR_DEVICE_CLASS)
in OpenCloseTrait.COVER_2FA
):
_verify_pin_challenge(data, self.state, challenge)
await self.hass.services.async_call(
domain,
service,
svc_params,
blocking=not self.config.should_report_state,
context=data.context,
)
@register_trait
class VolumeTrait(_Trait):
"""Trait to control volume of a device.
https://developers.google.com/actions/smarthome/traits/volume
"""
name = TRAIT_VOLUME
commands = [COMMAND_SET_VOLUME, COMMAND_VOLUME_RELATIVE, COMMAND_MUTE]
@staticmethod
def supported(domain, features, device_class, _):
"""Test if trait is supported."""
if domain == media_player.DOMAIN:
return features & (
MediaPlayerEntityFeature.VOLUME_SET
| MediaPlayerEntityFeature.VOLUME_STEP
)
return False
def sync_attributes(self) -> dict[str, Any]:
"""Return volume attributes for a sync request."""
features = self.state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
return {
"volumeCanMuteAndUnmute": bool(
features & MediaPlayerEntityFeature.VOLUME_MUTE
),
"commandOnlyVolume": self.state.attributes.get(ATTR_ASSUMED_STATE, False),
# Volume amounts in SET_VOLUME and VOLUME_RELATIVE are on a scale
# from 0 to this value.
"volumeMaxLevel": 100,
# Default change for queries like "Hey Google, volume up".
# 10% corresponds to the default behavior for the
# media_player.volume{up,down} services.
"levelStepSize": 10,
}
def query_attributes(self) -> dict[str, Any]:
"""Return volume query attributes."""
response = {}
level = self.state.attributes.get(media_player.ATTR_MEDIA_VOLUME_LEVEL)
if level is not None:
# Convert 0.0-1.0 to 0-100
response["currentVolume"] = round(level * 100)
muted = self.state.attributes.get(media_player.ATTR_MEDIA_VOLUME_MUTED)
if muted is not None:
response["isMuted"] = bool(muted)
return response
async def _set_volume_absolute(self, data, level):
await self.hass.services.async_call(
media_player.DOMAIN,
media_player.SERVICE_VOLUME_SET,
{
ATTR_ENTITY_ID: self.state.entity_id,
media_player.ATTR_MEDIA_VOLUME_LEVEL: level,
},
blocking=not self.config.should_report_state,
context=data.context,
)
async def _execute_set_volume(self, data, params):
level = max(0, min(100, params["volumeLevel"]))
if not (
self.state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
& MediaPlayerEntityFeature.VOLUME_SET
):
raise SmartHomeError(ERR_NOT_SUPPORTED, "Command not supported")
await self._set_volume_absolute(data, level / 100)
async def _execute_volume_relative(self, data, params):
relative = params["relativeSteps"]
features = self.state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
if features & MediaPlayerEntityFeature.VOLUME_SET:
current = self.state.attributes.get(media_player.ATTR_MEDIA_VOLUME_LEVEL)
target = max(0.0, min(1.0, current + relative / 100))
await self._set_volume_absolute(data, target)
elif features & MediaPlayerEntityFeature.VOLUME_STEP:
svc = media_player.SERVICE_VOLUME_UP
if relative < 0:
svc = media_player.SERVICE_VOLUME_DOWN
relative = -relative
for _ in range(relative):
await self.hass.services.async_call(
media_player.DOMAIN,
svc,
{ATTR_ENTITY_ID: self.state.entity_id},
blocking=not self.config.should_report_state,
context=data.context,
)
else:
raise SmartHomeError(ERR_NOT_SUPPORTED, "Command not supported")
async def _execute_mute(self, data, params):
mute = params["mute"]
if not (
self.state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
& MediaPlayerEntityFeature.VOLUME_MUTE
):
raise SmartHomeError(ERR_NOT_SUPPORTED, "Command not supported")
await self.hass.services.async_call(
media_player.DOMAIN,
media_player.SERVICE_VOLUME_MUTE,
{
ATTR_ENTITY_ID: self.state.entity_id,
media_player.ATTR_MEDIA_VOLUME_MUTED: mute,
},
blocking=not self.config.should_report_state,
context=data.context,
)
async def execute(self, command, data, params, challenge):
"""Execute a volume command."""
if command == COMMAND_SET_VOLUME:
await self._execute_set_volume(data, params)
elif command == COMMAND_VOLUME_RELATIVE:
await self._execute_volume_relative(data, params)
elif command == COMMAND_MUTE:
await self._execute_mute(data, params)
else:
raise SmartHomeError(ERR_NOT_SUPPORTED, "Command not supported")
def _verify_pin_challenge(data, state, challenge):
"""Verify a pin challenge."""
if not data.config.should_2fa(state):
return
if not data.config.secure_devices_pin:
raise SmartHomeError(ERR_CHALLENGE_NOT_SETUP, "Challenge is not set up")
if not challenge:
raise ChallengeNeeded(CHALLENGE_PIN_NEEDED)
if challenge.get("pin") != data.config.secure_devices_pin:
raise ChallengeNeeded(CHALLENGE_FAILED_PIN_NEEDED)
MEDIA_COMMAND_SUPPORT_MAPPING = {
COMMAND_MEDIA_NEXT: MediaPlayerEntityFeature.NEXT_TRACK,
COMMAND_MEDIA_PAUSE: MediaPlayerEntityFeature.PAUSE,
COMMAND_MEDIA_PREVIOUS: MediaPlayerEntityFeature.PREVIOUS_TRACK,
COMMAND_MEDIA_RESUME: MediaPlayerEntityFeature.PLAY,
COMMAND_MEDIA_SEEK_RELATIVE: MediaPlayerEntityFeature.SEEK,
COMMAND_MEDIA_SEEK_TO_POSITION: MediaPlayerEntityFeature.SEEK,
COMMAND_MEDIA_SHUFFLE: MediaPlayerEntityFeature.SHUFFLE_SET,
COMMAND_MEDIA_STOP: MediaPlayerEntityFeature.STOP,
}
MEDIA_COMMAND_ATTRIBUTES = {
COMMAND_MEDIA_NEXT: "NEXT",
COMMAND_MEDIA_PAUSE: "PAUSE",
COMMAND_MEDIA_PREVIOUS: "PREVIOUS",
COMMAND_MEDIA_RESUME: "RESUME",
COMMAND_MEDIA_SEEK_RELATIVE: "SEEK_RELATIVE",
COMMAND_MEDIA_SEEK_TO_POSITION: "SEEK_TO_POSITION",
COMMAND_MEDIA_SHUFFLE: "SHUFFLE",
COMMAND_MEDIA_STOP: "STOP",
}
@register_trait
class TransportControlTrait(_Trait):
"""Trait to control media playback.
https://developers.google.com/actions/smarthome/traits/transportcontrol
"""
name = TRAIT_TRANSPORT_CONTROL
commands = [
COMMAND_MEDIA_NEXT,
COMMAND_MEDIA_PAUSE,
COMMAND_MEDIA_PREVIOUS,
COMMAND_MEDIA_RESUME,
COMMAND_MEDIA_SEEK_RELATIVE,
COMMAND_MEDIA_SEEK_TO_POSITION,
COMMAND_MEDIA_SHUFFLE,
COMMAND_MEDIA_STOP,
]
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
if domain == media_player.DOMAIN:
for feature in MEDIA_COMMAND_SUPPORT_MAPPING.values():
if features & feature:
return True
return False
def sync_attributes(self) -> dict[str, Any]:
"""Return opening direction."""
response = {}
if self.state.domain == media_player.DOMAIN:
features = self.state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
support = []
for command, feature in MEDIA_COMMAND_SUPPORT_MAPPING.items():
if features & feature:
support.append(MEDIA_COMMAND_ATTRIBUTES[command])
response["transportControlSupportedCommands"] = support
return response
def query_attributes(self) -> dict[str, Any]:
"""Return the attributes of this trait for this entity."""
return {}
async def execute(self, command, data, params, challenge):
"""Execute a media command."""
service_attrs = {ATTR_ENTITY_ID: self.state.entity_id}
if command == COMMAND_MEDIA_SEEK_RELATIVE:
service = media_player.SERVICE_MEDIA_SEEK
rel_position = params["relativePositionMs"] / 1000
seconds_since = 0 # Default to 0 seconds
if self.state.state == STATE_PLAYING:
now = dt_util.utcnow()
upd_at = self.state.attributes.get(
media_player.ATTR_MEDIA_POSITION_UPDATED_AT, now
)
seconds_since = (now - upd_at).total_seconds()
position = self.state.attributes.get(media_player.ATTR_MEDIA_POSITION, 0)
max_position = self.state.attributes.get(
media_player.ATTR_MEDIA_DURATION, 0
)
service_attrs[media_player.ATTR_MEDIA_SEEK_POSITION] = min(
max(position + seconds_since + rel_position, 0), max_position
)
elif command == COMMAND_MEDIA_SEEK_TO_POSITION:
service = media_player.SERVICE_MEDIA_SEEK
max_position = self.state.attributes.get(
media_player.ATTR_MEDIA_DURATION, 0
)
service_attrs[media_player.ATTR_MEDIA_SEEK_POSITION] = min(
max(params["absPositionMs"] / 1000, 0), max_position
)
elif command == COMMAND_MEDIA_NEXT:
service = media_player.SERVICE_MEDIA_NEXT_TRACK
elif command == COMMAND_MEDIA_PAUSE:
service = media_player.SERVICE_MEDIA_PAUSE
elif command == COMMAND_MEDIA_PREVIOUS:
service = media_player.SERVICE_MEDIA_PREVIOUS_TRACK
elif command == COMMAND_MEDIA_RESUME:
service = media_player.SERVICE_MEDIA_PLAY
elif command == COMMAND_MEDIA_SHUFFLE:
service = media_player.SERVICE_SHUFFLE_SET
# Google Assistant only supports enabling shuffle
service_attrs[media_player.ATTR_MEDIA_SHUFFLE] = True
elif command == COMMAND_MEDIA_STOP:
service = media_player.SERVICE_MEDIA_STOP
else:
raise SmartHomeError(ERR_NOT_SUPPORTED, "Command not supported")
await self.hass.services.async_call(
media_player.DOMAIN,
service,
service_attrs,
blocking=not self.config.should_report_state,
context=data.context,
)
@register_trait
class MediaStateTrait(_Trait):
"""Trait to get media playback state.
https://developers.google.com/actions/smarthome/traits/mediastate
"""
name = TRAIT_MEDIA_STATE
commands: list[str] = []
activity_lookup = {
STATE_OFF: "INACTIVE",
STATE_IDLE: "STANDBY",
STATE_PLAYING: "ACTIVE",
STATE_ON: "STANDBY",
STATE_PAUSED: "STANDBY",
STATE_STANDBY: "STANDBY",
STATE_UNAVAILABLE: "INACTIVE",
STATE_UNKNOWN: "INACTIVE",
}
playback_lookup = {
STATE_OFF: "STOPPED",
STATE_IDLE: "STOPPED",
STATE_PLAYING: "PLAYING",
STATE_ON: "STOPPED",
STATE_PAUSED: "PAUSED",
STATE_STANDBY: "STOPPED",
STATE_UNAVAILABLE: "STOPPED",
STATE_UNKNOWN: "STOPPED",
}
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
return domain == media_player.DOMAIN
def sync_attributes(self) -> dict[str, Any]:
"""Return attributes for a sync request."""
return {"supportActivityState": True, "supportPlaybackState": True}
def query_attributes(self) -> dict[str, Any]:
"""Return the attributes of this trait for this entity."""
return {
"activityState": self.activity_lookup.get(self.state.state, "INACTIVE"),
"playbackState": self.playback_lookup.get(self.state.state, "STOPPED"),
}
@register_trait
class ChannelTrait(_Trait):
"""Trait to get media playback state.
https://developers.google.com/actions/smarthome/traits/channel
"""
name = TRAIT_CHANNEL
commands = [COMMAND_SELECT_CHANNEL]
@staticmethod
def supported(domain, features, device_class, _):
"""Test if state is supported."""
if (
domain == media_player.DOMAIN
and (features & MediaPlayerEntityFeature.PLAY_MEDIA)
and device_class == media_player.MediaPlayerDeviceClass.TV
):
return True
return False
def sync_attributes(self) -> dict[str, Any]:
"""Return attributes for a sync request."""
return {"availableChannels": [], "commandOnlyChannels": True}
def query_attributes(self) -> dict[str, Any]:
"""Return channel query attributes."""
return {}
async def execute(self, command, data, params, challenge):
"""Execute an setChannel command."""
if command == COMMAND_SELECT_CHANNEL:
channel_number = params.get("channelNumber")
else:
raise SmartHomeError(ERR_NOT_SUPPORTED, "Unsupported command")
if not channel_number:
raise SmartHomeError(
ERR_NO_AVAILABLE_CHANNEL,
"Channel is not available",
)
await self.hass.services.async_call(
media_player.DOMAIN,
media_player.SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: self.state.entity_id,
media_player.ATTR_MEDIA_CONTENT_ID: channel_number,
media_player.ATTR_MEDIA_CONTENT_TYPE: MediaType.CHANNEL,
},
blocking=not self.config.should_report_state,
context=data.context,
)
@register_trait
class SensorStateTrait(_Trait):
"""Trait to get sensor state.
https://developers.google.com/actions/smarthome/traits/sensorstate
"""
sensor_types = {
sensor.SensorDeviceClass.AQI: ("AirQuality", "AQI"),
sensor.SensorDeviceClass.CO: ("CarbonMonoxideLevel", "PARTS_PER_MILLION"),
sensor.SensorDeviceClass.CO2: ("CarbonDioxideLevel", "PARTS_PER_MILLION"),
sensor.SensorDeviceClass.PM25: ("PM2.5", "MICROGRAMS_PER_CUBIC_METER"),
sensor.SensorDeviceClass.PM10: ("PM10", "MICROGRAMS_PER_CUBIC_METER"),
sensor.SensorDeviceClass.VOLATILE_ORGANIC_COMPOUNDS: (
"VolatileOrganicCompounds",
"PARTS_PER_MILLION",
),
}
binary_sensor_types = {
binary_sensor.BinarySensorDeviceClass.CO: (
"CarbonMonoxideLevel",
["carbon monoxide detected", "no carbon monoxide detected", "unknown"],
),
binary_sensor.BinarySensorDeviceClass.SMOKE: (
"SmokeLevel",
["smoke detected", "no smoke detected", "unknown"],
),
binary_sensor.BinarySensorDeviceClass.MOISTURE: (
"WaterLeak",
["leak", "no leak", "unknown"],
),
}
name = TRAIT_SENSOR_STATE
commands: list[str] = []
def _air_quality_description_for_aqi(self, aqi: float | None) -> str:
if aqi is None or aqi < 0:
return "unknown"
if aqi <= 50:
return "healthy"
if aqi <= 100:
return "moderate"
if aqi <= 150:
return "unhealthy for sensitive groups"
if aqi <= 200:
return "unhealthy"
if aqi <= 300:
return "very unhealthy"
return "hazardous"
@classmethod
def supported(cls, domain, features, device_class, _):
"""Test if state is supported."""
return (domain == sensor.DOMAIN and device_class in cls.sensor_types) or (
domain == binary_sensor.DOMAIN and device_class in cls.binary_sensor_types
)
def sync_attributes(self) -> dict[str, Any]:
"""Return attributes for a sync request."""
device_class = self.state.attributes.get(ATTR_DEVICE_CLASS)
def create_sensor_state(
name: str,
raw_value_unit: str | None = None,
available_states: list[str] | None = None,
) -> dict[str, Any]:
sensor_state: dict[str, Any] = {
"name": name,
}
if raw_value_unit:
sensor_state["numericCapabilities"] = {"rawValueUnit": raw_value_unit}
if available_states:
sensor_state["descriptiveCapabilities"] = {
"availableStates": available_states
}
return {"sensorStatesSupported": [sensor_state]}
if self.state.domain == sensor.DOMAIN:
sensor_data = self.sensor_types.get(device_class)
if device_class is None or sensor_data is None:
return {}
available_states: list[str] | None = None
if device_class == sensor.SensorDeviceClass.AQI:
available_states = [
"healthy",
"moderate",
"unhealthy for sensitive groups",
"unhealthy",
"very unhealthy",
"hazardous",
"unknown",
]
return create_sensor_state(sensor_data[0], sensor_data[1], available_states)
binary_sensor_data = self.binary_sensor_types.get(device_class)
if device_class is None or binary_sensor_data is None:
return {}
return create_sensor_state(
binary_sensor_data[0], available_states=binary_sensor_data[1]
)
def query_attributes(self) -> dict[str, Any]:
"""Return the attributes of this trait for this entity."""
device_class = self.state.attributes.get(ATTR_DEVICE_CLASS)
def create_sensor_state(
name: str, raw_value: float | None = None, current_state: str | None = None
) -> dict[str, Any]:
sensor_state: dict[str, Any] = {
"name": name,
"rawValue": raw_value,
}
if current_state:
sensor_state["currentSensorState"] = current_state
return {"currentSensorStateData": [sensor_state]}
if self.state.domain == sensor.DOMAIN:
sensor_data = self.sensor_types.get(device_class)
if device_class is None or sensor_data is None:
return {}
try:
value = float(self.state.state)
except ValueError:
value = None
if self.state.state == STATE_UNKNOWN:
value = None
current_state: str | None = None
if device_class == sensor.SensorDeviceClass.AQI:
current_state = self._air_quality_description_for_aqi(value)
return create_sensor_state(sensor_data[0], value, current_state)
binary_sensor_data = self.binary_sensor_types.get(device_class)
if device_class is None or binary_sensor_data is None:
return {}
value = {
STATE_ON: 0,
STATE_OFF: 1,
STATE_UNKNOWN: 2,
}[self.state.state]
return create_sensor_state(
binary_sensor_data[0], current_state=binary_sensor_data[1][value]
)