mirror of
https://github.com/home-assistant/core.git
synced 2025-07-19 19:27:45 +00:00
Dyson removal (#59401)
Co-authored-by: epenet <epenet@users.noreply.github.com>
This commit is contained in:
parent
a102c425a9
commit
d226df2511
@ -1,153 +0,0 @@
|
||||
"""Support for Dyson Pure Cool Link devices."""
|
||||
import logging
|
||||
|
||||
from libpurecool.dyson import DysonAccount
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.const import CONF_DEVICES, CONF_PASSWORD, CONF_TIMEOUT, CONF_USERNAME
|
||||
from homeassistant.helpers import discovery
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity import Entity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
CONF_LANGUAGE = "language"
|
||||
CONF_RETRY = "retry"
|
||||
|
||||
DEFAULT_TIMEOUT = 5
|
||||
DEFAULT_RETRY = 10
|
||||
DYSON_DEVICES = "dyson_devices"
|
||||
PLATFORMS = ["sensor", "fan", "vacuum", "climate", "air_quality"]
|
||||
|
||||
DOMAIN = "dyson"
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema(
|
||||
{
|
||||
DOMAIN: vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_USERNAME): cv.string,
|
||||
vol.Required(CONF_PASSWORD): cv.string,
|
||||
vol.Required(CONF_LANGUAGE): cv.string,
|
||||
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
|
||||
vol.Optional(CONF_RETRY, default=DEFAULT_RETRY): cv.positive_int,
|
||||
vol.Optional(CONF_DEVICES, default=[]): vol.All(cv.ensure_list, [dict]),
|
||||
}
|
||||
)
|
||||
},
|
||||
extra=vol.ALLOW_EXTRA,
|
||||
)
|
||||
|
||||
|
||||
def setup(hass, config):
|
||||
"""Set up the Dyson parent component."""
|
||||
_LOGGER.info("Creating new Dyson component")
|
||||
|
||||
if DYSON_DEVICES not in hass.data:
|
||||
hass.data[DYSON_DEVICES] = []
|
||||
|
||||
dyson_account = DysonAccount(
|
||||
config[DOMAIN].get(CONF_USERNAME),
|
||||
config[DOMAIN].get(CONF_PASSWORD),
|
||||
config[DOMAIN].get(CONF_LANGUAGE),
|
||||
)
|
||||
|
||||
logged = dyson_account.login()
|
||||
|
||||
timeout = config[DOMAIN].get(CONF_TIMEOUT)
|
||||
retry = config[DOMAIN].get(CONF_RETRY)
|
||||
|
||||
if not logged:
|
||||
_LOGGER.error("Not connected to Dyson account. Unable to add devices")
|
||||
return False
|
||||
|
||||
_LOGGER.info("Connected to Dyson account")
|
||||
dyson_devices = dyson_account.devices()
|
||||
if CONF_DEVICES in config[DOMAIN] and config[DOMAIN].get(CONF_DEVICES):
|
||||
configured_devices = config[DOMAIN].get(CONF_DEVICES)
|
||||
for device in configured_devices:
|
||||
dyson_device = next(
|
||||
(d for d in dyson_devices if d.serial == device["device_id"]), None
|
||||
)
|
||||
if dyson_device:
|
||||
try:
|
||||
connected = dyson_device.connect(device["device_ip"])
|
||||
if connected:
|
||||
_LOGGER.info("Connected to device %s", dyson_device)
|
||||
hass.data[DYSON_DEVICES].append(dyson_device)
|
||||
else:
|
||||
_LOGGER.warning("Unable to connect to device %s", dyson_device)
|
||||
except OSError as ose:
|
||||
_LOGGER.error(
|
||||
"Unable to connect to device %s: %s",
|
||||
str(dyson_device.network_device),
|
||||
str(ose),
|
||||
)
|
||||
else:
|
||||
_LOGGER.warning(
|
||||
"Unable to find device %s in Dyson account", device["device_id"]
|
||||
)
|
||||
else:
|
||||
# Not yet reliable
|
||||
for device in dyson_devices:
|
||||
_LOGGER.info(
|
||||
"Trying to connect to device %s with timeout=%i and retry=%i",
|
||||
device,
|
||||
timeout,
|
||||
retry,
|
||||
)
|
||||
connected = device.auto_connect(timeout, retry)
|
||||
if connected:
|
||||
_LOGGER.info("Connected to device %s", device)
|
||||
hass.data[DYSON_DEVICES].append(device)
|
||||
else:
|
||||
_LOGGER.warning("Unable to connect to device %s", device)
|
||||
|
||||
# Start fan/sensors components
|
||||
if hass.data[DYSON_DEVICES]:
|
||||
_LOGGER.debug("Starting sensor/fan components")
|
||||
for platform in PLATFORMS:
|
||||
discovery.load_platform(hass, platform, DOMAIN, {}, config)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
class DysonEntity(Entity):
|
||||
"""Representation of a Dyson entity."""
|
||||
|
||||
def __init__(self, device, state_type):
|
||||
"""Initialize the entity."""
|
||||
self._device = device
|
||||
self._state_type = state_type
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
"""Call when entity is added to hass."""
|
||||
self._device.add_message_listener(self.on_message_filter)
|
||||
|
||||
def on_message_filter(self, message):
|
||||
"""Filter new messages received."""
|
||||
if self._state_type is None or isinstance(message, self._state_type):
|
||||
_LOGGER.debug(
|
||||
"Message received for device %s : %s",
|
||||
self.name,
|
||||
message,
|
||||
)
|
||||
self.on_message(message)
|
||||
|
||||
def on_message(self, message):
|
||||
"""Handle new messages received."""
|
||||
self.schedule_update_ha_state()
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
"""No polling needed."""
|
||||
return False
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the Dyson sensor."""
|
||||
return self._device.name
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""Return the sensor's unique id."""
|
||||
return self._device.serial
|
@ -1,93 +0,0 @@
|
||||
"""Support for Dyson Pure Cool Air Quality Sensors."""
|
||||
from libpurecool.dyson_pure_cool import DysonPureCool
|
||||
from libpurecool.dyson_pure_state_v2 import DysonEnvironmentalSensorV2State
|
||||
|
||||
from homeassistant.components.air_quality import AirQualityEntity
|
||||
|
||||
from . import DYSON_DEVICES, DysonEntity
|
||||
|
||||
ATTRIBUTION = "Dyson purifier air quality sensor"
|
||||
|
||||
DYSON_AIQ_DEVICES = "dyson_aiq_devices"
|
||||
|
||||
ATTR_VOC = "volatile_organic_compounds"
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
"""Set up the Dyson Sensors."""
|
||||
|
||||
if discovery_info is None:
|
||||
return
|
||||
|
||||
hass.data.setdefault(DYSON_AIQ_DEVICES, [])
|
||||
|
||||
# Get Dyson Devices from parent component
|
||||
device_ids = [device.unique_id for device in hass.data[DYSON_AIQ_DEVICES]]
|
||||
new_entities = []
|
||||
for device in hass.data[DYSON_DEVICES]:
|
||||
if isinstance(device, DysonPureCool) and device.serial not in device_ids:
|
||||
new_entities.append(DysonAirSensor(device))
|
||||
|
||||
if not new_entities:
|
||||
return
|
||||
|
||||
hass.data[DYSON_AIQ_DEVICES].extend(new_entities)
|
||||
add_entities(hass.data[DYSON_AIQ_DEVICES])
|
||||
|
||||
|
||||
class DysonAirSensor(DysonEntity, AirQualityEntity):
|
||||
"""Representation of a generic Dyson air quality sensor."""
|
||||
|
||||
def __init__(self, device):
|
||||
"""Create a new generic air quality Dyson sensor."""
|
||||
super().__init__(device, DysonEnvironmentalSensorV2State)
|
||||
self._old_value = None
|
||||
|
||||
def on_message(self, message):
|
||||
"""Handle new messages which are received from the fan."""
|
||||
if (
|
||||
self._old_value is None
|
||||
or self._old_value != self._device.environmental_state
|
||||
):
|
||||
self._old_value = self._device.environmental_state
|
||||
self.schedule_update_ha_state()
|
||||
|
||||
@property
|
||||
def attribution(self):
|
||||
"""Return the attribution."""
|
||||
return ATTRIBUTION
|
||||
|
||||
@property
|
||||
def air_quality_index(self):
|
||||
"""Return the Air Quality Index (AQI)."""
|
||||
return max(
|
||||
self.particulate_matter_2_5,
|
||||
self.particulate_matter_10,
|
||||
self.nitrogen_dioxide,
|
||||
self.volatile_organic_compounds,
|
||||
)
|
||||
|
||||
@property
|
||||
def particulate_matter_2_5(self):
|
||||
"""Return the particulate matter 2.5 level."""
|
||||
return int(self._device.environmental_state.particulate_matter_25)
|
||||
|
||||
@property
|
||||
def particulate_matter_10(self):
|
||||
"""Return the particulate matter 10 level."""
|
||||
return int(self._device.environmental_state.particulate_matter_10)
|
||||
|
||||
@property
|
||||
def nitrogen_dioxide(self):
|
||||
"""Return the NO2 (nitrogen dioxide) level."""
|
||||
return int(self._device.environmental_state.nitrogen_dioxide)
|
||||
|
||||
@property
|
||||
def volatile_organic_compounds(self):
|
||||
"""Return the VOC (Volatile Organic Compounds) level."""
|
||||
return int(self._device.environmental_state.volatile_organic_compounds)
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self):
|
||||
"""Return the device state attributes."""
|
||||
return {ATTR_VOC: self.volatile_organic_compounds}
|
@ -1,316 +0,0 @@
|
||||
"""Support for Dyson Pure Hot+Cool link fan."""
|
||||
import logging
|
||||
|
||||
from libpurecool.const import (
|
||||
AutoMode,
|
||||
FanPower,
|
||||
FanSpeed,
|
||||
FanState,
|
||||
FocusMode,
|
||||
HeatMode,
|
||||
HeatState,
|
||||
HeatTarget,
|
||||
)
|
||||
from libpurecool.dyson_pure_hotcool import DysonPureHotCool
|
||||
from libpurecool.dyson_pure_hotcool_link import DysonPureHotCoolLink
|
||||
from libpurecool.dyson_pure_state import DysonPureHotCoolState
|
||||
from libpurecool.dyson_pure_state_v2 import DysonPureHotCoolV2State
|
||||
|
||||
from homeassistant.components.climate import ClimateEntity
|
||||
from homeassistant.components.climate.const import (
|
||||
CURRENT_HVAC_COOL,
|
||||
CURRENT_HVAC_HEAT,
|
||||
CURRENT_HVAC_IDLE,
|
||||
CURRENT_HVAC_OFF,
|
||||
FAN_AUTO,
|
||||
FAN_DIFFUSE,
|
||||
FAN_FOCUS,
|
||||
FAN_HIGH,
|
||||
FAN_LOW,
|
||||
FAN_MEDIUM,
|
||||
FAN_OFF,
|
||||
HVAC_MODE_COOL,
|
||||
HVAC_MODE_HEAT,
|
||||
HVAC_MODE_OFF,
|
||||
SUPPORT_FAN_MODE,
|
||||
SUPPORT_TARGET_TEMPERATURE,
|
||||
)
|
||||
from homeassistant.const import ATTR_TEMPERATURE, TEMP_CELSIUS
|
||||
|
||||
from . import DYSON_DEVICES, DysonEntity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
SUPPORT_FAN = [FAN_FOCUS, FAN_DIFFUSE]
|
||||
SUPPORT_FAN_PCOOL = [FAN_OFF, FAN_AUTO, FAN_LOW, FAN_MEDIUM, FAN_HIGH]
|
||||
SUPPORT_HVAC = [HVAC_MODE_COOL, HVAC_MODE_HEAT]
|
||||
SUPPORT_HVAC_PCOOL = [HVAC_MODE_COOL, HVAC_MODE_HEAT, HVAC_MODE_OFF]
|
||||
SUPPORT_FLAGS = SUPPORT_TARGET_TEMPERATURE | SUPPORT_FAN_MODE
|
||||
|
||||
DYSON_KNOWN_CLIMATE_DEVICES = "dyson_known_climate_devices"
|
||||
|
||||
SPEED_MAP = {
|
||||
FanSpeed.FAN_SPEED_1.value: FAN_LOW,
|
||||
FanSpeed.FAN_SPEED_2.value: FAN_LOW,
|
||||
FanSpeed.FAN_SPEED_3.value: FAN_LOW,
|
||||
FanSpeed.FAN_SPEED_4.value: FAN_LOW,
|
||||
FanSpeed.FAN_SPEED_AUTO.value: FAN_AUTO,
|
||||
FanSpeed.FAN_SPEED_5.value: FAN_MEDIUM,
|
||||
FanSpeed.FAN_SPEED_6.value: FAN_MEDIUM,
|
||||
FanSpeed.FAN_SPEED_7.value: FAN_MEDIUM,
|
||||
FanSpeed.FAN_SPEED_8.value: FAN_HIGH,
|
||||
FanSpeed.FAN_SPEED_9.value: FAN_HIGH,
|
||||
FanSpeed.FAN_SPEED_10.value: FAN_HIGH,
|
||||
}
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
"""Set up the Dyson fan components."""
|
||||
if discovery_info is None:
|
||||
return
|
||||
|
||||
known_devices = hass.data.setdefault(DYSON_KNOWN_CLIMATE_DEVICES, set())
|
||||
|
||||
# Get Dyson Devices from parent component
|
||||
new_entities = []
|
||||
|
||||
for device in hass.data[DYSON_DEVICES]:
|
||||
if device.serial not in known_devices:
|
||||
if isinstance(device, DysonPureHotCool):
|
||||
dyson_entity = DysonPureHotCoolEntity(device)
|
||||
new_entities.append(dyson_entity)
|
||||
known_devices.add(device.serial)
|
||||
elif isinstance(device, DysonPureHotCoolLink):
|
||||
dyson_entity = DysonPureHotCoolLinkEntity(device)
|
||||
new_entities.append(dyson_entity)
|
||||
known_devices.add(device.serial)
|
||||
|
||||
add_entities(new_entities)
|
||||
|
||||
|
||||
class DysonClimateEntity(DysonEntity, ClimateEntity):
|
||||
"""Representation of a Dyson climate fan."""
|
||||
|
||||
@property
|
||||
def supported_features(self):
|
||||
"""Return the list of supported features."""
|
||||
return SUPPORT_FLAGS
|
||||
|
||||
@property
|
||||
def temperature_unit(self):
|
||||
"""Return the unit of measurement."""
|
||||
return TEMP_CELSIUS
|
||||
|
||||
@property
|
||||
def current_temperature(self):
|
||||
"""Return the current temperature."""
|
||||
if (
|
||||
self._device.environmental_state
|
||||
and self._device.environmental_state.temperature
|
||||
):
|
||||
temperature_kelvin = self._device.environmental_state.temperature
|
||||
return float(f"{temperature_kelvin - 273:.1f}")
|
||||
return None
|
||||
|
||||
@property
|
||||
def target_temperature(self):
|
||||
"""Return the target temperature."""
|
||||
heat_target = int(self._device.state.heat_target) / 10
|
||||
return int(heat_target - 273)
|
||||
|
||||
@property
|
||||
def current_humidity(self):
|
||||
"""Return the current humidity."""
|
||||
# Humidity equaling to 0 means invalid value so we don't check for None here
|
||||
# https://github.com/home-assistant/core/pull/45172#discussion_r559069756
|
||||
if (
|
||||
self._device.environmental_state
|
||||
and self._device.environmental_state.humidity
|
||||
):
|
||||
return self._device.environmental_state.humidity
|
||||
return None
|
||||
|
||||
@property
|
||||
def min_temp(self):
|
||||
"""Return the minimum temperature."""
|
||||
return 1
|
||||
|
||||
@property
|
||||
def max_temp(self):
|
||||
"""Return the maximum temperature."""
|
||||
return 37
|
||||
|
||||
def set_temperature(self, **kwargs):
|
||||
"""Set new target temperature."""
|
||||
if (target_temp := kwargs.get(ATTR_TEMPERATURE)) is None:
|
||||
_LOGGER.error("Missing target temperature %s", kwargs)
|
||||
return
|
||||
target_temp = int(target_temp)
|
||||
_LOGGER.debug("Set %s temperature %s", self.name, target_temp)
|
||||
# Limit the target temperature into acceptable range.
|
||||
target_temp = min(self.max_temp, target_temp)
|
||||
target_temp = max(self.min_temp, target_temp)
|
||||
self.set_heat_target(HeatTarget.celsius(target_temp))
|
||||
|
||||
def set_heat_target(self, heat_target):
|
||||
"""Set heating target temperature."""
|
||||
|
||||
|
||||
class DysonPureHotCoolLinkEntity(DysonClimateEntity):
|
||||
"""Representation of a Dyson climate fan."""
|
||||
|
||||
def __init__(self, device):
|
||||
"""Initialize the fan."""
|
||||
super().__init__(device, DysonPureHotCoolState)
|
||||
|
||||
@property
|
||||
def hvac_mode(self):
|
||||
"""Return hvac operation ie. heat, cool mode.
|
||||
|
||||
Need to be one of HVAC_MODE_*.
|
||||
"""
|
||||
if self._device.state.heat_mode == HeatMode.HEAT_ON.value:
|
||||
return HVAC_MODE_HEAT
|
||||
return HVAC_MODE_COOL
|
||||
|
||||
@property
|
||||
def hvac_modes(self):
|
||||
"""Return the list of available hvac operation modes.
|
||||
|
||||
Need to be a subset of HVAC_MODES.
|
||||
"""
|
||||
return SUPPORT_HVAC
|
||||
|
||||
@property
|
||||
def hvac_action(self):
|
||||
"""Return the current running hvac operation if supported.
|
||||
|
||||
Need to be one of CURRENT_HVAC_*.
|
||||
"""
|
||||
if self._device.state.heat_mode == HeatMode.HEAT_ON.value:
|
||||
if self._device.state.heat_state == HeatState.HEAT_STATE_ON.value:
|
||||
return CURRENT_HVAC_HEAT
|
||||
return CURRENT_HVAC_IDLE
|
||||
return CURRENT_HVAC_COOL
|
||||
|
||||
@property
|
||||
def fan_mode(self):
|
||||
"""Return the fan setting."""
|
||||
if self._device.state.focus_mode == FocusMode.FOCUS_ON.value:
|
||||
return FAN_FOCUS
|
||||
return FAN_DIFFUSE
|
||||
|
||||
@property
|
||||
def fan_modes(self):
|
||||
"""Return the list of available fan modes."""
|
||||
return SUPPORT_FAN
|
||||
|
||||
def set_heat_target(self, heat_target):
|
||||
"""Set heating target temperature."""
|
||||
self._device.set_configuration(
|
||||
heat_target=heat_target, heat_mode=HeatMode.HEAT_ON
|
||||
)
|
||||
|
||||
def set_fan_mode(self, fan_mode):
|
||||
"""Set new fan mode."""
|
||||
_LOGGER.debug("Set %s focus mode %s", self.name, fan_mode)
|
||||
if fan_mode == FAN_FOCUS:
|
||||
self._device.set_configuration(focus_mode=FocusMode.FOCUS_ON)
|
||||
elif fan_mode == FAN_DIFFUSE:
|
||||
self._device.set_configuration(focus_mode=FocusMode.FOCUS_OFF)
|
||||
|
||||
def set_hvac_mode(self, hvac_mode):
|
||||
"""Set new target hvac mode."""
|
||||
_LOGGER.debug("Set %s heat mode %s", self.name, hvac_mode)
|
||||
if hvac_mode == HVAC_MODE_HEAT:
|
||||
self._device.set_configuration(heat_mode=HeatMode.HEAT_ON)
|
||||
elif hvac_mode == HVAC_MODE_COOL:
|
||||
self._device.set_configuration(heat_mode=HeatMode.HEAT_OFF)
|
||||
|
||||
|
||||
class DysonPureHotCoolEntity(DysonClimateEntity):
|
||||
"""Representation of a Dyson climate hot+cool fan."""
|
||||
|
||||
def __init__(self, device):
|
||||
"""Initialize the fan."""
|
||||
super().__init__(device, DysonPureHotCoolV2State)
|
||||
|
||||
@property
|
||||
def hvac_mode(self):
|
||||
"""Return hvac operation ie. heat, cool mode.
|
||||
|
||||
Need to be one of HVAC_MODE_*.
|
||||
"""
|
||||
if self._device.state.fan_power == FanPower.POWER_OFF.value:
|
||||
return HVAC_MODE_OFF
|
||||
if self._device.state.heat_mode == HeatMode.HEAT_ON.value:
|
||||
return HVAC_MODE_HEAT
|
||||
return HVAC_MODE_COOL
|
||||
|
||||
@property
|
||||
def hvac_modes(self):
|
||||
"""Return the list of available hvac operation modes.
|
||||
|
||||
Need to be a subset of HVAC_MODES.
|
||||
"""
|
||||
return SUPPORT_HVAC_PCOOL
|
||||
|
||||
@property
|
||||
def hvac_action(self):
|
||||
"""Return the current running hvac operation if supported.
|
||||
|
||||
Need to be one of CURRENT_HVAC_*.
|
||||
"""
|
||||
if self._device.state.fan_power == FanPower.POWER_OFF.value:
|
||||
return CURRENT_HVAC_OFF
|
||||
if self._device.state.heat_mode == HeatMode.HEAT_ON.value:
|
||||
if self._device.state.heat_state == HeatState.HEAT_STATE_ON.value:
|
||||
return CURRENT_HVAC_HEAT
|
||||
return CURRENT_HVAC_IDLE
|
||||
return CURRENT_HVAC_COOL
|
||||
|
||||
@property
|
||||
def fan_mode(self):
|
||||
"""Return the fan setting."""
|
||||
if (
|
||||
self._device.state.auto_mode != AutoMode.AUTO_ON.value
|
||||
and self._device.state.fan_state == FanState.FAN_OFF.value
|
||||
):
|
||||
return FAN_OFF
|
||||
|
||||
return SPEED_MAP[self._device.state.speed]
|
||||
|
||||
@property
|
||||
def fan_modes(self):
|
||||
"""Return the list of available fan modes."""
|
||||
return SUPPORT_FAN_PCOOL
|
||||
|
||||
def set_heat_target(self, heat_target):
|
||||
"""Set heating target temperature."""
|
||||
self._device.set_heat_target(heat_target)
|
||||
|
||||
def set_fan_mode(self, fan_mode):
|
||||
"""Set new fan mode."""
|
||||
_LOGGER.debug("Set %s focus mode %s", self.name, fan_mode)
|
||||
if fan_mode == FAN_OFF:
|
||||
self._device.turn_off()
|
||||
elif fan_mode == FAN_LOW:
|
||||
self._device.set_fan_speed(FanSpeed.FAN_SPEED_4)
|
||||
elif fan_mode == FAN_MEDIUM:
|
||||
self._device.set_fan_speed(FanSpeed.FAN_SPEED_7)
|
||||
elif fan_mode == FAN_HIGH:
|
||||
self._device.set_fan_speed(FanSpeed.FAN_SPEED_10)
|
||||
elif fan_mode == FAN_AUTO:
|
||||
self._device.enable_auto_mode()
|
||||
|
||||
def set_hvac_mode(self, hvac_mode):
|
||||
"""Set new target hvac mode."""
|
||||
_LOGGER.debug("Set %s heat mode %s", self.name, hvac_mode)
|
||||
if hvac_mode == HVAC_MODE_OFF:
|
||||
self._device.turn_off()
|
||||
elif self._device.state.fan_power == FanPower.POWER_OFF.value:
|
||||
self._device.turn_on()
|
||||
if hvac_mode == HVAC_MODE_HEAT:
|
||||
self._device.enable_heat_mode()
|
||||
elif hvac_mode == HVAC_MODE_COOL:
|
||||
self._device.disable_heat_mode()
|
@ -1,469 +0,0 @@
|
||||
"""Support for Dyson Pure Cool link fan."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import math
|
||||
|
||||
from libpurecool.const import FanMode, FanSpeed, NightMode, Oscillation
|
||||
from libpurecool.dyson_pure_cool import DysonPureCool
|
||||
from libpurecool.dyson_pure_cool_link import DysonPureCoolLink
|
||||
from libpurecool.dyson_pure_state import DysonPureCoolState
|
||||
from libpurecool.dyson_pure_state_v2 import DysonPureCoolV2State
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.fan import SUPPORT_OSCILLATE, SUPPORT_SET_SPEED, FanEntity
|
||||
from homeassistant.helpers import config_validation as cv, entity_platform
|
||||
from homeassistant.util.percentage import (
|
||||
int_states_in_range,
|
||||
percentage_to_ranged_value,
|
||||
ranged_value_to_percentage,
|
||||
)
|
||||
|
||||
from . import DYSON_DEVICES, DysonEntity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ATTR_NIGHT_MODE = "night_mode"
|
||||
ATTR_AUTO_MODE = "auto_mode"
|
||||
ATTR_ANGLE_LOW = "angle_low"
|
||||
ATTR_ANGLE_HIGH = "angle_high"
|
||||
ATTR_FLOW_DIRECTION_FRONT = "flow_direction_front"
|
||||
ATTR_TIMER = "timer"
|
||||
ATTR_HEPA_FILTER = "hepa_filter"
|
||||
ATTR_CARBON_FILTER = "carbon_filter"
|
||||
ATTR_DYSON_SPEED = "dyson_speed"
|
||||
ATTR_DYSON_SPEED_LIST = "dyson_speed_list"
|
||||
|
||||
DYSON_DOMAIN = "dyson"
|
||||
DYSON_FAN_DEVICES = "dyson_fan_devices"
|
||||
|
||||
SERVICE_SET_NIGHT_MODE = "set_night_mode"
|
||||
SERVICE_SET_AUTO_MODE = "set_auto_mode"
|
||||
SERVICE_SET_ANGLE = "set_angle"
|
||||
SERVICE_SET_FLOW_DIRECTION_FRONT = "set_flow_direction_front"
|
||||
SERVICE_SET_TIMER = "set_timer"
|
||||
SERVICE_SET_DYSON_SPEED = "set_speed"
|
||||
|
||||
SET_NIGHT_MODE_SCHEMA = {
|
||||
vol.Required(ATTR_NIGHT_MODE): cv.boolean,
|
||||
}
|
||||
|
||||
SET_AUTO_MODE_SCHEMA = {
|
||||
vol.Required(ATTR_AUTO_MODE): cv.boolean,
|
||||
}
|
||||
|
||||
SET_ANGLE_SCHEMA = {
|
||||
vol.Required(ATTR_ANGLE_LOW): cv.positive_int,
|
||||
vol.Required(ATTR_ANGLE_HIGH): cv.positive_int,
|
||||
}
|
||||
|
||||
SET_FLOW_DIRECTION_FRONT_SCHEMA = {
|
||||
vol.Required(ATTR_FLOW_DIRECTION_FRONT): cv.boolean,
|
||||
}
|
||||
|
||||
SET_TIMER_SCHEMA = {
|
||||
vol.Required(ATTR_TIMER): cv.positive_int,
|
||||
}
|
||||
|
||||
SET_DYSON_SPEED_SCHEMA = {
|
||||
vol.Required(ATTR_DYSON_SPEED): cv.positive_int,
|
||||
}
|
||||
|
||||
|
||||
PRESET_MODE_AUTO = "auto"
|
||||
PRESET_MODES = [PRESET_MODE_AUTO]
|
||||
|
||||
ORDERED_DYSON_SPEEDS = [
|
||||
FanSpeed.FAN_SPEED_1,
|
||||
FanSpeed.FAN_SPEED_2,
|
||||
FanSpeed.FAN_SPEED_3,
|
||||
FanSpeed.FAN_SPEED_4,
|
||||
FanSpeed.FAN_SPEED_5,
|
||||
FanSpeed.FAN_SPEED_6,
|
||||
FanSpeed.FAN_SPEED_7,
|
||||
FanSpeed.FAN_SPEED_8,
|
||||
FanSpeed.FAN_SPEED_9,
|
||||
FanSpeed.FAN_SPEED_10,
|
||||
]
|
||||
DYSON_SPEED_TO_INT_VALUE = {k: int(k.value) for k in ORDERED_DYSON_SPEEDS}
|
||||
INT_VALUE_TO_DYSON_SPEED = {v: k for k, v in DYSON_SPEED_TO_INT_VALUE.items()}
|
||||
|
||||
SPEED_LIST_DYSON = list(DYSON_SPEED_TO_INT_VALUE.values())
|
||||
|
||||
SPEED_RANGE = (
|
||||
SPEED_LIST_DYSON[0],
|
||||
SPEED_LIST_DYSON[-1],
|
||||
) # off is not included
|
||||
|
||||
|
||||
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
|
||||
"""Set up the Dyson fan components."""
|
||||
|
||||
if discovery_info is None:
|
||||
return
|
||||
|
||||
_LOGGER.debug("Creating new Dyson fans")
|
||||
if DYSON_FAN_DEVICES not in hass.data:
|
||||
hass.data[DYSON_FAN_DEVICES] = []
|
||||
|
||||
# Get Dyson Devices from parent component
|
||||
has_purecool_devices = False
|
||||
device_serials = [device.serial for device in hass.data[DYSON_FAN_DEVICES]]
|
||||
for device in hass.data[DYSON_DEVICES]:
|
||||
if device.serial not in device_serials:
|
||||
if isinstance(device, DysonPureCool):
|
||||
has_purecool_devices = True
|
||||
dyson_entity = DysonPureCoolEntity(device)
|
||||
hass.data[DYSON_FAN_DEVICES].append(dyson_entity)
|
||||
elif isinstance(device, DysonPureCoolLink):
|
||||
dyson_entity = DysonPureCoolLinkEntity(device)
|
||||
hass.data[DYSON_FAN_DEVICES].append(dyson_entity)
|
||||
|
||||
async_add_entities(hass.data[DYSON_FAN_DEVICES])
|
||||
|
||||
# Register custom services
|
||||
platform = entity_platform.async_get_current_platform()
|
||||
platform.async_register_entity_service(
|
||||
SERVICE_SET_NIGHT_MODE, SET_NIGHT_MODE_SCHEMA, "set_night_mode"
|
||||
)
|
||||
platform.async_register_entity_service(
|
||||
SERVICE_SET_AUTO_MODE, SET_AUTO_MODE_SCHEMA, "set_auto_mode"
|
||||
)
|
||||
platform.async_register_entity_service(
|
||||
SERVICE_SET_DYSON_SPEED, SET_DYSON_SPEED_SCHEMA, "service_set_dyson_speed"
|
||||
)
|
||||
if has_purecool_devices:
|
||||
platform.async_register_entity_service(
|
||||
SERVICE_SET_ANGLE, SET_ANGLE_SCHEMA, "set_angle"
|
||||
)
|
||||
platform.async_register_entity_service(
|
||||
SERVICE_SET_FLOW_DIRECTION_FRONT,
|
||||
SET_FLOW_DIRECTION_FRONT_SCHEMA,
|
||||
"set_flow_direction_front",
|
||||
)
|
||||
platform.async_register_entity_service(
|
||||
SERVICE_SET_TIMER, SET_TIMER_SCHEMA, "set_timer"
|
||||
)
|
||||
|
||||
|
||||
class DysonFanEntity(DysonEntity, FanEntity):
|
||||
"""Representation of a Dyson fan."""
|
||||
|
||||
@property
|
||||
def percentage(self):
|
||||
"""Return the current speed percentage."""
|
||||
if self.auto_mode:
|
||||
return None
|
||||
return ranged_value_to_percentage(SPEED_RANGE, int(self._device.state.speed))
|
||||
|
||||
@property
|
||||
def speed_count(self) -> int:
|
||||
"""Return the number of speeds the fan supports."""
|
||||
return int_states_in_range(SPEED_RANGE)
|
||||
|
||||
@property
|
||||
def preset_modes(self):
|
||||
"""Return the available preset modes."""
|
||||
return PRESET_MODES
|
||||
|
||||
@property
|
||||
def preset_mode(self):
|
||||
"""Return the current preset mode."""
|
||||
if self.auto_mode:
|
||||
return PRESET_MODE_AUTO
|
||||
return None
|
||||
|
||||
@property
|
||||
def dyson_speed(self):
|
||||
"""Return the current speed."""
|
||||
if self._device.state.speed == FanSpeed.FAN_SPEED_AUTO.value:
|
||||
return self._device.state.speed
|
||||
return int(self._device.state.speed)
|
||||
|
||||
@property
|
||||
def dyson_speed_list(self) -> list:
|
||||
"""Get the list of available dyson speeds."""
|
||||
return SPEED_LIST_DYSON
|
||||
|
||||
@property
|
||||
def night_mode(self):
|
||||
"""Return Night mode."""
|
||||
return self._device.state.night_mode == "ON"
|
||||
|
||||
@property
|
||||
def auto_mode(self):
|
||||
"""Return auto mode."""
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
def supported_features(self) -> int:
|
||||
"""Flag supported features."""
|
||||
return SUPPORT_OSCILLATE | SUPPORT_SET_SPEED
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> dict:
|
||||
"""Return optional state attributes."""
|
||||
return {
|
||||
ATTR_NIGHT_MODE: self.night_mode,
|
||||
ATTR_AUTO_MODE: self.auto_mode,
|
||||
ATTR_DYSON_SPEED: self.dyson_speed,
|
||||
ATTR_DYSON_SPEED_LIST: self.dyson_speed_list,
|
||||
}
|
||||
|
||||
def set_auto_mode(self, auto_mode: bool) -> None:
|
||||
"""Set auto mode."""
|
||||
raise NotImplementedError
|
||||
|
||||
def set_percentage(self, percentage: int) -> None:
|
||||
"""Set the speed percentage of the fan."""
|
||||
if percentage == 0:
|
||||
self.turn_off()
|
||||
return
|
||||
dyson_speed = INT_VALUE_TO_DYSON_SPEED[
|
||||
math.ceil(percentage_to_ranged_value(SPEED_RANGE, percentage))
|
||||
]
|
||||
self.set_dyson_speed(dyson_speed)
|
||||
|
||||
def set_preset_mode(self, preset_mode: str) -> None:
|
||||
"""Set a preset mode on the fan."""
|
||||
self._valid_preset_mode_or_raise(preset_mode)
|
||||
# There currently is only one
|
||||
self.set_auto_mode(True)
|
||||
|
||||
def set_dyson_speed(self, speed: FanSpeed) -> None:
|
||||
"""Set the exact speed of the fan."""
|
||||
raise NotImplementedError
|
||||
|
||||
def service_set_dyson_speed(self, dyson_speed: int) -> None:
|
||||
"""Handle the service to set dyson speed."""
|
||||
if dyson_speed not in SPEED_LIST_DYSON:
|
||||
raise ValueError(f'"{dyson_speed}" is not a valid Dyson speed')
|
||||
_LOGGER.debug("Set exact speed to %s", dyson_speed)
|
||||
speed = FanSpeed(f"{int(dyson_speed):04d}")
|
||||
self.set_dyson_speed(speed)
|
||||
|
||||
def turn_on(
|
||||
self,
|
||||
speed: str | None = None,
|
||||
percentage: int | None = None,
|
||||
preset_mode: str | None = None,
|
||||
**kwargs,
|
||||
) -> None:
|
||||
"""Turn on the fan."""
|
||||
_LOGGER.debug("Turn on fan %s with percentage %s", self.name, percentage)
|
||||
if preset_mode:
|
||||
self.set_preset_mode(preset_mode)
|
||||
elif percentage is None:
|
||||
# percentage not set, just turn on
|
||||
self._device.set_configuration(fan_mode=FanMode.FAN)
|
||||
else:
|
||||
self.set_percentage(percentage)
|
||||
|
||||
|
||||
class DysonPureCoolLinkEntity(DysonFanEntity):
|
||||
"""Representation of a Dyson fan."""
|
||||
|
||||
def __init__(self, device):
|
||||
"""Initialize the fan."""
|
||||
super().__init__(device, DysonPureCoolState)
|
||||
|
||||
def turn_off(self, **kwargs) -> None:
|
||||
"""Turn off the fan."""
|
||||
_LOGGER.debug("Turn off fan %s", self.name)
|
||||
self._device.set_configuration(fan_mode=FanMode.OFF)
|
||||
|
||||
def set_dyson_speed(self, speed: FanSpeed) -> None:
|
||||
"""Set the exact speed of the fan."""
|
||||
self._device.set_configuration(fan_mode=FanMode.FAN, fan_speed=speed)
|
||||
|
||||
def oscillate(self, oscillating: bool) -> None:
|
||||
"""Turn on/off oscillating."""
|
||||
_LOGGER.debug("Turn oscillation %s for device %s", oscillating, self.name)
|
||||
|
||||
if oscillating:
|
||||
self._device.set_configuration(oscillation=Oscillation.OSCILLATION_ON)
|
||||
else:
|
||||
self._device.set_configuration(oscillation=Oscillation.OSCILLATION_OFF)
|
||||
|
||||
@property
|
||||
def oscillating(self):
|
||||
"""Return the oscillation state."""
|
||||
return self._device.state.oscillation == "ON"
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Return true if the entity is on."""
|
||||
return self._device.state.fan_mode in ["FAN", "AUTO"]
|
||||
|
||||
def set_night_mode(self, night_mode: bool) -> None:
|
||||
"""Turn fan in night mode."""
|
||||
_LOGGER.debug("Set %s night mode %s", self.name, night_mode)
|
||||
if night_mode:
|
||||
self._device.set_configuration(night_mode=NightMode.NIGHT_MODE_ON)
|
||||
else:
|
||||
self._device.set_configuration(night_mode=NightMode.NIGHT_MODE_OFF)
|
||||
|
||||
@property
|
||||
def auto_mode(self):
|
||||
"""Return auto mode."""
|
||||
return self._device.state.fan_mode == "AUTO"
|
||||
|
||||
def set_auto_mode(self, auto_mode: bool) -> None:
|
||||
"""Turn fan in auto mode."""
|
||||
_LOGGER.debug("Set %s auto mode %s", self.name, auto_mode)
|
||||
if auto_mode:
|
||||
self._device.set_configuration(fan_mode=FanMode.AUTO)
|
||||
else:
|
||||
self._device.set_configuration(fan_mode=FanMode.FAN)
|
||||
|
||||
|
||||
class DysonPureCoolEntity(DysonFanEntity):
|
||||
"""Representation of a Dyson Purecool (TP04/DP04) fan."""
|
||||
|
||||
def __init__(self, device):
|
||||
"""Initialize the fan."""
|
||||
super().__init__(device, DysonPureCoolV2State)
|
||||
|
||||
def turn_on(
|
||||
self,
|
||||
speed: str | None = None,
|
||||
percentage: int | None = None,
|
||||
preset_mode: str | None = None,
|
||||
**kwargs,
|
||||
) -> None:
|
||||
"""Turn on the fan."""
|
||||
_LOGGER.debug("Turn on fan %s with percentage %s", self.name, percentage)
|
||||
if preset_mode:
|
||||
self.set_preset_mode(preset_mode)
|
||||
elif percentage is None:
|
||||
# percentage not set, just turn on
|
||||
self._device.turn_on()
|
||||
else:
|
||||
self.set_percentage(percentage)
|
||||
|
||||
def turn_off(self, **kwargs):
|
||||
"""Turn off the fan."""
|
||||
_LOGGER.debug("Turn off fan %s", self.name)
|
||||
self._device.turn_off()
|
||||
|
||||
def set_dyson_speed(self, speed: FanSpeed) -> None:
|
||||
"""Set the exact speed of the purecool fan."""
|
||||
self._device.set_fan_speed(speed)
|
||||
|
||||
def oscillate(self, oscillating: bool) -> None:
|
||||
"""Turn on/off oscillating."""
|
||||
_LOGGER.debug("Turn oscillation %s for device %s", oscillating, self.name)
|
||||
|
||||
if oscillating:
|
||||
self._device.enable_oscillation()
|
||||
else:
|
||||
self._device.disable_oscillation()
|
||||
|
||||
def set_night_mode(self, night_mode: bool) -> None:
|
||||
"""Turn on/off night mode."""
|
||||
_LOGGER.debug("Turn night mode %s for device %s", night_mode, self.name)
|
||||
|
||||
if night_mode:
|
||||
self._device.enable_night_mode()
|
||||
else:
|
||||
self._device.disable_night_mode()
|
||||
|
||||
def set_auto_mode(self, auto_mode: bool) -> None:
|
||||
"""Turn auto mode on/off."""
|
||||
_LOGGER.debug("Turn auto mode %s for device %s", auto_mode, self.name)
|
||||
if auto_mode:
|
||||
self._device.enable_auto_mode()
|
||||
else:
|
||||
self._device.disable_auto_mode()
|
||||
|
||||
def set_angle(self, angle_low: int, angle_high: int) -> None:
|
||||
"""Set device angle."""
|
||||
_LOGGER.debug(
|
||||
"set low %s and high angle %s for device %s",
|
||||
angle_low,
|
||||
angle_high,
|
||||
self.name,
|
||||
)
|
||||
self._device.enable_oscillation(angle_low, angle_high)
|
||||
|
||||
def set_flow_direction_front(self, flow_direction_front: bool) -> None:
|
||||
"""Set frontal airflow direction."""
|
||||
_LOGGER.debug(
|
||||
"Set frontal flow direction to %s for device %s",
|
||||
flow_direction_front,
|
||||
self.name,
|
||||
)
|
||||
|
||||
if flow_direction_front:
|
||||
self._device.enable_frontal_direction()
|
||||
else:
|
||||
self._device.disable_frontal_direction()
|
||||
|
||||
def set_timer(self, timer) -> None:
|
||||
"""Set timer."""
|
||||
_LOGGER.debug("Set timer to %s for device %s", timer, self.name)
|
||||
|
||||
if timer == 0:
|
||||
self._device.disable_sleep_timer()
|
||||
else:
|
||||
self._device.enable_sleep_timer(timer)
|
||||
|
||||
@property
|
||||
def oscillating(self):
|
||||
"""Return the oscillation state."""
|
||||
return self._device.state and self._device.state.oscillation == "OION"
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Return true if the entity is on."""
|
||||
return self._device.state.fan_power == "ON"
|
||||
|
||||
@property
|
||||
def auto_mode(self):
|
||||
"""Return Auto mode."""
|
||||
return self._device.state.auto_mode == "ON"
|
||||
|
||||
@property
|
||||
def angle_low(self):
|
||||
"""Return angle high."""
|
||||
return int(self._device.state.oscillation_angle_low)
|
||||
|
||||
@property
|
||||
def angle_high(self):
|
||||
"""Return angle low."""
|
||||
return int(self._device.state.oscillation_angle_high)
|
||||
|
||||
@property
|
||||
def flow_direction_front(self):
|
||||
"""Return frontal flow direction."""
|
||||
return self._device.state.front_direction == "ON"
|
||||
|
||||
@property
|
||||
def timer(self):
|
||||
"""Return timer."""
|
||||
return self._device.state.sleep_timer
|
||||
|
||||
@property
|
||||
def hepa_filter(self):
|
||||
"""Return the HEPA filter state."""
|
||||
return int(self._device.state.hepa_filter_state)
|
||||
|
||||
@property
|
||||
def carbon_filter(self):
|
||||
"""Return the carbon filter state."""
|
||||
if self._device.state.carbon_filter_state == "INV":
|
||||
return self._device.state.carbon_filter_state
|
||||
return int(self._device.state.carbon_filter_state)
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> dict:
|
||||
"""Return optional state attributes."""
|
||||
return {
|
||||
**super().extra_state_attributes,
|
||||
ATTR_ANGLE_LOW: self.angle_low,
|
||||
ATTR_ANGLE_HIGH: self.angle_high,
|
||||
ATTR_FLOW_DIRECTION_FRONT: self.flow_direction_front,
|
||||
ATTR_TIMER: self.timer,
|
||||
ATTR_HEPA_FILTER: self.hepa_filter,
|
||||
ATTR_CARBON_FILTER: self.carbon_filter,
|
||||
}
|
@ -1,9 +0,0 @@
|
||||
{
|
||||
"domain": "dyson",
|
||||
"name": "Dyson",
|
||||
"documentation": "https://www.home-assistant.io/integrations/dyson",
|
||||
"requirements": ["libpurecool==0.6.4"],
|
||||
"after_dependencies": ["zeroconf"],
|
||||
"codeowners": [],
|
||||
"iot_class": "local_push"
|
||||
}
|
@ -1,248 +0,0 @@
|
||||
"""Support for Dyson Pure Cool Link Sensors."""
|
||||
from libpurecool.dyson_pure_cool import DysonPureCool
|
||||
from libpurecool.dyson_pure_cool_link import DysonPureCoolLink
|
||||
|
||||
from homeassistant.components.sensor import SensorEntity
|
||||
from homeassistant.const import (
|
||||
ATTR_DEVICE_CLASS,
|
||||
ATTR_ICON,
|
||||
ATTR_UNIT_OF_MEASUREMENT,
|
||||
DEVICE_CLASS_HUMIDITY,
|
||||
DEVICE_CLASS_TEMPERATURE,
|
||||
PERCENTAGE,
|
||||
STATE_OFF,
|
||||
TEMP_CELSIUS,
|
||||
TIME_HOURS,
|
||||
)
|
||||
|
||||
from . import DYSON_DEVICES, DysonEntity
|
||||
|
||||
SENSOR_ATTRIBUTES = {
|
||||
"air_quality": {ATTR_ICON: "mdi:fan"},
|
||||
"dust": {ATTR_ICON: "mdi:cloud"},
|
||||
"humidity": {
|
||||
ATTR_DEVICE_CLASS: DEVICE_CLASS_HUMIDITY,
|
||||
ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE,
|
||||
},
|
||||
"temperature": {ATTR_DEVICE_CLASS: DEVICE_CLASS_TEMPERATURE},
|
||||
"filter_life": {
|
||||
ATTR_ICON: "mdi:filter-outline",
|
||||
ATTR_UNIT_OF_MEASUREMENT: TIME_HOURS,
|
||||
},
|
||||
"carbon_filter_state": {
|
||||
ATTR_ICON: "mdi:filter-outline",
|
||||
ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE,
|
||||
},
|
||||
"combi_filter_state": {
|
||||
ATTR_ICON: "mdi:filter-outline",
|
||||
ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE,
|
||||
},
|
||||
"hepa_filter_state": {
|
||||
ATTR_ICON: "mdi:filter-outline",
|
||||
ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE,
|
||||
},
|
||||
}
|
||||
|
||||
SENSOR_NAMES = {
|
||||
"air_quality": "AQI",
|
||||
"dust": "Dust",
|
||||
"humidity": "Humidity",
|
||||
"temperature": "Temperature",
|
||||
"filter_life": "Filter Life",
|
||||
"carbon_filter_state": "Carbon Filter Remaining Life",
|
||||
"combi_filter_state": "Combi Filter Remaining Life",
|
||||
"hepa_filter_state": "HEPA Filter Remaining Life",
|
||||
}
|
||||
|
||||
DYSON_SENSOR_DEVICES = "dyson_sensor_devices"
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
"""Set up the Dyson Sensors."""
|
||||
|
||||
if discovery_info is None:
|
||||
return
|
||||
|
||||
hass.data.setdefault(DYSON_SENSOR_DEVICES, [])
|
||||
unit = hass.config.units.temperature_unit
|
||||
devices = hass.data[DYSON_SENSOR_DEVICES]
|
||||
|
||||
# Get Dyson Devices from parent component
|
||||
device_ids = [device.unique_id for device in hass.data[DYSON_SENSOR_DEVICES]]
|
||||
new_entities = []
|
||||
for device in hass.data[DYSON_DEVICES]:
|
||||
if isinstance(device, DysonPureCool):
|
||||
if f"{device.serial}-temperature" not in device_ids:
|
||||
new_entities.append(DysonTemperatureSensor(device, unit))
|
||||
if f"{device.serial}-humidity" not in device_ids:
|
||||
new_entities.append(DysonHumiditySensor(device))
|
||||
|
||||
# For PureCool+Humidify devices, a single filter exists, called "Combi Filter".
|
||||
# It's reported with the HEPA state, while the Carbon state is set to INValid.
|
||||
if device.state and device.state.carbon_filter_state == "INV":
|
||||
if f"{device.serial}-hepa_filter_state" not in device_ids:
|
||||
new_entities.append(DysonHepaFilterLifeSensor(device, "combi"))
|
||||
else:
|
||||
if f"{device.serial}-hepa_filter_state" not in device_ids:
|
||||
new_entities.append(DysonHepaFilterLifeSensor(device))
|
||||
if f"{device.serial}-carbon_filter_state" not in device_ids:
|
||||
new_entities.append(DysonCarbonFilterLifeSensor(device))
|
||||
elif isinstance(device, DysonPureCoolLink):
|
||||
new_entities.append(DysonFilterLifeSensor(device))
|
||||
new_entities.append(DysonDustSensor(device))
|
||||
new_entities.append(DysonHumiditySensor(device))
|
||||
new_entities.append(DysonTemperatureSensor(device, unit))
|
||||
new_entities.append(DysonAirQualitySensor(device))
|
||||
|
||||
if not new_entities:
|
||||
return
|
||||
|
||||
devices.extend(new_entities)
|
||||
add_entities(devices)
|
||||
|
||||
|
||||
class DysonSensor(DysonEntity, SensorEntity):
|
||||
"""Representation of a generic Dyson sensor."""
|
||||
|
||||
def __init__(self, device, sensor_type):
|
||||
"""Create a new generic Dyson sensor."""
|
||||
super().__init__(device, None)
|
||||
self._old_value = None
|
||||
self._sensor_type = sensor_type
|
||||
self._attributes = SENSOR_ATTRIBUTES[sensor_type]
|
||||
|
||||
def on_message(self, message):
|
||||
"""Handle new messages which are received from the fan."""
|
||||
# Prevent refreshing if not needed
|
||||
if self._old_value is None or self._old_value != self.state:
|
||||
self._old_value = self.state
|
||||
self.schedule_update_ha_state()
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the Dyson sensor name."""
|
||||
return f"{super().name} {SENSOR_NAMES[self._sensor_type]}"
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""Return the sensor's unique id."""
|
||||
return f"{self._device.serial}-{self._sensor_type}"
|
||||
|
||||
@property
|
||||
def native_unit_of_measurement(self):
|
||||
"""Return the unit the value is expressed in."""
|
||||
return self._attributes.get(ATTR_UNIT_OF_MEASUREMENT)
|
||||
|
||||
@property
|
||||
def icon(self):
|
||||
"""Return the icon for this sensor."""
|
||||
return self._attributes.get(ATTR_ICON)
|
||||
|
||||
@property
|
||||
def device_class(self):
|
||||
"""Return the device class of this sensor."""
|
||||
return self._attributes.get(ATTR_DEVICE_CLASS)
|
||||
|
||||
|
||||
class DysonFilterLifeSensor(DysonSensor):
|
||||
"""Representation of Dyson Filter Life sensor (in hours)."""
|
||||
|
||||
def __init__(self, device):
|
||||
"""Create a new Dyson Filter Life sensor."""
|
||||
super().__init__(device, "filter_life")
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
"""Return filter life in hours."""
|
||||
return int(self._device.state.filter_life)
|
||||
|
||||
|
||||
class DysonCarbonFilterLifeSensor(DysonSensor):
|
||||
"""Representation of Dyson Carbon Filter Life sensor (in percent)."""
|
||||
|
||||
def __init__(self, device):
|
||||
"""Create a new Dyson Carbon Filter Life sensor."""
|
||||
super().__init__(device, "carbon_filter_state")
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
"""Return filter life remaining in percent."""
|
||||
return int(self._device.state.carbon_filter_state)
|
||||
|
||||
|
||||
class DysonHepaFilterLifeSensor(DysonSensor):
|
||||
"""Representation of Dyson HEPA (or Combi) Filter Life sensor (in percent)."""
|
||||
|
||||
def __init__(self, device, filter_type="hepa"):
|
||||
"""Create a new Dyson Filter Life sensor."""
|
||||
super().__init__(device, f"{filter_type}_filter_state")
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
"""Return filter life remaining in percent."""
|
||||
return int(self._device.state.hepa_filter_state)
|
||||
|
||||
|
||||
class DysonDustSensor(DysonSensor):
|
||||
"""Representation of Dyson Dust sensor (lower is better)."""
|
||||
|
||||
def __init__(self, device):
|
||||
"""Create a new Dyson Dust sensor."""
|
||||
super().__init__(device, "dust")
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
"""Return Dust value."""
|
||||
return self._device.environmental_state.dust
|
||||
|
||||
|
||||
class DysonHumiditySensor(DysonSensor):
|
||||
"""Representation of Dyson Humidity sensor."""
|
||||
|
||||
def __init__(self, device):
|
||||
"""Create a new Dyson Humidity sensor."""
|
||||
super().__init__(device, "humidity")
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
"""Return Humidity value."""
|
||||
if self._device.environmental_state.humidity == 0:
|
||||
return STATE_OFF
|
||||
return self._device.environmental_state.humidity
|
||||
|
||||
|
||||
class DysonTemperatureSensor(DysonSensor):
|
||||
"""Representation of Dyson Temperature sensor."""
|
||||
|
||||
def __init__(self, device, unit):
|
||||
"""Create a new Dyson Temperature sensor."""
|
||||
super().__init__(device, "temperature")
|
||||
self._unit = unit
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
"""Return Temperature value."""
|
||||
temperature_kelvin = self._device.environmental_state.temperature
|
||||
if temperature_kelvin == 0:
|
||||
return STATE_OFF
|
||||
if self._unit == TEMP_CELSIUS:
|
||||
return float(f"{(temperature_kelvin - 273.15):.1f}")
|
||||
return float(f"{(temperature_kelvin * 9 / 5 - 459.67):.1f}")
|
||||
|
||||
@property
|
||||
def native_unit_of_measurement(self):
|
||||
"""Return the unit the value is expressed in."""
|
||||
return self._unit
|
||||
|
||||
|
||||
class DysonAirQualitySensor(DysonSensor):
|
||||
"""Representation of Dyson Air Quality sensor (lower is better)."""
|
||||
|
||||
def __init__(self, device):
|
||||
"""Create a new Dyson Air Quality sensor."""
|
||||
super().__init__(device, "air_quality")
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
"""Return Air Quality value."""
|
||||
return int(self._device.environmental_state.volatil_organic_compounds)
|
@ -1,108 +0,0 @@
|
||||
# Describes the format for available fan services
|
||||
|
||||
set_night_mode:
|
||||
name: Set night mode
|
||||
description: Set the fan in night mode.
|
||||
target:
|
||||
entity:
|
||||
integration: dyson
|
||||
domain: fan
|
||||
fields:
|
||||
night_mode:
|
||||
name: Night mode
|
||||
description: Night mode status
|
||||
required: true
|
||||
selector:
|
||||
boolean:
|
||||
|
||||
set_auto_mode:
|
||||
name: Set auto mode
|
||||
description: Set the fan in auto mode.
|
||||
target:
|
||||
entity:
|
||||
integration: dyson
|
||||
domain: fan
|
||||
fields:
|
||||
auto_mode:
|
||||
name: Auto Mode
|
||||
description: Auto mode status
|
||||
required: true
|
||||
selector:
|
||||
boolean:
|
||||
|
||||
set_angle:
|
||||
name: Set angle
|
||||
description: Set the oscillation angle of the selected fan(s).
|
||||
target:
|
||||
entity:
|
||||
integration: dyson
|
||||
domain: fan
|
||||
fields:
|
||||
angle_low:
|
||||
name: Angle low
|
||||
description: The angle at which the oscillation should start
|
||||
required: true
|
||||
selector:
|
||||
number:
|
||||
min: 5
|
||||
max: 355
|
||||
unit_of_measurement: '°'
|
||||
angle_high:
|
||||
name: Angle high
|
||||
description: The angle at which the oscillation should end
|
||||
required: true
|
||||
selector:
|
||||
number:
|
||||
min: 5
|
||||
max: 355
|
||||
unit_of_measurement: '°'
|
||||
|
||||
set_flow_direction_front:
|
||||
name: Set flow direction front
|
||||
description: Set the fan flow direction.
|
||||
target:
|
||||
entity:
|
||||
integration: dyson
|
||||
domain: fan
|
||||
fields:
|
||||
flow_direction_front:
|
||||
name: Flow direction front
|
||||
description: Frontal flow direction
|
||||
required: true
|
||||
selector:
|
||||
boolean:
|
||||
|
||||
set_timer:
|
||||
name: Set timer
|
||||
description: Set the sleep timer.
|
||||
target:
|
||||
entity:
|
||||
integration: dyson
|
||||
domain: fan
|
||||
fields:
|
||||
timer:
|
||||
name: Timer
|
||||
description: The value in minutes to set the timer to, 0 to disable it
|
||||
required: true
|
||||
selector:
|
||||
number:
|
||||
min: 0
|
||||
max: 720
|
||||
unit_of_measurement: minutes
|
||||
|
||||
set_speed:
|
||||
name: Set speed
|
||||
description: Set the exact speed of the fan.
|
||||
target:
|
||||
entity:
|
||||
integration: dyson
|
||||
domain: fan
|
||||
fields:
|
||||
dyson_speed:
|
||||
name: Speed
|
||||
description: Speed
|
||||
required: true
|
||||
selector:
|
||||
number:
|
||||
min: 1
|
||||
max: 10
|
@ -1,171 +0,0 @@
|
||||
"""Support for the Dyson 360 eye vacuum cleaner robot."""
|
||||
import logging
|
||||
|
||||
from libpurecool.const import Dyson360EyeMode, PowerMode
|
||||
from libpurecool.dyson_360_eye import Dyson360Eye
|
||||
|
||||
from homeassistant.components.vacuum import (
|
||||
SUPPORT_BATTERY,
|
||||
SUPPORT_FAN_SPEED,
|
||||
SUPPORT_PAUSE,
|
||||
SUPPORT_RETURN_HOME,
|
||||
SUPPORT_STATUS,
|
||||
SUPPORT_STOP,
|
||||
SUPPORT_TURN_OFF,
|
||||
SUPPORT_TURN_ON,
|
||||
VacuumEntity,
|
||||
)
|
||||
from homeassistant.helpers.icon import icon_for_battery_level
|
||||
|
||||
from . import DYSON_DEVICES, DysonEntity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ATTR_CLEAN_ID = "clean_id"
|
||||
ATTR_FULL_CLEAN_TYPE = "full_clean_type"
|
||||
ATTR_POSITION = "position"
|
||||
|
||||
DYSON_360_EYE_DEVICES = "dyson_360_eye_devices"
|
||||
|
||||
SUPPORT_DYSON = (
|
||||
SUPPORT_TURN_ON
|
||||
| SUPPORT_TURN_OFF
|
||||
| SUPPORT_PAUSE
|
||||
| SUPPORT_RETURN_HOME
|
||||
| SUPPORT_FAN_SPEED
|
||||
| SUPPORT_STATUS
|
||||
| SUPPORT_BATTERY
|
||||
| SUPPORT_STOP
|
||||
)
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
"""Set up the Dyson 360 Eye robot vacuum platform."""
|
||||
_LOGGER.debug("Creating new Dyson 360 Eye robot vacuum")
|
||||
if DYSON_360_EYE_DEVICES not in hass.data:
|
||||
hass.data[DYSON_360_EYE_DEVICES] = []
|
||||
|
||||
# Get Dyson Devices from parent component
|
||||
for device in [d for d in hass.data[DYSON_DEVICES] if isinstance(d, Dyson360Eye)]:
|
||||
dyson_entity = Dyson360EyeDevice(device)
|
||||
hass.data[DYSON_360_EYE_DEVICES].append(dyson_entity)
|
||||
|
||||
add_entities(hass.data[DYSON_360_EYE_DEVICES])
|
||||
return True
|
||||
|
||||
|
||||
class Dyson360EyeDevice(DysonEntity, VacuumEntity):
|
||||
"""Dyson 360 Eye robot vacuum device."""
|
||||
|
||||
def __init__(self, device):
|
||||
"""Dyson 360 Eye robot vacuum device."""
|
||||
super().__init__(device, None)
|
||||
|
||||
@property
|
||||
def status(self):
|
||||
"""Return the status of the vacuum cleaner."""
|
||||
dyson_labels = {
|
||||
Dyson360EyeMode.INACTIVE_CHARGING: "Stopped - Charging",
|
||||
Dyson360EyeMode.INACTIVE_CHARGED: "Stopped - Charged",
|
||||
Dyson360EyeMode.FULL_CLEAN_PAUSED: "Paused",
|
||||
Dyson360EyeMode.FULL_CLEAN_RUNNING: "Cleaning",
|
||||
Dyson360EyeMode.FULL_CLEAN_ABORTED: "Returning home",
|
||||
Dyson360EyeMode.FULL_CLEAN_INITIATED: "Start cleaning",
|
||||
Dyson360EyeMode.FAULT_USER_RECOVERABLE: "Error - device blocked",
|
||||
Dyson360EyeMode.FAULT_REPLACE_ON_DOCK: "Error - Replace device on dock",
|
||||
Dyson360EyeMode.FULL_CLEAN_FINISHED: "Finished",
|
||||
Dyson360EyeMode.FULL_CLEAN_NEEDS_CHARGE: "Need charging",
|
||||
}
|
||||
return dyson_labels.get(self._device.state.state, self._device.state.state)
|
||||
|
||||
@property
|
||||
def battery_level(self):
|
||||
"""Return the battery level of the vacuum cleaner."""
|
||||
return self._device.state.battery_level
|
||||
|
||||
@property
|
||||
def fan_speed(self):
|
||||
"""Return the fan speed of the vacuum cleaner."""
|
||||
speed_labels = {PowerMode.MAX: "Max", PowerMode.QUIET: "Quiet"}
|
||||
return speed_labels[self._device.state.power_mode]
|
||||
|
||||
@property
|
||||
def fan_speed_list(self):
|
||||
"""Get the list of available fan speed steps of the vacuum cleaner."""
|
||||
return ["Quiet", "Max"]
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self):
|
||||
"""Return the specific state attributes of this vacuum cleaner."""
|
||||
return {ATTR_POSITION: str(self._device.state.position)}
|
||||
|
||||
@property
|
||||
def is_on(self) -> bool:
|
||||
"""Return True if entity is on."""
|
||||
return self._device.state.state in [
|
||||
Dyson360EyeMode.FULL_CLEAN_INITIATED,
|
||||
Dyson360EyeMode.FULL_CLEAN_ABORTED,
|
||||
Dyson360EyeMode.FULL_CLEAN_RUNNING,
|
||||
]
|
||||
|
||||
@property
|
||||
def available(self) -> bool:
|
||||
"""Return True if entity is available."""
|
||||
return True
|
||||
|
||||
@property
|
||||
def supported_features(self):
|
||||
"""Flag vacuum cleaner robot features that are supported."""
|
||||
return SUPPORT_DYSON
|
||||
|
||||
@property
|
||||
def battery_icon(self):
|
||||
"""Return the battery icon for the vacuum cleaner."""
|
||||
charging = self._device.state.state in [Dyson360EyeMode.INACTIVE_CHARGING]
|
||||
return icon_for_battery_level(
|
||||
battery_level=self.battery_level, charging=charging
|
||||
)
|
||||
|
||||
def turn_on(self, **kwargs):
|
||||
"""Turn the vacuum on."""
|
||||
_LOGGER.debug("Turn on device %s", self.name)
|
||||
if self._device.state.state in [Dyson360EyeMode.FULL_CLEAN_PAUSED]:
|
||||
self._device.resume()
|
||||
else:
|
||||
self._device.start()
|
||||
|
||||
def turn_off(self, **kwargs):
|
||||
"""Turn the vacuum off and return to home."""
|
||||
_LOGGER.debug("Turn off device %s", self.name)
|
||||
self._device.pause()
|
||||
|
||||
def stop(self, **kwargs):
|
||||
"""Stop the vacuum cleaner."""
|
||||
_LOGGER.debug("Stop device %s", self.name)
|
||||
self._device.pause()
|
||||
|
||||
def set_fan_speed(self, fan_speed, **kwargs):
|
||||
"""Set fan speed."""
|
||||
_LOGGER.debug("Set fan speed %s on device %s", fan_speed, self.name)
|
||||
power_modes = {"Quiet": PowerMode.QUIET, "Max": PowerMode.MAX}
|
||||
self._device.set_power_mode(power_modes[fan_speed])
|
||||
|
||||
def start_pause(self, **kwargs):
|
||||
"""Start, pause or resume the cleaning task."""
|
||||
if self._device.state.state in [Dyson360EyeMode.FULL_CLEAN_PAUSED]:
|
||||
_LOGGER.debug("Resume device %s", self.name)
|
||||
self._device.resume()
|
||||
elif self._device.state.state in [
|
||||
Dyson360EyeMode.INACTIVE_CHARGED,
|
||||
Dyson360EyeMode.INACTIVE_CHARGING,
|
||||
]:
|
||||
_LOGGER.debug("Start device %s", self.name)
|
||||
self._device.start()
|
||||
else:
|
||||
_LOGGER.debug("Pause device %s", self.name)
|
||||
self._device.pause()
|
||||
|
||||
def return_to_base(self, **kwargs):
|
||||
"""Set the vacuum cleaner to return to the dock."""
|
||||
_LOGGER.debug("Return to base device %s", self.name)
|
||||
self._device.abort()
|
@ -923,9 +923,6 @@ krakenex==2.1.0
|
||||
# homeassistant.components.eufy
|
||||
lakeside==0.12
|
||||
|
||||
# homeassistant.components.dyson
|
||||
libpurecool==0.6.4
|
||||
|
||||
# homeassistant.components.foscam
|
||||
libpyfoscam==1.0
|
||||
|
||||
|
@ -566,9 +566,6 @@ kostal_plenticore==0.2.0
|
||||
# homeassistant.components.kraken
|
||||
krakenex==2.1.0
|
||||
|
||||
# homeassistant.components.dyson
|
||||
libpurecool==0.6.4
|
||||
|
||||
# homeassistant.components.foscam
|
||||
libpyfoscam==1.0
|
||||
|
||||
|
@ -1 +0,0 @@
|
||||
"""Tests for the dyson component."""
|
@ -1,103 +0,0 @@
|
||||
"""Common utils for Dyson tests."""
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest import mock
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from libpurecool.const import SLEEP_TIMER_OFF, Dyson360EyeMode, FanMode, PowerMode
|
||||
from libpurecool.dyson_360_eye import Dyson360Eye
|
||||
from libpurecool.dyson_device import DysonDevice
|
||||
from libpurecool.dyson_pure_cool import DysonPureCool, FanSpeed
|
||||
from libpurecool.dyson_pure_cool_link import DysonPureCoolLink
|
||||
|
||||
from homeassistant.components.dyson import CONF_LANGUAGE, DOMAIN
|
||||
from homeassistant.const import CONF_DEVICES, CONF_PASSWORD, CONF_USERNAME
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
|
||||
SERIAL = "XX-XXXXX-XX"
|
||||
NAME = "Temp Name"
|
||||
ENTITY_NAME = "temp_name"
|
||||
IP_ADDRESS = "0.0.0.0"
|
||||
|
||||
BASE_PATH = "homeassistant.components.dyson"
|
||||
|
||||
CONFIG = {
|
||||
DOMAIN: {
|
||||
CONF_USERNAME: "user@example.com",
|
||||
CONF_PASSWORD: "password",
|
||||
CONF_LANGUAGE: "US",
|
||||
CONF_DEVICES: [
|
||||
{
|
||||
"device_id": SERIAL,
|
||||
"device_ip": IP_ADDRESS,
|
||||
}
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@callback
|
||||
def async_get_basic_device(spec: type[DysonDevice]) -> DysonDevice:
|
||||
"""Return a basic device with common fields filled out."""
|
||||
device = MagicMock(spec=spec)
|
||||
device.serial = SERIAL
|
||||
device.name = NAME
|
||||
device.connect = mock.Mock(return_value=True)
|
||||
device.auto_connect = mock.Mock(return_value=True)
|
||||
return device
|
||||
|
||||
|
||||
@callback
|
||||
def async_get_360eye_device(state=Dyson360EyeMode.FULL_CLEAN_RUNNING) -> Dyson360Eye:
|
||||
"""Return a Dyson 360 Eye device."""
|
||||
device = async_get_basic_device(Dyson360Eye)
|
||||
device.state.state = state
|
||||
device.state.battery_level = 85
|
||||
device.state.power_mode = PowerMode.QUIET
|
||||
device.state.position = (0, 0)
|
||||
return device
|
||||
|
||||
|
||||
@callback
|
||||
def async_get_purecoollink_device() -> DysonPureCoolLink:
|
||||
"""Return a Dyson Pure Cool Link device."""
|
||||
device = async_get_basic_device(DysonPureCoolLink)
|
||||
device.state.fan_mode = FanMode.FAN.value
|
||||
device.state.speed = FanSpeed.FAN_SPEED_1.value
|
||||
device.state.night_mode = "ON"
|
||||
device.state.oscillation = "ON"
|
||||
return device
|
||||
|
||||
|
||||
@callback
|
||||
def async_get_purecool_device() -> DysonPureCool:
|
||||
"""Return a Dyson Pure Cool device."""
|
||||
device = async_get_basic_device(DysonPureCool)
|
||||
device.state.fan_power = "ON"
|
||||
device.state.speed = FanSpeed.FAN_SPEED_1.value
|
||||
device.state.night_mode = "ON"
|
||||
device.state.oscillation = "OION"
|
||||
device.state.oscillation_angle_low = "0024"
|
||||
device.state.oscillation_angle_high = "0254"
|
||||
device.state.auto_mode = "OFF"
|
||||
device.state.front_direction = "ON"
|
||||
device.state.sleep_timer = SLEEP_TIMER_OFF
|
||||
device.state.hepa_filter_state = "0100"
|
||||
device.state.carbon_filter_state = "0100"
|
||||
return device
|
||||
|
||||
|
||||
async def async_update_device(
|
||||
hass: HomeAssistant, device: DysonDevice, state_type: type | None = None
|
||||
) -> None:
|
||||
"""Update the device using callback function."""
|
||||
callbacks = [args[0][0] for args in device.add_message_listener.call_args_list]
|
||||
message = MagicMock(spec=state_type)
|
||||
|
||||
# Combining sync calls to avoid multiple executors
|
||||
def _run_callbacks():
|
||||
for callback_fn in callbacks:
|
||||
callback_fn(message)
|
||||
|
||||
await hass.async_add_executor_job(_run_callbacks)
|
||||
await hass.async_block_till_done()
|
@ -1,38 +0,0 @@
|
||||
"""Configure pytest for Dyson tests."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from libpurecool.dyson_device import DysonDevice
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.dyson import DOMAIN
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .common import BASE_PATH, CONFIG
|
||||
|
||||
from tests.common import async_setup_component
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
async def device(hass: HomeAssistant, request) -> DysonDevice:
|
||||
"""Fixture to provide Dyson 360 Eye device."""
|
||||
platform = request.module.PLATFORM_DOMAIN
|
||||
get_device = request.module.async_get_device
|
||||
if hasattr(request, "param"):
|
||||
if isinstance(request.param, list):
|
||||
device = get_device(*request.param)
|
||||
else:
|
||||
device = get_device(request.param)
|
||||
else:
|
||||
device = get_device()
|
||||
with patch(f"{BASE_PATH}.DysonAccount.login", return_value=True), patch(
|
||||
f"{BASE_PATH}.DysonAccount.devices", return_value=[device]
|
||||
), patch(f"{BASE_PATH}.PLATFORMS", [platform]):
|
||||
# PLATFORMS is patched so that only the platform being tested is set up
|
||||
await async_setup_component(
|
||||
hass,
|
||||
DOMAIN,
|
||||
CONFIG,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
return device
|
@ -1,67 +0,0 @@
|
||||
"""Test the Dyson air quality component."""
|
||||
|
||||
from libpurecool.dyson_pure_cool import DysonPureCool
|
||||
from libpurecool.dyson_pure_state_v2 import DysonEnvironmentalSensorV2State
|
||||
|
||||
from homeassistant.components.air_quality import (
|
||||
ATTR_AQI,
|
||||
ATTR_NO2,
|
||||
ATTR_PM_2_5,
|
||||
ATTR_PM_10,
|
||||
DOMAIN as PLATFORM_DOMAIN,
|
||||
)
|
||||
from homeassistant.components.dyson.air_quality import ATTR_VOC
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
|
||||
from .common import ENTITY_NAME, async_get_purecool_device, async_update_device
|
||||
|
||||
ENTITY_ID = f"{PLATFORM_DOMAIN}.{ENTITY_NAME}"
|
||||
|
||||
MOCKED_VALUES = {
|
||||
ATTR_PM_2_5: 10,
|
||||
ATTR_PM_10: 20,
|
||||
ATTR_NO2: 30,
|
||||
ATTR_VOC: 40,
|
||||
}
|
||||
|
||||
MOCKED_UPDATED_VALUES = {
|
||||
ATTR_PM_2_5: 60,
|
||||
ATTR_PM_10: 50,
|
||||
ATTR_NO2: 40,
|
||||
ATTR_VOC: 30,
|
||||
}
|
||||
|
||||
|
||||
def _async_assign_values(device: DysonPureCool, values=MOCKED_VALUES) -> None:
|
||||
"""Assign mocked environmental states to the device."""
|
||||
device.environmental_state.particulate_matter_25 = values[ATTR_PM_2_5]
|
||||
device.environmental_state.particulate_matter_10 = values[ATTR_PM_10]
|
||||
device.environmental_state.nitrogen_dioxide = values[ATTR_NO2]
|
||||
device.environmental_state.volatile_organic_compounds = values[ATTR_VOC]
|
||||
|
||||
|
||||
@callback
|
||||
def async_get_device() -> DysonPureCool:
|
||||
"""Return a device of the given type."""
|
||||
device = async_get_purecool_device()
|
||||
_async_assign_values(device)
|
||||
return device
|
||||
|
||||
|
||||
async def test_air_quality(hass: HomeAssistant, device: DysonPureCool) -> None:
|
||||
"""Test the state and attributes of the air quality entity."""
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == str(MOCKED_VALUES[ATTR_PM_2_5])
|
||||
attributes = state.attributes
|
||||
for attr, value in MOCKED_VALUES.items():
|
||||
assert attributes[attr] == value
|
||||
assert attributes[ATTR_AQI] == 40
|
||||
|
||||
_async_assign_values(device, MOCKED_UPDATED_VALUES)
|
||||
await async_update_device(hass, device, DysonEnvironmentalSensorV2State)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == str(MOCKED_UPDATED_VALUES[ATTR_PM_2_5])
|
||||
attributes = state.attributes
|
||||
for attr, value in MOCKED_UPDATED_VALUES.items():
|
||||
assert attributes[attr] == value
|
||||
assert attributes[ATTR_AQI] == 60
|
@ -1,348 +0,0 @@
|
||||
"""Test the Dyson fan component."""
|
||||
from __future__ import annotations
|
||||
|
||||
from libpurecool.const import (
|
||||
AutoMode,
|
||||
FanPower,
|
||||
FanSpeed,
|
||||
FanState,
|
||||
FocusMode,
|
||||
HeatMode,
|
||||
HeatState,
|
||||
)
|
||||
from libpurecool.dyson_device import DysonDevice
|
||||
from libpurecool.dyson_pure_hotcool import DysonPureHotCool
|
||||
from libpurecool.dyson_pure_hotcool_link import DysonPureHotCoolLink
|
||||
from libpurecool.dyson_pure_state import DysonPureHotCoolState
|
||||
from libpurecool.dyson_pure_state_v2 import DysonPureHotCoolV2State
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.climate import DOMAIN as PLATFORM_DOMAIN
|
||||
from homeassistant.components.climate.const import (
|
||||
ATTR_CURRENT_HUMIDITY,
|
||||
ATTR_CURRENT_TEMPERATURE,
|
||||
ATTR_FAN_MODE,
|
||||
ATTR_FAN_MODES,
|
||||
ATTR_HVAC_ACTION,
|
||||
ATTR_HVAC_MODE,
|
||||
ATTR_HVAC_MODES,
|
||||
ATTR_MAX_TEMP,
|
||||
ATTR_MIN_TEMP,
|
||||
CURRENT_HVAC_COOL,
|
||||
CURRENT_HVAC_HEAT,
|
||||
CURRENT_HVAC_IDLE,
|
||||
CURRENT_HVAC_OFF,
|
||||
FAN_AUTO,
|
||||
FAN_DIFFUSE,
|
||||
FAN_FOCUS,
|
||||
FAN_HIGH,
|
||||
FAN_LOW,
|
||||
FAN_MEDIUM,
|
||||
FAN_OFF,
|
||||
HVAC_MODE_COOL,
|
||||
HVAC_MODE_HEAT,
|
||||
HVAC_MODE_OFF,
|
||||
SERVICE_SET_FAN_MODE,
|
||||
SERVICE_SET_HVAC_MODE,
|
||||
SERVICE_SET_TEMPERATURE,
|
||||
)
|
||||
from homeassistant.components.dyson.climate import (
|
||||
SUPPORT_FAN,
|
||||
SUPPORT_FAN_PCOOL,
|
||||
SUPPORT_FLAGS,
|
||||
SUPPORT_HVAC,
|
||||
SUPPORT_HVAC_PCOOL,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
ATTR_SUPPORTED_FEATURES,
|
||||
ATTR_TEMPERATURE,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from .common import (
|
||||
ENTITY_NAME,
|
||||
NAME,
|
||||
SERIAL,
|
||||
async_get_basic_device,
|
||||
async_update_device,
|
||||
)
|
||||
|
||||
ENTITY_ID = f"{PLATFORM_DOMAIN}.{ENTITY_NAME}"
|
||||
|
||||
|
||||
@callback
|
||||
def async_get_device(spec: type[DysonDevice]) -> DysonDevice:
|
||||
"""Return a Dyson climate device."""
|
||||
device = async_get_basic_device(spec)
|
||||
device.state.heat_target = 2900
|
||||
device.environmental_state.temperature = 275
|
||||
device.environmental_state.humidity = 50
|
||||
if spec == DysonPureHotCoolLink:
|
||||
device.state.heat_mode = HeatMode.HEAT_ON.value
|
||||
device.state.heat_state = HeatState.HEAT_STATE_ON.value
|
||||
device.state.focus_mode = FocusMode.FOCUS_ON.value
|
||||
else:
|
||||
device.state.fan_power = FanPower.POWER_ON.value
|
||||
device.state.heat_mode = HeatMode.HEAT_ON.value
|
||||
device.state.heat_state = HeatState.HEAT_STATE_ON.value
|
||||
device.state.auto_mode = AutoMode.AUTO_ON.value
|
||||
device.state.fan_state = FanState.FAN_OFF.value
|
||||
device.state.speed = FanSpeed.FAN_SPEED_AUTO.value
|
||||
return device
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"device", [DysonPureHotCoolLink, DysonPureHotCool], indirect=True
|
||||
)
|
||||
async def test_state_common(hass: HomeAssistant, device: DysonDevice) -> None:
|
||||
"""Test common state and attributes of two types of climate entities."""
|
||||
entity_registry = er.async_get(hass)
|
||||
assert entity_registry.async_get(ENTITY_ID).unique_id == SERIAL
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.name == NAME
|
||||
attributes = state.attributes
|
||||
assert attributes[ATTR_SUPPORTED_FEATURES] == SUPPORT_FLAGS
|
||||
assert attributes[ATTR_CURRENT_TEMPERATURE] == 2
|
||||
assert attributes[ATTR_CURRENT_HUMIDITY] == 50
|
||||
assert attributes[ATTR_TEMPERATURE] == 17
|
||||
assert attributes[ATTR_MIN_TEMP] == 1
|
||||
assert attributes[ATTR_MAX_TEMP] == 37
|
||||
|
||||
device.state.heat_target = 2800
|
||||
device.environmental_state.temperature = 0
|
||||
device.environmental_state.humidity = 0
|
||||
await async_update_device(
|
||||
hass,
|
||||
device,
|
||||
DysonPureHotCoolState
|
||||
if isinstance(device, DysonPureHotCoolLink)
|
||||
else DysonPureHotCoolV2State,
|
||||
)
|
||||
attributes = hass.states.get(ENTITY_ID).attributes
|
||||
assert attributes[ATTR_CURRENT_TEMPERATURE] is None
|
||||
assert ATTR_CURRENT_HUMIDITY not in attributes
|
||||
assert attributes[ATTR_TEMPERATURE] == 7
|
||||
|
||||
|
||||
@pytest.mark.parametrize("device", [DysonPureHotCoolLink], indirect=True)
|
||||
async def test_state_purehotcoollink(
|
||||
hass: HomeAssistant, device: DysonPureHotCoolLink
|
||||
) -> None:
|
||||
"""Test common state and attributes of a PureHotCoolLink entity."""
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == HVAC_MODE_HEAT
|
||||
attributes = state.attributes
|
||||
assert attributes[ATTR_HVAC_MODES] == SUPPORT_HVAC
|
||||
assert attributes[ATTR_HVAC_ACTION] == CURRENT_HVAC_HEAT
|
||||
assert attributes[ATTR_FAN_MODE] == FAN_FOCUS
|
||||
assert attributes[ATTR_FAN_MODES] == SUPPORT_FAN
|
||||
|
||||
device.state.heat_state = HeatState.HEAT_STATE_OFF.value
|
||||
device.state.focus_mode = FocusMode.FOCUS_OFF
|
||||
await async_update_device(hass, device, DysonPureHotCoolState)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == HVAC_MODE_HEAT
|
||||
attributes = state.attributes
|
||||
assert attributes[ATTR_HVAC_ACTION] == CURRENT_HVAC_IDLE
|
||||
assert attributes[ATTR_FAN_MODE] == FAN_DIFFUSE
|
||||
|
||||
device.state.heat_mode = HeatMode.HEAT_OFF.value
|
||||
await async_update_device(hass, device, DysonPureHotCoolState)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == HVAC_MODE_COOL
|
||||
attributes = state.attributes
|
||||
assert attributes[ATTR_HVAC_ACTION] == CURRENT_HVAC_COOL
|
||||
|
||||
|
||||
@pytest.mark.parametrize("device", [DysonPureHotCool], indirect=True)
|
||||
async def test_state_purehotcool(hass: HomeAssistant, device: DysonPureHotCool) -> None:
|
||||
"""Test common state and attributes of a PureHotCool entity."""
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == HVAC_MODE_HEAT
|
||||
attributes = state.attributes
|
||||
assert attributes[ATTR_HVAC_MODES] == SUPPORT_HVAC_PCOOL
|
||||
assert attributes[ATTR_HVAC_ACTION] == CURRENT_HVAC_HEAT
|
||||
assert attributes[ATTR_FAN_MODE] == FAN_AUTO
|
||||
assert attributes[ATTR_FAN_MODES] == SUPPORT_FAN_PCOOL
|
||||
|
||||
device.state.heat_state = HeatState.HEAT_STATE_OFF.value
|
||||
device.state.auto_mode = AutoMode.AUTO_OFF.value
|
||||
await async_update_device(hass, device, DysonPureHotCoolV2State)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == HVAC_MODE_HEAT
|
||||
attributes = state.attributes
|
||||
assert attributes[ATTR_HVAC_ACTION] == CURRENT_HVAC_IDLE
|
||||
assert attributes[ATTR_FAN_MODE] == FAN_OFF
|
||||
|
||||
device.state.heat_mode = HeatMode.HEAT_OFF.value
|
||||
device.state.fan_state = FanState.FAN_ON.value
|
||||
device.state.speed = FanSpeed.FAN_SPEED_1.value
|
||||
await async_update_device(hass, device, DysonPureHotCoolV2State)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == HVAC_MODE_COOL
|
||||
attributes = state.attributes
|
||||
assert attributes[ATTR_HVAC_ACTION] == CURRENT_HVAC_COOL
|
||||
assert attributes[ATTR_FAN_MODE] == FAN_LOW
|
||||
|
||||
device.state.fan_power = FanPower.POWER_OFF.value
|
||||
await async_update_device(hass, device, DysonPureHotCoolV2State)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == HVAC_MODE_OFF
|
||||
attributes = state.attributes
|
||||
assert attributes[ATTR_HVAC_ACTION] == CURRENT_HVAC_OFF
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"service,service_data,configuration_data",
|
||||
[
|
||||
(
|
||||
SERVICE_SET_TEMPERATURE,
|
||||
{ATTR_TEMPERATURE: -5},
|
||||
{"heat_target": "2740", "heat_mode": HeatMode.HEAT_ON},
|
||||
),
|
||||
(
|
||||
SERVICE_SET_TEMPERATURE,
|
||||
{ATTR_TEMPERATURE: 40},
|
||||
{"heat_target": "3100", "heat_mode": HeatMode.HEAT_ON},
|
||||
),
|
||||
(
|
||||
SERVICE_SET_TEMPERATURE,
|
||||
{ATTR_TEMPERATURE: 20},
|
||||
{"heat_target": "2930", "heat_mode": HeatMode.HEAT_ON},
|
||||
),
|
||||
(
|
||||
SERVICE_SET_FAN_MODE,
|
||||
{ATTR_FAN_MODE: FAN_FOCUS},
|
||||
{"focus_mode": FocusMode.FOCUS_ON},
|
||||
),
|
||||
(
|
||||
SERVICE_SET_FAN_MODE,
|
||||
{ATTR_FAN_MODE: FAN_DIFFUSE},
|
||||
{"focus_mode": FocusMode.FOCUS_OFF},
|
||||
),
|
||||
(
|
||||
SERVICE_SET_HVAC_MODE,
|
||||
{ATTR_HVAC_MODE: HVAC_MODE_HEAT},
|
||||
{"heat_mode": HeatMode.HEAT_ON},
|
||||
),
|
||||
(
|
||||
SERVICE_SET_HVAC_MODE,
|
||||
{ATTR_HVAC_MODE: HVAC_MODE_COOL},
|
||||
{"heat_mode": HeatMode.HEAT_OFF},
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("device", [DysonPureHotCoolLink], indirect=True)
|
||||
async def test_commands_purehotcoollink(
|
||||
hass: HomeAssistant,
|
||||
device: DysonPureHotCoolLink,
|
||||
service: str,
|
||||
service_data: dict,
|
||||
configuration_data: dict,
|
||||
) -> None:
|
||||
"""Test sending commands to a PureHotCoolLink entity."""
|
||||
await hass.services.async_call(
|
||||
PLATFORM_DOMAIN,
|
||||
service,
|
||||
{
|
||||
ATTR_ENTITY_ID: ENTITY_ID,
|
||||
**service_data,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
device.set_configuration.assert_called_once_with(**configuration_data)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"service,service_data,command,command_args",
|
||||
[
|
||||
(SERVICE_SET_TEMPERATURE, {ATTR_TEMPERATURE: 20}, "set_heat_target", ["2930"]),
|
||||
(SERVICE_SET_FAN_MODE, {ATTR_FAN_MODE: FAN_OFF}, "turn_off", []),
|
||||
(
|
||||
SERVICE_SET_FAN_MODE,
|
||||
{ATTR_FAN_MODE: FAN_LOW},
|
||||
"set_fan_speed",
|
||||
[FanSpeed.FAN_SPEED_4],
|
||||
),
|
||||
(
|
||||
SERVICE_SET_FAN_MODE,
|
||||
{ATTR_FAN_MODE: FAN_MEDIUM},
|
||||
"set_fan_speed",
|
||||
[FanSpeed.FAN_SPEED_7],
|
||||
),
|
||||
(
|
||||
SERVICE_SET_FAN_MODE,
|
||||
{ATTR_FAN_MODE: FAN_HIGH},
|
||||
"set_fan_speed",
|
||||
[FanSpeed.FAN_SPEED_10],
|
||||
),
|
||||
(SERVICE_SET_FAN_MODE, {ATTR_FAN_MODE: FAN_AUTO}, "enable_auto_mode", []),
|
||||
(SERVICE_SET_HVAC_MODE, {ATTR_HVAC_MODE: HVAC_MODE_OFF}, "turn_off", []),
|
||||
(
|
||||
SERVICE_SET_HVAC_MODE,
|
||||
{ATTR_HVAC_MODE: HVAC_MODE_HEAT},
|
||||
"enable_heat_mode",
|
||||
[],
|
||||
),
|
||||
(
|
||||
SERVICE_SET_HVAC_MODE,
|
||||
{ATTR_HVAC_MODE: HVAC_MODE_COOL},
|
||||
"disable_heat_mode",
|
||||
[],
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("device", [DysonPureHotCool], indirect=True)
|
||||
async def test_commands_purehotcool(
|
||||
hass: HomeAssistant,
|
||||
device: DysonPureHotCoolLink,
|
||||
service: str,
|
||||
service_data: dict,
|
||||
command: str,
|
||||
command_args: list,
|
||||
) -> None:
|
||||
"""Test sending commands to a PureHotCool entity."""
|
||||
await hass.services.async_call(
|
||||
PLATFORM_DOMAIN,
|
||||
service,
|
||||
{
|
||||
ATTR_ENTITY_ID: ENTITY_ID,
|
||||
**service_data,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
getattr(device, command).assert_called_once_with(*command_args)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("hvac_mode", [HVAC_MODE_HEAT, HVAC_MODE_COOL])
|
||||
@pytest.mark.parametrize(
|
||||
"fan_power,turn_on_call_count",
|
||||
[
|
||||
(FanPower.POWER_ON.value, 0),
|
||||
(FanPower.POWER_OFF.value, 1),
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("device", [DysonPureHotCool], indirect=True)
|
||||
async def test_set_hvac_mode_purehotcool(
|
||||
hass: HomeAssistant,
|
||||
device: DysonPureHotCoolLink,
|
||||
hvac_mode: str,
|
||||
fan_power: str,
|
||||
turn_on_call_count: int,
|
||||
) -> None:
|
||||
"""Test setting HVAC mode of a PureHotCool entity turns on the device when it's off."""
|
||||
device.state.fan_power = fan_power
|
||||
await async_update_device(hass, device)
|
||||
await hass.services.async_call(
|
||||
PLATFORM_DOMAIN,
|
||||
SERVICE_SET_HVAC_MODE,
|
||||
{
|
||||
ATTR_ENTITY_ID: ENTITY_ID,
|
||||
ATTR_HVAC_MODE: hvac_mode,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
assert device.turn_on.call_count == turn_on_call_count
|
@ -1,441 +0,0 @@
|
||||
"""Test the Dyson fan component."""
|
||||
from __future__ import annotations
|
||||
|
||||
from libpurecool.const import FanMode, FanSpeed, NightMode, Oscillation
|
||||
from libpurecool.dyson_pure_cool import DysonPureCool, DysonPureCoolLink
|
||||
from libpurecool.dyson_pure_state import DysonPureCoolState
|
||||
from libpurecool.dyson_pure_state_v2 import DysonPureCoolV2State
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.dyson import DOMAIN
|
||||
from homeassistant.components.dyson.fan import (
|
||||
ATTR_ANGLE_HIGH,
|
||||
ATTR_ANGLE_LOW,
|
||||
ATTR_AUTO_MODE,
|
||||
ATTR_CARBON_FILTER,
|
||||
ATTR_DYSON_SPEED,
|
||||
ATTR_DYSON_SPEED_LIST,
|
||||
ATTR_FLOW_DIRECTION_FRONT,
|
||||
ATTR_HEPA_FILTER,
|
||||
ATTR_NIGHT_MODE,
|
||||
ATTR_TIMER,
|
||||
PRESET_MODE_AUTO,
|
||||
SERVICE_SET_ANGLE,
|
||||
SERVICE_SET_AUTO_MODE,
|
||||
SERVICE_SET_DYSON_SPEED,
|
||||
SERVICE_SET_FLOW_DIRECTION_FRONT,
|
||||
SERVICE_SET_NIGHT_MODE,
|
||||
SERVICE_SET_TIMER,
|
||||
)
|
||||
from homeassistant.components.fan import (
|
||||
ATTR_OSCILLATING,
|
||||
ATTR_PERCENTAGE,
|
||||
ATTR_PRESET_MODE,
|
||||
ATTR_SPEED,
|
||||
ATTR_SPEED_LIST,
|
||||
DOMAIN as PLATFORM_DOMAIN,
|
||||
SERVICE_OSCILLATE,
|
||||
SERVICE_SET_SPEED,
|
||||
SERVICE_TURN_OFF,
|
||||
SERVICE_TURN_ON,
|
||||
SPEED_HIGH,
|
||||
SPEED_LOW,
|
||||
SPEED_MEDIUM,
|
||||
SPEED_OFF,
|
||||
SUPPORT_OSCILLATE,
|
||||
SUPPORT_SET_SPEED,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
ATTR_SUPPORTED_FEATURES,
|
||||
STATE_OFF,
|
||||
STATE_ON,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from .common import (
|
||||
ENTITY_NAME,
|
||||
NAME,
|
||||
SERIAL,
|
||||
async_get_purecool_device,
|
||||
async_get_purecoollink_device,
|
||||
async_update_device,
|
||||
)
|
||||
|
||||
ENTITY_ID = f"{PLATFORM_DOMAIN}.{ENTITY_NAME}"
|
||||
|
||||
|
||||
@callback
|
||||
def async_get_device(spec: type[DysonPureCoolLink]) -> DysonPureCoolLink:
|
||||
"""Return a Dyson fan device."""
|
||||
if spec == DysonPureCoolLink:
|
||||
return async_get_purecoollink_device()
|
||||
return async_get_purecool_device()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("device", [DysonPureCoolLink], indirect=True)
|
||||
async def test_state_purecoollink(
|
||||
hass: HomeAssistant, device: DysonPureCoolLink
|
||||
) -> None:
|
||||
"""Test the state of a PureCoolLink fan."""
|
||||
entity_registry = er.async_get(hass)
|
||||
assert entity_registry.async_get(ENTITY_ID).unique_id == SERIAL
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_ON
|
||||
assert state.name == NAME
|
||||
attributes = state.attributes
|
||||
assert attributes[ATTR_NIGHT_MODE] is True
|
||||
assert attributes[ATTR_OSCILLATING] is True
|
||||
assert attributes[ATTR_PERCENTAGE] == 10
|
||||
assert attributes[ATTR_PRESET_MODE] is None
|
||||
assert attributes[ATTR_SPEED] == SPEED_LOW
|
||||
assert attributes[ATTR_SPEED_LIST] == [
|
||||
SPEED_OFF,
|
||||
SPEED_LOW,
|
||||
SPEED_MEDIUM,
|
||||
SPEED_HIGH,
|
||||
PRESET_MODE_AUTO,
|
||||
]
|
||||
assert attributes[ATTR_DYSON_SPEED] == 1
|
||||
assert attributes[ATTR_DYSON_SPEED_LIST] == list(range(1, 11))
|
||||
assert attributes[ATTR_AUTO_MODE] is False
|
||||
assert attributes[ATTR_SUPPORTED_FEATURES] == SUPPORT_OSCILLATE | SUPPORT_SET_SPEED
|
||||
|
||||
device.state.fan_mode = FanMode.OFF.value
|
||||
await async_update_device(hass, device, DysonPureCoolState)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_OFF
|
||||
|
||||
device.state.fan_mode = FanMode.AUTO.value
|
||||
device.state.speed = FanSpeed.FAN_SPEED_AUTO.value
|
||||
device.state.night_mode = "OFF"
|
||||
device.state.oscillation = "OFF"
|
||||
await async_update_device(hass, device, DysonPureCoolState)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_ON
|
||||
attributes = state.attributes
|
||||
assert attributes[ATTR_NIGHT_MODE] is False
|
||||
assert attributes[ATTR_OSCILLATING] is False
|
||||
assert attributes[ATTR_PERCENTAGE] is None
|
||||
assert attributes[ATTR_PRESET_MODE] == "auto"
|
||||
assert attributes[ATTR_SPEED] == PRESET_MODE_AUTO
|
||||
assert attributes[ATTR_DYSON_SPEED] == "AUTO"
|
||||
assert attributes[ATTR_AUTO_MODE] is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("device", [DysonPureCool], indirect=True)
|
||||
async def test_state_purecool(hass: HomeAssistant, device: DysonPureCool) -> None:
|
||||
"""Test the state of a PureCool fan."""
|
||||
entity_registry = er.async_get(hass)
|
||||
assert entity_registry.async_get(ENTITY_ID).unique_id == SERIAL
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_ON
|
||||
assert state.name == NAME
|
||||
attributes = state.attributes
|
||||
assert attributes[ATTR_NIGHT_MODE] is True
|
||||
assert attributes[ATTR_OSCILLATING] is True
|
||||
assert attributes[ATTR_ANGLE_LOW] == 24
|
||||
assert attributes[ATTR_ANGLE_HIGH] == 254
|
||||
assert attributes[ATTR_PERCENTAGE] == 10
|
||||
assert attributes[ATTR_PRESET_MODE] is None
|
||||
assert attributes[ATTR_SPEED] == SPEED_LOW
|
||||
assert attributes[ATTR_SPEED_LIST] == [
|
||||
SPEED_OFF,
|
||||
SPEED_LOW,
|
||||
SPEED_MEDIUM,
|
||||
SPEED_HIGH,
|
||||
PRESET_MODE_AUTO,
|
||||
]
|
||||
assert attributes[ATTR_DYSON_SPEED] == 1
|
||||
assert attributes[ATTR_DYSON_SPEED_LIST] == list(range(1, 11))
|
||||
assert attributes[ATTR_AUTO_MODE] is False
|
||||
assert attributes[ATTR_FLOW_DIRECTION_FRONT] is True
|
||||
assert attributes[ATTR_TIMER] == "OFF"
|
||||
assert attributes[ATTR_HEPA_FILTER] == 100
|
||||
assert attributes[ATTR_CARBON_FILTER] == 100
|
||||
assert attributes[ATTR_SUPPORTED_FEATURES] == SUPPORT_OSCILLATE | SUPPORT_SET_SPEED
|
||||
|
||||
device.state.auto_mode = "ON"
|
||||
device.state.night_mode = "OFF"
|
||||
device.state.oscillation = "OIOF"
|
||||
device.state.speed = "AUTO"
|
||||
device.state.front_direction = "OFF"
|
||||
device.state.sleep_timer = "0120"
|
||||
device.state.carbon_filter_state = "INV"
|
||||
await async_update_device(hass, device, DysonPureCoolV2State)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
attributes = state.attributes
|
||||
assert attributes[ATTR_NIGHT_MODE] is False
|
||||
assert attributes[ATTR_OSCILLATING] is False
|
||||
assert attributes[ATTR_PERCENTAGE] is None
|
||||
assert attributes[ATTR_PRESET_MODE] == "auto"
|
||||
assert attributes[ATTR_SPEED] == PRESET_MODE_AUTO
|
||||
assert attributes[ATTR_DYSON_SPEED] == "AUTO"
|
||||
assert attributes[ATTR_AUTO_MODE] is True
|
||||
assert attributes[ATTR_FLOW_DIRECTION_FRONT] is False
|
||||
assert attributes[ATTR_TIMER] == "0120"
|
||||
assert attributes[ATTR_CARBON_FILTER] == "INV"
|
||||
|
||||
device.state.fan_power = "OFF"
|
||||
await async_update_device(hass, device, DysonPureCoolV2State)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_OFF
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"service,service_data,configuration_args",
|
||||
[
|
||||
(SERVICE_TURN_ON, {}, {"fan_mode": FanMode.FAN}),
|
||||
(
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_SPEED: SPEED_LOW},
|
||||
{"fan_mode": FanMode.FAN, "fan_speed": FanSpeed.FAN_SPEED_4},
|
||||
),
|
||||
(
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_PERCENTAGE: 40},
|
||||
{"fan_mode": FanMode.FAN, "fan_speed": FanSpeed.FAN_SPEED_4},
|
||||
),
|
||||
(SERVICE_TURN_OFF, {}, {"fan_mode": FanMode.OFF}),
|
||||
(
|
||||
SERVICE_OSCILLATE,
|
||||
{ATTR_OSCILLATING: True},
|
||||
{"oscillation": Oscillation.OSCILLATION_ON},
|
||||
),
|
||||
(
|
||||
SERVICE_OSCILLATE,
|
||||
{ATTR_OSCILLATING: False},
|
||||
{"oscillation": Oscillation.OSCILLATION_OFF},
|
||||
),
|
||||
(
|
||||
SERVICE_SET_SPEED,
|
||||
{ATTR_SPEED: SPEED_LOW},
|
||||
{"fan_mode": FanMode.FAN, "fan_speed": FanSpeed.FAN_SPEED_4},
|
||||
),
|
||||
(
|
||||
SERVICE_SET_SPEED,
|
||||
{ATTR_SPEED: SPEED_MEDIUM},
|
||||
{"fan_mode": FanMode.FAN, "fan_speed": FanSpeed.FAN_SPEED_7},
|
||||
),
|
||||
(
|
||||
SERVICE_SET_SPEED,
|
||||
{ATTR_SPEED: SPEED_HIGH},
|
||||
{"fan_mode": FanMode.FAN, "fan_speed": FanSpeed.FAN_SPEED_10},
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("device", [DysonPureCoolLink], indirect=True)
|
||||
async def test_commands_purecoollink(
|
||||
hass: HomeAssistant,
|
||||
device: DysonPureCoolLink,
|
||||
service: str,
|
||||
service_data: dict,
|
||||
configuration_args: dict,
|
||||
) -> None:
|
||||
"""Test sending commands to a PureCoolLink fan."""
|
||||
await hass.services.async_call(
|
||||
PLATFORM_DOMAIN,
|
||||
service,
|
||||
{
|
||||
ATTR_ENTITY_ID: ENTITY_ID,
|
||||
**service_data,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
device.set_configuration.assert_called_once_with(**configuration_args)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"service,service_data,command,command_args",
|
||||
[
|
||||
(SERVICE_TURN_ON, {}, "turn_on", []),
|
||||
(
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_SPEED: SPEED_LOW},
|
||||
"set_fan_speed",
|
||||
[FanSpeed.FAN_SPEED_4],
|
||||
),
|
||||
(
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_PERCENTAGE: 40},
|
||||
"set_fan_speed",
|
||||
[FanSpeed.FAN_SPEED_4],
|
||||
),
|
||||
(
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_PRESET_MODE: "auto"},
|
||||
"enable_auto_mode",
|
||||
[],
|
||||
),
|
||||
(SERVICE_TURN_OFF, {}, "turn_off", []),
|
||||
(SERVICE_OSCILLATE, {ATTR_OSCILLATING: True}, "enable_oscillation", []),
|
||||
(SERVICE_OSCILLATE, {ATTR_OSCILLATING: False}, "disable_oscillation", []),
|
||||
(
|
||||
SERVICE_SET_SPEED,
|
||||
{ATTR_SPEED: SPEED_LOW},
|
||||
"set_fan_speed",
|
||||
[FanSpeed.FAN_SPEED_4],
|
||||
),
|
||||
(
|
||||
SERVICE_SET_SPEED,
|
||||
{ATTR_SPEED: SPEED_MEDIUM},
|
||||
"set_fan_speed",
|
||||
[FanSpeed.FAN_SPEED_7],
|
||||
),
|
||||
(
|
||||
SERVICE_SET_SPEED,
|
||||
{ATTR_SPEED: SPEED_HIGH},
|
||||
"set_fan_speed",
|
||||
[FanSpeed.FAN_SPEED_10],
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("device", [DysonPureCool], indirect=True)
|
||||
async def test_commands_purecool(
|
||||
hass: HomeAssistant,
|
||||
device: DysonPureCool,
|
||||
service: str,
|
||||
service_data: dict,
|
||||
command: str,
|
||||
command_args: list,
|
||||
) -> None:
|
||||
"""Test sending commands to a PureCool fan."""
|
||||
await hass.services.async_call(
|
||||
PLATFORM_DOMAIN,
|
||||
service,
|
||||
{
|
||||
ATTR_ENTITY_ID: ENTITY_ID,
|
||||
**service_data,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
getattr(device, command).assert_called_once_with(*command_args)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"service,service_data,configuration_args",
|
||||
[
|
||||
(
|
||||
SERVICE_SET_NIGHT_MODE,
|
||||
{ATTR_NIGHT_MODE: True},
|
||||
{"night_mode": NightMode.NIGHT_MODE_ON},
|
||||
),
|
||||
(
|
||||
SERVICE_SET_NIGHT_MODE,
|
||||
{ATTR_NIGHT_MODE: False},
|
||||
{"night_mode": NightMode.NIGHT_MODE_OFF},
|
||||
),
|
||||
(SERVICE_SET_AUTO_MODE, {"auto_mode": True}, {"fan_mode": FanMode.AUTO}),
|
||||
(SERVICE_SET_AUTO_MODE, {"auto_mode": False}, {"fan_mode": FanMode.FAN}),
|
||||
(
|
||||
SERVICE_SET_DYSON_SPEED,
|
||||
{ATTR_DYSON_SPEED: "4"},
|
||||
{"fan_mode": FanMode.FAN, "fan_speed": FanSpeed.FAN_SPEED_4},
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("device", [DysonPureCoolLink], indirect=True)
|
||||
async def test_custom_services_purecoollink(
|
||||
hass: HomeAssistant,
|
||||
device: DysonPureCoolLink,
|
||||
service: str,
|
||||
service_data: dict,
|
||||
configuration_args: dict,
|
||||
) -> None:
|
||||
"""Test custom services of a PureCoolLink fan."""
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
service,
|
||||
{
|
||||
ATTR_ENTITY_ID: ENTITY_ID,
|
||||
**service_data,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
device.set_configuration.assert_called_once_with(**configuration_args)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"service,service_data,command,command_args",
|
||||
[
|
||||
(SERVICE_SET_NIGHT_MODE, {ATTR_NIGHT_MODE: True}, "enable_night_mode", []),
|
||||
(SERVICE_SET_NIGHT_MODE, {ATTR_NIGHT_MODE: False}, "disable_night_mode", []),
|
||||
(SERVICE_SET_AUTO_MODE, {ATTR_AUTO_MODE: True}, "enable_auto_mode", []),
|
||||
(SERVICE_SET_AUTO_MODE, {ATTR_AUTO_MODE: False}, "disable_auto_mode", []),
|
||||
(SERVICE_SET_AUTO_MODE, {ATTR_AUTO_MODE: False}, "disable_auto_mode", []),
|
||||
(
|
||||
SERVICE_SET_ANGLE,
|
||||
{ATTR_ANGLE_LOW: 10, ATTR_ANGLE_HIGH: 200},
|
||||
"enable_oscillation",
|
||||
[10, 200],
|
||||
),
|
||||
(
|
||||
SERVICE_SET_FLOW_DIRECTION_FRONT,
|
||||
{ATTR_FLOW_DIRECTION_FRONT: True},
|
||||
"enable_frontal_direction",
|
||||
[],
|
||||
),
|
||||
(
|
||||
SERVICE_SET_FLOW_DIRECTION_FRONT,
|
||||
{ATTR_FLOW_DIRECTION_FRONT: False},
|
||||
"disable_frontal_direction",
|
||||
[],
|
||||
),
|
||||
(SERVICE_SET_TIMER, {ATTR_TIMER: 0}, "disable_sleep_timer", []),
|
||||
(SERVICE_SET_TIMER, {ATTR_TIMER: 10}, "enable_sleep_timer", [10]),
|
||||
(
|
||||
SERVICE_SET_DYSON_SPEED,
|
||||
{ATTR_DYSON_SPEED: "4"},
|
||||
"set_fan_speed",
|
||||
[FanSpeed("0004")],
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("device", [DysonPureCool], indirect=True)
|
||||
async def test_custom_services_purecool(
|
||||
hass: HomeAssistant,
|
||||
device: DysonPureCool,
|
||||
service: str,
|
||||
service_data: dict,
|
||||
command: str,
|
||||
command_args: list,
|
||||
) -> None:
|
||||
"""Test custom services of a PureCool fan."""
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
service,
|
||||
{
|
||||
ATTR_ENTITY_ID: ENTITY_ID,
|
||||
**service_data,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
getattr(device, command).assert_called_once_with(*command_args)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"domain,service,data",
|
||||
[
|
||||
(PLATFORM_DOMAIN, SERVICE_TURN_ON, {ATTR_SPEED: "AUTO"}),
|
||||
(PLATFORM_DOMAIN, SERVICE_SET_SPEED, {ATTR_SPEED: "AUTO"}),
|
||||
(DOMAIN, SERVICE_SET_DYSON_SPEED, {ATTR_DYSON_SPEED: "11"}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("device", [DysonPureCool], indirect=True)
|
||||
async def test_custom_services_invalid_data(
|
||||
hass: HomeAssistant, device: DysonPureCool, domain: str, service: str, data: dict
|
||||
) -> None:
|
||||
"""Test custom services calling with invalid data."""
|
||||
with pytest.raises(ValueError):
|
||||
await hass.services.async_call(
|
||||
domain,
|
||||
service,
|
||||
{
|
||||
ATTR_ENTITY_ID: ENTITY_ID,
|
||||
**data,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
@ -1,100 +0,0 @@
|
||||
"""Test the parent Dyson component."""
|
||||
import copy
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from homeassistant.components.dyson import DOMAIN
|
||||
from homeassistant.const import CONF_DEVICES
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .common import (
|
||||
BASE_PATH,
|
||||
CONFIG,
|
||||
ENTITY_NAME,
|
||||
IP_ADDRESS,
|
||||
async_get_360eye_device,
|
||||
async_get_purecool_device,
|
||||
async_get_purecoollink_device,
|
||||
)
|
||||
|
||||
from tests.common import async_setup_component
|
||||
|
||||
|
||||
async def test_setup_manual(hass: HomeAssistant):
|
||||
"""Test set up the component with manually configured device IPs."""
|
||||
SERIAL_TEMPLATE = "XX-XXXXX-X{}"
|
||||
|
||||
# device1 works
|
||||
device1 = async_get_purecoollink_device()
|
||||
device1.serial = SERIAL_TEMPLATE.format(1)
|
||||
|
||||
# device2 failed to connect
|
||||
device2 = async_get_purecool_device()
|
||||
device2.serial = SERIAL_TEMPLATE.format(2)
|
||||
device2.connect = MagicMock(return_value=False)
|
||||
|
||||
# device3 throws exception during connection
|
||||
device3 = async_get_360eye_device()
|
||||
device3.serial = SERIAL_TEMPLATE.format(3)
|
||||
device3.connect = MagicMock(side_effect=OSError)
|
||||
|
||||
# device4 not configured in configuration
|
||||
device4 = async_get_360eye_device()
|
||||
device4.serial = SERIAL_TEMPLATE.format(4)
|
||||
|
||||
devices = [device1, device2, device3, device4]
|
||||
config = copy.deepcopy(CONFIG)
|
||||
config[DOMAIN][CONF_DEVICES] = [
|
||||
{
|
||||
"device_id": SERIAL_TEMPLATE.format(i),
|
||||
"device_ip": IP_ADDRESS,
|
||||
}
|
||||
for i in [1, 2, 3, 5] # 1 device missing and 1 device not existed
|
||||
]
|
||||
|
||||
with patch(f"{BASE_PATH}.DysonAccount.login", return_value=True) as login, patch(
|
||||
f"{BASE_PATH}.DysonAccount.devices", return_value=devices
|
||||
) as devices_method, patch(
|
||||
f"{BASE_PATH}.PLATFORMS", ["fan", "vacuum"]
|
||||
): # Patch platforms to get rid of sensors
|
||||
assert await async_setup_component(hass, DOMAIN, config)
|
||||
await hass.async_block_till_done()
|
||||
login.assert_called_once_with()
|
||||
devices_method.assert_called_once_with()
|
||||
|
||||
# Only one fan and zero vacuum is set up successfully
|
||||
assert hass.states.async_entity_ids() == [f"fan.{ENTITY_NAME}"]
|
||||
device1.connect.assert_called_once_with(IP_ADDRESS)
|
||||
device2.connect.assert_called_once_with(IP_ADDRESS)
|
||||
device3.connect.assert_called_once_with(IP_ADDRESS)
|
||||
device4.connect.assert_not_called()
|
||||
|
||||
|
||||
async def test_setup_autoconnect(hass: HomeAssistant):
|
||||
"""Test set up the component with auto connect."""
|
||||
# device1 works
|
||||
device1 = async_get_purecoollink_device()
|
||||
|
||||
# device2 failed to auto connect
|
||||
device2 = async_get_purecool_device()
|
||||
device2.auto_connect = MagicMock(return_value=False)
|
||||
|
||||
devices = [device1, device2]
|
||||
config = copy.deepcopy(CONFIG)
|
||||
config[DOMAIN].pop(CONF_DEVICES)
|
||||
|
||||
with patch(f"{BASE_PATH}.DysonAccount.login", return_value=True), patch(
|
||||
f"{BASE_PATH}.DysonAccount.devices", return_value=devices
|
||||
), patch(
|
||||
f"{BASE_PATH}.PLATFORMS", ["fan"]
|
||||
): # Patch platforms to get rid of sensors
|
||||
assert await async_setup_component(hass, DOMAIN, config)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.states.async_entity_ids_count() == 1
|
||||
|
||||
|
||||
async def test_login_failed(hass: HomeAssistant):
|
||||
"""Test login failure during setup."""
|
||||
with patch(f"{BASE_PATH}.DysonAccount.login", return_value=False):
|
||||
assert not await async_setup_component(hass, DOMAIN, CONFIG)
|
||||
await hass.async_block_till_done()
|
@ -1,183 +0,0 @@
|
||||
"""Test the Dyson sensor(s) component."""
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from libpurecool.dyson_pure_cool import DysonPureCool
|
||||
from libpurecool.dyson_pure_cool_link import DysonPureCoolLink
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.dyson import DOMAIN
|
||||
from homeassistant.components.dyson.sensor import SENSOR_ATTRIBUTES, SENSOR_NAMES
|
||||
from homeassistant.components.sensor import DOMAIN as PLATFORM_DOMAIN
|
||||
from homeassistant.const import (
|
||||
ATTR_UNIT_OF_MEASUREMENT,
|
||||
STATE_OFF,
|
||||
TEMP_CELSIUS,
|
||||
TEMP_FAHRENHEIT,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
from homeassistant.util.unit_system import IMPERIAL_SYSTEM, METRIC_SYSTEM, UnitSystem
|
||||
|
||||
from .common import (
|
||||
BASE_PATH,
|
||||
CONFIG,
|
||||
ENTITY_NAME,
|
||||
NAME,
|
||||
SERIAL,
|
||||
async_get_basic_device,
|
||||
async_update_device,
|
||||
)
|
||||
|
||||
from tests.common import async_setup_component
|
||||
|
||||
ENTITY_ID_PREFIX = f"{PLATFORM_DOMAIN}.{ENTITY_NAME}"
|
||||
|
||||
MOCKED_VALUES = {
|
||||
"filter_life": 100,
|
||||
"dust": 5,
|
||||
"humidity": 45,
|
||||
"temperature_kelvin": 295,
|
||||
"temperature": 21.9,
|
||||
"air_quality": 5,
|
||||
"hepa_filter_state": 50,
|
||||
"combi_filter_state": 50,
|
||||
"carbon_filter_state": 10,
|
||||
}
|
||||
|
||||
MOCKED_UPDATED_VALUES = {
|
||||
"filter_life": 30,
|
||||
"dust": 2,
|
||||
"humidity": 80,
|
||||
"temperature_kelvin": 240,
|
||||
"temperature": -33.1,
|
||||
"air_quality": 3,
|
||||
"hepa_filter_state": 30,
|
||||
"combi_filter_state": 30,
|
||||
"carbon_filter_state": 20,
|
||||
}
|
||||
|
||||
|
||||
@callback
|
||||
def _async_assign_values(
|
||||
device: DysonPureCoolLink, values=MOCKED_VALUES, combi=False
|
||||
) -> None:
|
||||
"""Assign mocked values to the device."""
|
||||
if isinstance(device, DysonPureCool):
|
||||
device.state.hepa_filter_state = values["hepa_filter_state"]
|
||||
device.state.carbon_filter_state = (
|
||||
"INV" if combi else values["carbon_filter_state"]
|
||||
)
|
||||
device.environmental_state.humidity = values["humidity"]
|
||||
device.environmental_state.temperature = values["temperature_kelvin"]
|
||||
else: # DysonPureCoolLink
|
||||
device.state.filter_life = values["filter_life"]
|
||||
device.environmental_state.dust = values["dust"]
|
||||
device.environmental_state.humidity = values["humidity"]
|
||||
device.environmental_state.temperature = values["temperature_kelvin"]
|
||||
device.environmental_state.volatil_organic_compounds = values["air_quality"]
|
||||
|
||||
|
||||
@callback
|
||||
def async_get_device(spec: type[DysonPureCoolLink], combi=False) -> DysonPureCoolLink:
|
||||
"""Return a device of the given type."""
|
||||
device = async_get_basic_device(spec)
|
||||
_async_assign_values(device, combi=combi)
|
||||
return device
|
||||
|
||||
|
||||
@callback
|
||||
def _async_get_entity_id(sensor_type: str) -> str:
|
||||
"""Get the expected entity id from the type of the sensor."""
|
||||
sensor_name = SENSOR_NAMES[sensor_type]
|
||||
entity_id_suffix = sensor_name.lower().replace(" ", "_")
|
||||
return f"{ENTITY_ID_PREFIX}_{entity_id_suffix}"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"device,sensors",
|
||||
[
|
||||
(
|
||||
DysonPureCoolLink,
|
||||
["filter_life", "dust", "humidity", "temperature", "air_quality"],
|
||||
),
|
||||
(
|
||||
DysonPureCool,
|
||||
["hepa_filter_state", "carbon_filter_state", "humidity", "temperature"],
|
||||
),
|
||||
(
|
||||
[DysonPureCool, True],
|
||||
["combi_filter_state", "humidity", "temperature"],
|
||||
),
|
||||
],
|
||||
indirect=["device"],
|
||||
)
|
||||
async def test_sensors(
|
||||
hass: HomeAssistant, device: DysonPureCoolLink, sensors: list[str]
|
||||
) -> None:
|
||||
"""Test the sensors."""
|
||||
# Temperature is given by the device in kelvin
|
||||
# Make sure no other sensors are set up
|
||||
assert len(hass.states.async_all()) == len(sensors)
|
||||
|
||||
entity_registry = er.async_get(hass)
|
||||
for sensor in sensors:
|
||||
entity_id = _async_get_entity_id(sensor)
|
||||
|
||||
# Test unique id
|
||||
assert entity_registry.async_get(entity_id).unique_id == f"{SERIAL}-{sensor}"
|
||||
|
||||
# Test state
|
||||
state = hass.states.get(entity_id)
|
||||
assert state.state == str(MOCKED_VALUES[sensor])
|
||||
assert state.name == f"{NAME} {SENSOR_NAMES[sensor]}"
|
||||
|
||||
# Test attributes
|
||||
attributes = state.attributes
|
||||
for attr, value in SENSOR_ATTRIBUTES[sensor].items():
|
||||
assert attributes[attr] == value
|
||||
|
||||
# Test data update
|
||||
_async_assign_values(device, MOCKED_UPDATED_VALUES)
|
||||
await async_update_device(hass, device)
|
||||
for sensor in sensors:
|
||||
state = hass.states.get(_async_get_entity_id(sensor))
|
||||
assert state.state == str(MOCKED_UPDATED_VALUES[sensor])
|
||||
|
||||
|
||||
@pytest.mark.parametrize("device", [DysonPureCoolLink], indirect=True)
|
||||
async def test_sensors_off(hass: HomeAssistant, device: DysonPureCoolLink) -> None:
|
||||
"""Test the case where temperature and humidity are not available."""
|
||||
device.environmental_state.temperature = 0
|
||||
device.environmental_state.humidity = 0
|
||||
await async_update_device(hass, device)
|
||||
assert hass.states.get(f"{ENTITY_ID_PREFIX}_temperature").state == STATE_OFF
|
||||
assert hass.states.get(f"{ENTITY_ID_PREFIX}_humidity").state == STATE_OFF
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"unit_system,temp_unit,temperature",
|
||||
[(METRIC_SYSTEM, TEMP_CELSIUS, 21.9), (IMPERIAL_SYSTEM, TEMP_FAHRENHEIT, 71.3)],
|
||||
)
|
||||
async def test_temperature(
|
||||
hass: HomeAssistant, unit_system: UnitSystem, temp_unit: str, temperature: float
|
||||
) -> None:
|
||||
"""Test the temperature sensor in different units."""
|
||||
hass.config.units = unit_system
|
||||
|
||||
device = async_get_device(DysonPureCoolLink)
|
||||
with patch(f"{BASE_PATH}.DysonAccount.login", return_value=True), patch(
|
||||
f"{BASE_PATH}.DysonAccount.devices", return_value=[device]
|
||||
), patch(f"{BASE_PATH}.PLATFORMS", [PLATFORM_DOMAIN]):
|
||||
# PLATFORMS is patched so that only the platform being tested is set up
|
||||
await async_setup_component(
|
||||
hass,
|
||||
DOMAIN,
|
||||
CONFIG,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(f"{ENTITY_ID_PREFIX}_temperature")
|
||||
assert state.state == str(temperature)
|
||||
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == temp_unit
|
@ -1,115 +0,0 @@
|
||||
"""Test the Dyson 360 eye robot vacuum component."""
|
||||
from libpurecool.const import Dyson360EyeMode, PowerMode
|
||||
from libpurecool.dyson_360_eye import Dyson360Eye
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.dyson.vacuum import ATTR_POSITION, SUPPORT_DYSON
|
||||
from homeassistant.components.vacuum import (
|
||||
ATTR_FAN_SPEED,
|
||||
ATTR_FAN_SPEED_LIST,
|
||||
ATTR_STATUS,
|
||||
DOMAIN as PLATFORM_DOMAIN,
|
||||
SERVICE_RETURN_TO_BASE,
|
||||
SERVICE_SET_FAN_SPEED,
|
||||
SERVICE_START_PAUSE,
|
||||
SERVICE_STOP,
|
||||
SERVICE_TURN_OFF,
|
||||
SERVICE_TURN_ON,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
ATTR_BATTERY_LEVEL,
|
||||
ATTR_ENTITY_ID,
|
||||
ATTR_SUPPORTED_FEATURES,
|
||||
STATE_OFF,
|
||||
STATE_ON,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from .common import (
|
||||
ENTITY_NAME,
|
||||
NAME,
|
||||
SERIAL,
|
||||
async_get_360eye_device,
|
||||
async_update_device,
|
||||
)
|
||||
|
||||
ENTITY_ID = f"{PLATFORM_DOMAIN}.{ENTITY_NAME}"
|
||||
|
||||
|
||||
@callback
|
||||
def async_get_device(state=Dyson360EyeMode.FULL_CLEAN_RUNNING) -> Dyson360Eye:
|
||||
"""Return a Dyson 360 Eye device."""
|
||||
return async_get_360eye_device(state)
|
||||
|
||||
|
||||
async def test_state(hass: HomeAssistant, device: Dyson360Eye) -> None:
|
||||
"""Test the state of the vacuum."""
|
||||
entity_registry = er.async_get(hass)
|
||||
assert entity_registry.async_get(ENTITY_ID).unique_id == SERIAL
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.name == NAME
|
||||
assert state.state == STATE_ON
|
||||
attributes = state.attributes
|
||||
assert attributes[ATTR_STATUS] == "Cleaning"
|
||||
assert attributes[ATTR_SUPPORTED_FEATURES] == SUPPORT_DYSON
|
||||
assert attributes[ATTR_BATTERY_LEVEL] == 85
|
||||
assert attributes[ATTR_POSITION] == "(0, 0)"
|
||||
assert attributes[ATTR_FAN_SPEED] == "Quiet"
|
||||
assert attributes[ATTR_FAN_SPEED_LIST] == ["Quiet", "Max"]
|
||||
|
||||
device.state.state = Dyson360EyeMode.INACTIVE_CHARGING
|
||||
device.state.power_mode = PowerMode.MAX
|
||||
await async_update_device(hass, device)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_OFF
|
||||
assert state.attributes[ATTR_STATUS] == "Stopped - Charging"
|
||||
assert state.attributes[ATTR_FAN_SPEED] == "Max"
|
||||
|
||||
device.state.state = Dyson360EyeMode.FULL_CLEAN_PAUSED
|
||||
await async_update_device(hass, device)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_OFF
|
||||
assert state.attributes[ATTR_STATUS] == "Paused"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"service,command,device",
|
||||
[
|
||||
(SERVICE_TURN_ON, "start", Dyson360EyeMode.INACTIVE_CHARGED),
|
||||
(SERVICE_TURN_ON, "resume", Dyson360EyeMode.FULL_CLEAN_PAUSED),
|
||||
(SERVICE_TURN_OFF, "pause", Dyson360EyeMode.FULL_CLEAN_RUNNING),
|
||||
(SERVICE_STOP, "pause", Dyson360EyeMode.FULL_CLEAN_RUNNING),
|
||||
(SERVICE_START_PAUSE, "pause", Dyson360EyeMode.FULL_CLEAN_RUNNING),
|
||||
(SERVICE_START_PAUSE, "pause", Dyson360EyeMode.FULL_CLEAN_RUNNING),
|
||||
(SERVICE_START_PAUSE, "start", Dyson360EyeMode.INACTIVE_CHARGED),
|
||||
(SERVICE_START_PAUSE, "resume", Dyson360EyeMode.FULL_CLEAN_PAUSED),
|
||||
(SERVICE_RETURN_TO_BASE, "abort", Dyson360EyeMode.FULL_CLEAN_PAUSED),
|
||||
],
|
||||
indirect=["device"],
|
||||
)
|
||||
async def test_commands(
|
||||
hass: HomeAssistant, device: Dyson360Eye, service: str, command: str
|
||||
) -> None:
|
||||
"""Test sending commands to the vacuum."""
|
||||
await hass.services.async_call(
|
||||
PLATFORM_DOMAIN, service, {ATTR_ENTITY_ID: ENTITY_ID}, blocking=True
|
||||
)
|
||||
getattr(device, command).assert_called_once_with()
|
||||
|
||||
|
||||
async def test_set_fan_speed(hass: HomeAssistant, device: Dyson360Eye):
|
||||
"""Test setting fan speed of the vacuum."""
|
||||
fan_speed_map = {
|
||||
"Max": PowerMode.MAX,
|
||||
"Quiet": PowerMode.QUIET,
|
||||
}
|
||||
for service_speed, command_speed in fan_speed_map.items():
|
||||
await hass.services.async_call(
|
||||
PLATFORM_DOMAIN,
|
||||
SERVICE_SET_FAN_SPEED,
|
||||
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_FAN_SPEED: service_speed},
|
||||
blocking=True,
|
||||
)
|
||||
device.set_power_mode.assert_called_with(command_speed)
|
Loading…
x
Reference in New Issue
Block a user