1455 lines
48 KiB
Python
1455 lines
48 KiB
Python
"""Implement the Google Smart Home traits."""
|
|
import logging
|
|
|
|
from homeassistant.components import (
|
|
alarm_control_panel,
|
|
binary_sensor,
|
|
camera,
|
|
cover,
|
|
fan,
|
|
group,
|
|
input_boolean,
|
|
light,
|
|
lock,
|
|
media_player,
|
|
scene,
|
|
script,
|
|
sensor,
|
|
switch,
|
|
vacuum,
|
|
)
|
|
from homeassistant.components.climate import const as climate
|
|
from homeassistant.const import (
|
|
ATTR_ASSUMED_STATE,
|
|
ATTR_CODE,
|
|
ATTR_DEVICE_CLASS,
|
|
ATTR_ENTITY_ID,
|
|
ATTR_SUPPORTED_FEATURES,
|
|
ATTR_TEMPERATURE,
|
|
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_ALARM_ARMED_AWAY,
|
|
STATE_ALARM_ARMED_CUSTOM_BYPASS,
|
|
STATE_ALARM_ARMED_HOME,
|
|
STATE_ALARM_ARMED_NIGHT,
|
|
STATE_ALARM_DISARMED,
|
|
STATE_ALARM_PENDING,
|
|
STATE_ALARM_TRIGGERED,
|
|
STATE_LOCKED,
|
|
STATE_OFF,
|
|
STATE_ON,
|
|
STATE_UNAVAILABLE,
|
|
STATE_UNKNOWN,
|
|
TEMP_CELSIUS,
|
|
TEMP_FAHRENHEIT,
|
|
)
|
|
from homeassistant.core import DOMAIN as HA_DOMAIN
|
|
from homeassistant.util import color as color_util, temperature as temp_util
|
|
|
|
from .const import (
|
|
CHALLENGE_ACK_NEEDED,
|
|
CHALLENGE_FAILED_PIN_NEEDED,
|
|
CHALLENGE_PIN_NEEDED,
|
|
ERR_ALREADY_ARMED,
|
|
ERR_ALREADY_DISARMED,
|
|
ERR_CHALLENGE_NOT_SETUP,
|
|
ERR_FUNCTION_NOT_SUPPORTED,
|
|
ERR_NOT_SUPPORTED,
|
|
ERR_VALUE_OUT_OF_RANGE,
|
|
)
|
|
from .error import ChallengeNeeded, SmartHomeError
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
PREFIX_TRAITS = "action.devices.traits."
|
|
TRAIT_CAMERA_STREAM = PREFIX_TRAITS + "CameraStream"
|
|
TRAIT_ONOFF = PREFIX_TRAITS + "OnOff"
|
|
TRAIT_DOCK = PREFIX_TRAITS + "Dock"
|
|
TRAIT_STARTSTOP = PREFIX_TRAITS + "StartStop"
|
|
TRAIT_BRIGHTNESS = PREFIX_TRAITS + "Brightness"
|
|
TRAIT_COLOR_SETTING = PREFIX_TRAITS + "ColorSetting"
|
|
TRAIT_SCENE = PREFIX_TRAITS + "Scene"
|
|
TRAIT_TEMPERATURE_SETTING = PREFIX_TRAITS + "TemperatureSetting"
|
|
TRAIT_LOCKUNLOCK = PREFIX_TRAITS + "LockUnlock"
|
|
TRAIT_FANSPEED = PREFIX_TRAITS + "FanSpeed"
|
|
TRAIT_MODES = PREFIX_TRAITS + "Modes"
|
|
TRAIT_OPENCLOSE = PREFIX_TRAITS + "OpenClose"
|
|
TRAIT_VOLUME = PREFIX_TRAITS + "Volume"
|
|
TRAIT_ARMDISARM = PREFIX_TRAITS + "ArmDisarm"
|
|
TRAIT_HUMIDITY_SETTING = PREFIX_TRAITS + "HumiditySetting"
|
|
|
|
PREFIX_COMMANDS = "action.devices.commands."
|
|
COMMAND_ONOFF = PREFIX_COMMANDS + "OnOff"
|
|
COMMAND_GET_CAMERA_STREAM = PREFIX_COMMANDS + "GetCameraStream"
|
|
COMMAND_DOCK = PREFIX_COMMANDS + "Dock"
|
|
COMMAND_STARTSTOP = PREFIX_COMMANDS + "StartStop"
|
|
COMMAND_PAUSEUNPAUSE = PREFIX_COMMANDS + "PauseUnpause"
|
|
COMMAND_BRIGHTNESS_ABSOLUTE = PREFIX_COMMANDS + "BrightnessAbsolute"
|
|
COMMAND_COLOR_ABSOLUTE = PREFIX_COMMANDS + "ColorAbsolute"
|
|
COMMAND_ACTIVATE_SCENE = PREFIX_COMMANDS + "ActivateScene"
|
|
COMMAND_THERMOSTAT_TEMPERATURE_SETPOINT = (
|
|
PREFIX_COMMANDS + "ThermostatTemperatureSetpoint"
|
|
)
|
|
COMMAND_THERMOSTAT_TEMPERATURE_SET_RANGE = (
|
|
PREFIX_COMMANDS + "ThermostatTemperatureSetRange"
|
|
)
|
|
COMMAND_THERMOSTAT_SET_MODE = PREFIX_COMMANDS + "ThermostatSetMode"
|
|
COMMAND_LOCKUNLOCK = PREFIX_COMMANDS + "LockUnlock"
|
|
COMMAND_FANSPEED = PREFIX_COMMANDS + "SetFanSpeed"
|
|
COMMAND_MODES = PREFIX_COMMANDS + "SetModes"
|
|
COMMAND_OPENCLOSE = PREFIX_COMMANDS + "OpenClose"
|
|
COMMAND_SET_VOLUME = PREFIX_COMMANDS + "setVolume"
|
|
COMMAND_VOLUME_RELATIVE = PREFIX_COMMANDS + "volumeRelative"
|
|
COMMAND_ARMDISARM = PREFIX_COMMANDS + "ArmDisarm"
|
|
|
|
TRAITS = []
|
|
|
|
|
|
def register_trait(trait):
|
|
"""Decorate a function to register a trait."""
|
|
TRAITS.append(trait)
|
|
return trait
|
|
|
|
|
|
def _google_temp_unit(units):
|
|
"""Return Google temperature unit."""
|
|
if units == TEMP_FAHRENHEIT:
|
|
return "F"
|
|
return "C"
|
|
|
|
|
|
class _Trait:
|
|
"""Represents a Trait inside Google Assistant skill."""
|
|
|
|
commands = []
|
|
|
|
@staticmethod
|
|
def might_2fa(domain, features, device_class):
|
|
"""Return if the trait might ask for 2FA."""
|
|
return False
|
|
|
|
def __init__(self, hass, state, config):
|
|
"""Initialize a trait for a state."""
|
|
self.hass = hass
|
|
self.state = state
|
|
self.config = config
|
|
|
|
def sync_attributes(self):
|
|
"""Return attributes for a sync request."""
|
|
raise NotImplementedError
|
|
|
|
def query_attributes(self):
|
|
"""Return the attributes of this trait for this entity."""
|
|
raise NotImplementedError
|
|
|
|
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):
|
|
"""Test if state is supported."""
|
|
if domain == light.DOMAIN:
|
|
return features & light.SUPPORT_BRIGHTNESS
|
|
|
|
return False
|
|
|
|
def sync_attributes(self):
|
|
"""Return brightness attributes for a sync request."""
|
|
return {}
|
|
|
|
def query_attributes(self):
|
|
"""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"] = int(100 * (brightness / 255))
|
|
else:
|
|
response["brightness"] = 0
|
|
|
|
return response
|
|
|
|
async def execute(self, command, data, params, challenge):
|
|
"""Execute a brightness command."""
|
|
domain = self.state.domain
|
|
|
|
if 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=True,
|
|
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 = None
|
|
|
|
@staticmethod
|
|
def supported(domain, features, device_class):
|
|
"""Test if state is supported."""
|
|
if domain == camera.DOMAIN:
|
|
return features & camera.SUPPORT_STREAM
|
|
|
|
return False
|
|
|
|
def sync_attributes(self):
|
|
"""Return stream attributes for a sync request."""
|
|
return {
|
|
"cameraStreamSupportedProtocols": ["hls"],
|
|
"cameraStreamNeedAuthToken": False,
|
|
"cameraStreamNeedDrmEncryption": False,
|
|
}
|
|
|
|
def query_attributes(self):
|
|
"""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 self.hass.components.camera.async_request_stream(
|
|
self.state.entity_id, "hls"
|
|
)
|
|
self.stream_info = {
|
|
"cameraStreamAccessUrl": self.hass.config.api.base_url + url
|
|
}
|
|
|
|
|
|
@register_trait
|
|
class OnOffTrait(_Trait):
|
|
"""Trait to offer basic on and off functionality.
|
|
|
|
https://developers.google.com/actions/smarthome/traits/onoff
|
|
"""
|
|
|
|
name = TRAIT_ONOFF
|
|
commands = [COMMAND_ONOFF]
|
|
|
|
@staticmethod
|
|
def supported(domain, features, device_class):
|
|
"""Test if state is supported."""
|
|
return domain in (
|
|
group.DOMAIN,
|
|
input_boolean.DOMAIN,
|
|
switch.DOMAIN,
|
|
fan.DOMAIN,
|
|
light.DOMAIN,
|
|
media_player.DOMAIN,
|
|
)
|
|
|
|
def sync_attributes(self):
|
|
"""Return OnOff attributes for a sync request."""
|
|
return {}
|
|
|
|
def query_attributes(self):
|
|
"""Return OnOff query attributes."""
|
|
return {"on": self.state.state != STATE_OFF}
|
|
|
|
async def execute(self, command, data, params, challenge):
|
|
"""Execute an OnOff command."""
|
|
domain = self.state.domain
|
|
|
|
if domain == group.DOMAIN:
|
|
service_domain = HA_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=True,
|
|
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):
|
|
"""Test if state is supported."""
|
|
if domain != light.DOMAIN:
|
|
return False
|
|
|
|
return features & light.SUPPORT_COLOR_TEMP or features & light.SUPPORT_COLOR
|
|
|
|
def sync_attributes(self):
|
|
"""Return color temperature attributes for a sync request."""
|
|
attrs = self.state.attributes
|
|
features = attrs.get(ATTR_SUPPORTED_FEATURES, 0)
|
|
response = {}
|
|
|
|
if features & light.SUPPORT_COLOR:
|
|
response["colorModel"] = "hsv"
|
|
|
|
if features & light.SUPPORT_COLOR_TEMP:
|
|
# Max Kelvin is Min Mireds K = 1000000 / mireds
|
|
# Min Kelvin is Max Mireds K = 1000000 / mireds
|
|
response["colorTemperatureRange"] = {
|
|
"temperatureMaxK": color_util.color_temperature_mired_to_kelvin(
|
|
attrs.get(light.ATTR_MIN_MIREDS)
|
|
),
|
|
"temperatureMinK": color_util.color_temperature_mired_to_kelvin(
|
|
attrs.get(light.ATTR_MAX_MIREDS)
|
|
),
|
|
}
|
|
|
|
return response
|
|
|
|
def query_attributes(self):
|
|
"""Return color temperature query attributes."""
|
|
features = self.state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
|
|
color = {}
|
|
|
|
if features & light.SUPPORT_COLOR:
|
|
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 features & light.SUPPORT_COLOR_TEMP:
|
|
temp = self.state.attributes.get(light.ATTR_COLOR_TEMP)
|
|
# 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"] = color_util.color_temperature_mired_to_kelvin(
|
|
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 = color_util.color_temperature_kelvin_to_mired(
|
|
params["color"]["temperature"]
|
|
)
|
|
min_temp = self.state.attributes[light.ATTR_MIN_MIREDS]
|
|
max_temp = self.state.attributes[light.ATTR_MAX_MIREDS]
|
|
|
|
if temp < min_temp or temp > max_temp:
|
|
raise SmartHomeError(
|
|
ERR_VALUE_OUT_OF_RANGE,
|
|
"Temperature should be between {} and {}".format(
|
|
min_temp, max_temp
|
|
),
|
|
)
|
|
|
|
await self.hass.services.async_call(
|
|
light.DOMAIN,
|
|
SERVICE_TURN_ON,
|
|
{ATTR_ENTITY_ID: self.state.entity_id, light.ATTR_COLOR_TEMP: temp},
|
|
blocking=True,
|
|
context=data.context,
|
|
)
|
|
|
|
elif "spectrumRGB" in params["color"]:
|
|
# Convert integer to hex format and left pad with 0's till length 6
|
|
hex_value = "{0:06x}".format(params["color"]["spectrumRGB"])
|
|
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=True,
|
|
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=True,
|
|
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 (scene.DOMAIN, script.DOMAIN)
|
|
|
|
def sync_attributes(self):
|
|
"""Return scene attributes for a sync request."""
|
|
# Neither supported domain can support sceneReversible
|
|
return {}
|
|
|
|
def query_attributes(self):
|
|
"""Return scene query attributes."""
|
|
return {}
|
|
|
|
async def execute(self, command, data, params, challenge):
|
|
"""Execute a scene command."""
|
|
# Don't block for scripts as they can be slow.
|
|
await self.hass.services.async_call(
|
|
self.state.domain,
|
|
SERVICE_TURN_ON,
|
|
{ATTR_ENTITY_ID: self.state.entity_id},
|
|
blocking=self.state.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 == vacuum.DOMAIN
|
|
|
|
def sync_attributes(self):
|
|
"""Return dock attributes for a sync request."""
|
|
return {}
|
|
|
|
def query_attributes(self):
|
|
"""Return dock query attributes."""
|
|
return {"isDocked": self.state.state == vacuum.STATE_DOCKED}
|
|
|
|
async def execute(self, command, data, params, challenge):
|
|
"""Execute a dock command."""
|
|
await self.hass.services.async_call(
|
|
self.state.domain,
|
|
vacuum.SERVICE_RETURN_TO_BASE,
|
|
{ATTR_ENTITY_ID: self.state.entity_id},
|
|
blocking=True,
|
|
context=data.context,
|
|
)
|
|
|
|
|
|
@register_trait
|
|
class StartStopTrait(_Trait):
|
|
"""Trait to offer StartStop functionality.
|
|
|
|
https://developers.google.com/actions/smarthome/traits/startstop
|
|
"""
|
|
|
|
name = TRAIT_STARTSTOP
|
|
commands = [COMMAND_STARTSTOP, COMMAND_PAUSEUNPAUSE]
|
|
|
|
@staticmethod
|
|
def supported(domain, features, device_class):
|
|
"""Test if state is supported."""
|
|
return domain == vacuum.DOMAIN
|
|
|
|
def sync_attributes(self):
|
|
"""Return StartStop attributes for a sync request."""
|
|
return {
|
|
"pausable": self.state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
|
|
& vacuum.SUPPORT_PAUSE
|
|
!= 0
|
|
}
|
|
|
|
def query_attributes(self):
|
|
"""Return StartStop query attributes."""
|
|
return {
|
|
"isRunning": self.state.state == vacuum.STATE_CLEANING,
|
|
"isPaused": self.state.state == vacuum.STATE_PAUSED,
|
|
}
|
|
|
|
async def execute(self, command, data, params, challenge):
|
|
"""Execute a StartStop command."""
|
|
if command == COMMAND_STARTSTOP:
|
|
if params["start"]:
|
|
await self.hass.services.async_call(
|
|
self.state.domain,
|
|
vacuum.SERVICE_START,
|
|
{ATTR_ENTITY_ID: self.state.entity_id},
|
|
blocking=True,
|
|
context=data.context,
|
|
)
|
|
else:
|
|
await self.hass.services.async_call(
|
|
self.state.domain,
|
|
vacuum.SERVICE_STOP,
|
|
{ATTR_ENTITY_ID: self.state.entity_id},
|
|
blocking=True,
|
|
context=data.context,
|
|
)
|
|
elif command == COMMAND_PAUSEUNPAUSE:
|
|
if params["pause"]:
|
|
await self.hass.services.async_call(
|
|
self.state.domain,
|
|
vacuum.SERVICE_PAUSE,
|
|
{ATTR_ENTITY_ID: self.state.entity_id},
|
|
blocking=True,
|
|
context=data.context,
|
|
)
|
|
else:
|
|
await self.hass.services.async_call(
|
|
self.state.domain,
|
|
vacuum.SERVICE_START,
|
|
{ATTR_ENTITY_ID: self.state.entity_id},
|
|
blocking=True,
|
|
context=data.context,
|
|
)
|
|
|
|
|
|
@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.HVAC_MODE_HEAT: "heat",
|
|
climate.HVAC_MODE_COOL: "cool",
|
|
climate.HVAC_MODE_OFF: "off",
|
|
climate.HVAC_MODE_AUTO: "auto",
|
|
climate.HVAC_MODE_HEAT_COOL: "heatcool",
|
|
climate.HVAC_MODE_FAN_ONLY: "fan-only",
|
|
climate.HVAC_MODE_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."""
|
|
if domain == climate.DOMAIN:
|
|
return True
|
|
|
|
return (
|
|
domain == sensor.DOMAIN and device_class == sensor.DEVICE_CLASS_TEMPERATURE
|
|
)
|
|
|
|
@property
|
|
def climate_google_modes(self):
|
|
"""Return supported Google modes."""
|
|
modes = []
|
|
attrs = self.state.attributes
|
|
|
|
for mode in attrs.get(climate.ATTR_HVAC_MODES, []):
|
|
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, []):
|
|
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):
|
|
"""Return temperature point and modes attributes for a sync request."""
|
|
response = {}
|
|
attrs = self.state.attributes
|
|
domain = self.state.domain
|
|
response["thermostatTemperatureUnit"] = _google_temp_unit(
|
|
self.hass.config.units.temperature_unit
|
|
)
|
|
|
|
if domain == sensor.DOMAIN:
|
|
device_class = attrs.get(ATTR_DEVICE_CLASS)
|
|
if device_class == sensor.DEVICE_CLASS_TEMPERATURE:
|
|
response["queryOnlyTemperatureSetting"] = True
|
|
|
|
elif domain == climate.DOMAIN:
|
|
modes = self.climate_google_modes
|
|
if "off" in modes and any(
|
|
mode in modes for mode in ("heatcool", "heat", "cool")
|
|
):
|
|
modes.append("on")
|
|
response["availableThermostatModes"] = ",".join(modes)
|
|
|
|
return response
|
|
|
|
def query_attributes(self):
|
|
"""Return temperature point and modes query attributes."""
|
|
response = {}
|
|
attrs = self.state.attributes
|
|
domain = self.state.domain
|
|
unit = self.hass.config.units.temperature_unit
|
|
if domain == sensor.DOMAIN:
|
|
device_class = attrs.get(ATTR_DEVICE_CLASS)
|
|
if device_class == sensor.DEVICE_CLASS_TEMPERATURE:
|
|
current_temp = self.state.state
|
|
if current_temp not in (STATE_UNKNOWN, STATE_UNAVAILABLE):
|
|
response["thermostatTemperatureAmbient"] = round(
|
|
temp_util.convert(float(current_temp), unit, TEMP_CELSIUS), 1
|
|
)
|
|
|
|
elif domain == climate.DOMAIN:
|
|
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)
|
|
|
|
current_temp = attrs.get(climate.ATTR_CURRENT_TEMPERATURE)
|
|
if current_temp is not None:
|
|
response["thermostatTemperatureAmbient"] = round(
|
|
temp_util.convert(current_temp, unit, TEMP_CELSIUS), 1
|
|
)
|
|
|
|
current_humidity = attrs.get(climate.ATTR_CURRENT_HUMIDITY)
|
|
if current_humidity is not None:
|
|
response["thermostatHumidityAmbient"] = current_humidity
|
|
|
|
if operation in (climate.HVAC_MODE_AUTO, climate.HVAC_MODE_HEAT_COOL):
|
|
if supported & climate.SUPPORT_TARGET_TEMPERATURE_RANGE:
|
|
response["thermostatTemperatureSetpointHigh"] = round(
|
|
temp_util.convert(
|
|
attrs[climate.ATTR_TARGET_TEMP_HIGH], unit, TEMP_CELSIUS
|
|
),
|
|
1,
|
|
)
|
|
response["thermostatTemperatureSetpointLow"] = round(
|
|
temp_util.convert(
|
|
attrs[climate.ATTR_TARGET_TEMP_LOW], unit, TEMP_CELSIUS
|
|
),
|
|
1,
|
|
)
|
|
else:
|
|
target_temp = attrs.get(ATTR_TEMPERATURE)
|
|
if target_temp is not None:
|
|
target_temp = round(
|
|
temp_util.convert(target_temp, unit, TEMP_CELSIUS), 1
|
|
)
|
|
response["thermostatTemperatureSetpointHigh"] = target_temp
|
|
response["thermostatTemperatureSetpointLow"] = target_temp
|
|
else:
|
|
target_temp = attrs.get(ATTR_TEMPERATURE)
|
|
if target_temp is not None:
|
|
response["thermostatTemperatureSetpoint"] = round(
|
|
temp_util.convert(target_temp, unit, TEMP_CELSIUS), 1
|
|
)
|
|
|
|
return response
|
|
|
|
async def execute(self, command, data, params, challenge):
|
|
"""Execute a temperature point or mode command."""
|
|
domain = self.state.domain
|
|
if domain == sensor.DOMAIN:
|
|
raise SmartHomeError(
|
|
ERR_NOT_SUPPORTED, "Execute is not supported by sensor"
|
|
)
|
|
|
|
# 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 = temp_util.convert(
|
|
params["thermostatTemperatureSetpoint"], TEMP_CELSIUS, unit
|
|
)
|
|
if unit == TEMP_FAHRENHEIT:
|
|
temp = round(temp)
|
|
|
|
if temp < min_temp or temp > max_temp:
|
|
raise SmartHomeError(
|
|
ERR_VALUE_OUT_OF_RANGE,
|
|
"Temperature should be between {} and {}".format(
|
|
min_temp, 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=True,
|
|
context=data.context,
|
|
)
|
|
|
|
elif command == COMMAND_THERMOSTAT_TEMPERATURE_SET_RANGE:
|
|
temp_high = temp_util.convert(
|
|
params["thermostatTemperatureSetpointHigh"], TEMP_CELSIUS, unit
|
|
)
|
|
if unit == TEMP_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 "
|
|
"{} and {}".format(min_temp, max_temp),
|
|
)
|
|
|
|
temp_low = temp_util.convert(
|
|
params["thermostatTemperatureSetpointLow"], TEMP_CELSIUS, unit
|
|
)
|
|
if unit == TEMP_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 "
|
|
"{} and {}".format(min_temp, max_temp),
|
|
)
|
|
|
|
supported = self.state.attributes.get(ATTR_SUPPORTED_FEATURES)
|
|
svc_data = {ATTR_ENTITY_ID: self.state.entity_id}
|
|
|
|
if supported & climate.SUPPORT_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=True,
|
|
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=True,
|
|
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=True,
|
|
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=True,
|
|
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=True,
|
|
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 = []
|
|
|
|
@staticmethod
|
|
def supported(domain, features, device_class):
|
|
"""Test if state is supported."""
|
|
return domain == sensor.DOMAIN and device_class == sensor.DEVICE_CLASS_HUMIDITY
|
|
|
|
def sync_attributes(self):
|
|
"""Return humidity attributes for a sync request."""
|
|
response = {}
|
|
attrs = self.state.attributes
|
|
domain = self.state.domain
|
|
if domain == sensor.DOMAIN:
|
|
device_class = attrs.get(ATTR_DEVICE_CLASS)
|
|
if device_class == sensor.DEVICE_CLASS_HUMIDITY:
|
|
response["queryOnlyHumiditySetting"] = True
|
|
|
|
return response
|
|
|
|
def query_attributes(self):
|
|
"""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.DEVICE_CLASS_HUMIDITY:
|
|
current_humidity = self.state.state
|
|
if current_humidity not in (STATE_UNKNOWN, STATE_UNAVAILABLE):
|
|
response["humidityAmbientPercent"] = round(float(current_humidity))
|
|
|
|
return response
|
|
|
|
async def execute(self, command, data, params, challenge):
|
|
"""Execute a humidity command."""
|
|
domain = self.state.domain
|
|
if domain == sensor.DOMAIN:
|
|
raise SmartHomeError(
|
|
ERR_NOT_SUPPORTED, "Execute is not supported by sensor"
|
|
)
|
|
|
|
|
|
@register_trait
|
|
class LockUnlockTrait(_Trait):
|
|
"""Trait to lock or unlock a lock.
|
|
|
|
https://developers.google.com/actions/smarthome/traits/lockunlock
|
|
"""
|
|
|
|
name = TRAIT_LOCKUNLOCK
|
|
commands = [COMMAND_LOCKUNLOCK]
|
|
|
|
@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):
|
|
"""Return LockUnlock attributes for a sync request."""
|
|
return {}
|
|
|
|
def query_attributes(self):
|
|
"""Return LockUnlock query attributes."""
|
|
return {"isLocked": self.state.state == STATE_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=True,
|
|
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_ARMDISARM
|
|
commands = [COMMAND_ARMDISARM]
|
|
|
|
state_to_service = {
|
|
STATE_ALARM_ARMED_HOME: SERVICE_ALARM_ARM_HOME,
|
|
STATE_ALARM_ARMED_AWAY: SERVICE_ALARM_ARM_AWAY,
|
|
STATE_ALARM_ARMED_NIGHT: SERVICE_ALARM_ARM_NIGHT,
|
|
STATE_ALARM_ARMED_CUSTOM_BYPASS: SERVICE_ALARM_ARM_CUSTOM_BYPASS,
|
|
STATE_ALARM_TRIGGERED: SERVICE_ALARM_TRIGGER,
|
|
}
|
|
|
|
@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 sync_attributes(self):
|
|
"""Return ArmDisarm attributes for a sync request."""
|
|
response = {}
|
|
levels = []
|
|
for state in self.state_to_service:
|
|
# level synonyms are generated from state names
|
|
# 'armed_away' becomes 'armed away' or 'away'
|
|
level_synonym = [state.replace("_", " ")]
|
|
if state != STATE_ALARM_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": False}
|
|
return response
|
|
|
|
def query_attributes(self):
|
|
"""Return ArmDisarm query attributes."""
|
|
if "post_pending_state" in self.state.attributes:
|
|
armed_state = self.state.attributes["post_pending_state"]
|
|
else:
|
|
armed_state = self.state.state
|
|
response = {"isArmed": armed_state in self.state_to_service}
|
|
if response["isArmed"]:
|
|
response.update({"currentArmLevel": armed_state})
|
|
return response
|
|
|
|
async def execute(self, command, data, params, challenge):
|
|
"""Execute an ArmDisarm command."""
|
|
if params["arm"] and not params.get("cancel"):
|
|
if self.state.state == params["armLevel"]:
|
|
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[params["armLevel"]]
|
|
# 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 == STATE_ALARM_PENDING
|
|
):
|
|
service = SERVICE_ALARM_DISARM
|
|
else:
|
|
if self.state.state == STATE_ALARM_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=True,
|
|
context=data.context,
|
|
)
|
|
|
|
|
|
@register_trait
|
|
class FanSpeedTrait(_Trait):
|
|
"""Trait to control speed of Fan.
|
|
|
|
https://developers.google.com/actions/smarthome/traits/fanspeed
|
|
"""
|
|
|
|
name = TRAIT_FANSPEED
|
|
commands = [COMMAND_FANSPEED]
|
|
|
|
speed_synonyms = {
|
|
fan.SPEED_OFF: ["stop", "off"],
|
|
fan.SPEED_LOW: ["slow", "low", "slowest", "lowest"],
|
|
fan.SPEED_MEDIUM: ["medium", "mid", "middle"],
|
|
fan.SPEED_HIGH: ["high", "max", "fast", "highest", "fastest", "maximum"],
|
|
}
|
|
|
|
@staticmethod
|
|
def supported(domain, features, device_class):
|
|
"""Test if state is supported."""
|
|
if domain != fan.DOMAIN:
|
|
return False
|
|
|
|
return features & fan.SUPPORT_SET_SPEED
|
|
|
|
def sync_attributes(self):
|
|
"""Return speed point and modes attributes for a sync request."""
|
|
modes = self.state.attributes.get(fan.ATTR_SPEED_LIST, [])
|
|
speeds = []
|
|
for mode in modes:
|
|
if mode not in self.speed_synonyms:
|
|
continue
|
|
speed = {
|
|
"speed_name": mode,
|
|
"speed_values": [
|
|
{"speed_synonym": self.speed_synonyms.get(mode), "lang": "en"}
|
|
],
|
|
}
|
|
speeds.append(speed)
|
|
|
|
return {
|
|
"availableFanSpeeds": {"speeds": speeds, "ordered": True},
|
|
"reversible": bool(
|
|
self.state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
|
|
& fan.SUPPORT_DIRECTION
|
|
),
|
|
}
|
|
|
|
def query_attributes(self):
|
|
"""Return speed point and modes query attributes."""
|
|
attrs = self.state.attributes
|
|
response = {}
|
|
|
|
speed = attrs.get(fan.ATTR_SPEED)
|
|
if speed is not None:
|
|
response["on"] = speed != fan.SPEED_OFF
|
|
response["online"] = True
|
|
response["currentFanSpeedSetting"] = speed
|
|
|
|
return response
|
|
|
|
async def execute(self, command, data, params, challenge):
|
|
"""Execute an SetFanSpeed command."""
|
|
await self.hass.services.async_call(
|
|
fan.DOMAIN,
|
|
fan.SERVICE_SET_SPEED,
|
|
{ATTR_ENTITY_ID: self.state.entity_id, fan.ATTR_SPEED: params["fanSpeed"]},
|
|
blocking=True,
|
|
context=data.context,
|
|
)
|
|
|
|
|
|
@register_trait
|
|
class ModesTrait(_Trait):
|
|
"""Trait to set modes.
|
|
|
|
https://developers.google.com/actions/smarthome/traits/modes
|
|
"""
|
|
|
|
name = TRAIT_MODES
|
|
commands = [COMMAND_MODES]
|
|
|
|
SYNONYMS = {
|
|
"input source": ["input source", "input", "source"],
|
|
"sound mode": ["sound mode", "effects"],
|
|
}
|
|
|
|
@staticmethod
|
|
def supported(domain, features, device_class):
|
|
"""Test if state is supported."""
|
|
if domain != media_player.DOMAIN:
|
|
return False
|
|
|
|
return (
|
|
features & media_player.SUPPORT_SELECT_SOURCE
|
|
or features & media_player.SUPPORT_SELECT_SOUND_MODE
|
|
)
|
|
|
|
def sync_attributes(self):
|
|
"""Return mode attributes for a sync request."""
|
|
|
|
def _generate(name, settings):
|
|
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
|
|
|
|
attrs = self.state.attributes
|
|
modes = []
|
|
if media_player.ATTR_INPUT_SOURCE_LIST in attrs:
|
|
modes.append(
|
|
_generate("input source", attrs[media_player.ATTR_INPUT_SOURCE_LIST])
|
|
)
|
|
|
|
if media_player.ATTR_SOUND_MODE_LIST in attrs:
|
|
modes.append(
|
|
_generate("sound mode", attrs[media_player.ATTR_SOUND_MODE_LIST])
|
|
)
|
|
|
|
payload = {"availableModes": modes}
|
|
|
|
return payload
|
|
|
|
def query_attributes(self):
|
|
"""Return current modes."""
|
|
attrs = self.state.attributes
|
|
response = {}
|
|
mode_settings = {}
|
|
|
|
if media_player.ATTR_INPUT_SOURCE_LIST in attrs:
|
|
mode_settings["input source"] = attrs.get(media_player.ATTR_INPUT_SOURCE)
|
|
|
|
if media_player.ATTR_SOUND_MODE_LIST in attrs:
|
|
mode_settings["sound mode"] = attrs.get(media_player.ATTR_SOUND_MODE)
|
|
|
|
if mode_settings:
|
|
response["on"] = self.state.state != STATE_OFF
|
|
response["online"] = True
|
|
response["currentModeSettings"] = mode_settings
|
|
|
|
return response
|
|
|
|
async def execute(self, command, data, params, challenge):
|
|
"""Execute an SetModes command."""
|
|
settings = params.get("updateModeSettings")
|
|
requested_source = settings.get("input source")
|
|
sound_mode = settings.get("sound mode")
|
|
|
|
if requested_source:
|
|
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=True,
|
|
context=data.context,
|
|
)
|
|
|
|
if 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=True,
|
|
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.DEVICE_CLASS_DOOR, cover.DEVICE_CLASS_GARAGE)
|
|
|
|
name = TRAIT_OPENCLOSE
|
|
commands = [COMMAND_OPENCLOSE]
|
|
|
|
override_position = None
|
|
|
|
@staticmethod
|
|
def supported(domain, features, device_class):
|
|
"""Test if state is supported."""
|
|
if domain == cover.DOMAIN:
|
|
return True
|
|
|
|
return domain == binary_sensor.DOMAIN and device_class in (
|
|
binary_sensor.DEVICE_CLASS_DOOR,
|
|
binary_sensor.DEVICE_CLASS_GARAGE_DOOR,
|
|
binary_sensor.DEVICE_CLASS_LOCK,
|
|
binary_sensor.DEVICE_CLASS_OPENING,
|
|
binary_sensor.DEVICE_CLASS_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):
|
|
"""Return opening direction."""
|
|
response = {}
|
|
if self.state.domain == binary_sensor.DOMAIN:
|
|
response["queryOnlyOpenClose"] = True
|
|
return response
|
|
|
|
def query_attributes(self):
|
|
"""Return state query attributes."""
|
|
domain = self.state.domain
|
|
response = {}
|
|
|
|
if self.override_position is not None:
|
|
response["openPercent"] = self.override_position
|
|
|
|
elif domain == cover.DOMAIN:
|
|
# When it's an assumed state, we will return that querying state
|
|
# is not supported.
|
|
if self.state.attributes.get(ATTR_ASSUMED_STATE):
|
|
raise SmartHomeError(
|
|
ERR_NOT_SUPPORTED, "Querying state is not supported"
|
|
)
|
|
|
|
if self.state.state == STATE_UNKNOWN:
|
|
raise SmartHomeError(
|
|
ERR_NOT_SUPPORTED, "Querying state is not supported"
|
|
)
|
|
|
|
position = self.override_position or self.state.attributes.get(
|
|
cover.ATTR_CURRENT_POSITION
|
|
)
|
|
|
|
if position is not None:
|
|
response["openPercent"] = position
|
|
elif self.state.state != cover.STATE_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
|
|
|
|
if domain == cover.DOMAIN:
|
|
svc_params = {ATTR_ENTITY_ID: self.state.entity_id}
|
|
|
|
if params["openPercent"] == 0:
|
|
service = cover.SERVICE_CLOSE_COVER
|
|
should_verify = False
|
|
elif params["openPercent"] == 100:
|
|
service = cover.SERVICE_OPEN_COVER
|
|
should_verify = True
|
|
elif (
|
|
self.state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
|
|
& cover.SUPPORT_SET_POSITION
|
|
):
|
|
service = cover.SERVICE_SET_COVER_POSITION
|
|
should_verify = True
|
|
svc_params[cover.ATTR_POSITION] = params["openPercent"]
|
|
else:
|
|
raise SmartHomeError(
|
|
ERR_FUNCTION_NOT_SUPPORTED, "Setting a position is not supported"
|
|
)
|
|
|
|
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(
|
|
cover.DOMAIN, service, svc_params, blocking=True, context=data.context
|
|
)
|
|
|
|
if (
|
|
self.state.attributes.get(ATTR_ASSUMED_STATE)
|
|
or self.state.state == STATE_UNKNOWN
|
|
):
|
|
self.override_position = params["openPercent"]
|
|
|
|
|
|
@register_trait
|
|
class VolumeTrait(_Trait):
|
|
"""Trait to control brightness of a device.
|
|
|
|
https://developers.google.com/actions/smarthome/traits/volume
|
|
"""
|
|
|
|
name = TRAIT_VOLUME
|
|
commands = [COMMAND_SET_VOLUME, COMMAND_VOLUME_RELATIVE]
|
|
|
|
@staticmethod
|
|
def supported(domain, features, device_class):
|
|
"""Test if state is supported."""
|
|
if domain == media_player.DOMAIN:
|
|
return features & media_player.SUPPORT_VOLUME_SET
|
|
|
|
return False
|
|
|
|
def sync_attributes(self):
|
|
"""Return brightness attributes for a sync request."""
|
|
return {}
|
|
|
|
def query_attributes(self):
|
|
"""Return brightness query attributes."""
|
|
response = {}
|
|
|
|
level = self.state.attributes.get(media_player.ATTR_MEDIA_VOLUME_LEVEL)
|
|
muted = self.state.attributes.get(media_player.ATTR_MEDIA_VOLUME_MUTED)
|
|
if level is not None:
|
|
# Convert 0.0-1.0 to 0-100
|
|
response["currentVolume"] = int(level * 100)
|
|
response["isMuted"] = bool(muted)
|
|
|
|
return response
|
|
|
|
async def _execute_set_volume(self, data, params):
|
|
level = params["volumeLevel"]
|
|
|
|
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 / 100,
|
|
},
|
|
blocking=True,
|
|
context=data.context,
|
|
)
|
|
|
|
async def _execute_volume_relative(self, data, params):
|
|
# This could also support up/down commands using relativeSteps
|
|
relative = params["volumeRelativeLevel"]
|
|
current = self.state.attributes.get(media_player.ATTR_MEDIA_VOLUME_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: current + relative / 100,
|
|
},
|
|
blocking=True,
|
|
context=data.context,
|
|
)
|
|
|
|
async def execute(self, command, data, params, challenge):
|
|
"""Execute a brightness 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)
|
|
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)
|
|
|
|
pin = challenge.get("pin")
|
|
|
|
if pin != data.config.secure_devices_pin:
|
|
raise ChallengeNeeded(CHALLENGE_FAILED_PIN_NEEDED)
|
|
|
|
|
|
def _verify_ack_challenge(data, state, challenge):
|
|
"""Verify an ack challenge."""
|
|
if not data.config.should_2fa(state):
|
|
return
|
|
if not challenge or not challenge.get("ack"):
|
|
raise ChallengeNeeded(CHALLENGE_ACK_NEEDED)
|