github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Add eq3btsmart integration (#109291) 

Co-authored-by: Sid <27780930+autinerd@users.noreply.github.com>
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Lennard Beers 2024-03-29 02:20:56 +01:00 committed by GitHub
parent 4adbf7c730
commit 282cbfc048
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
23 changed files with 965 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.coveragerc
View File

@ -362,6 +362,11 @@ omit =
homeassistant/components/epson/__init__.py
homeassistant/components/epson/media_player.py
homeassistant/components/epsonworkforce/sensor.py
homeassistant/components/eq3btsmart/__init__.py
homeassistant/components/eq3btsmart/climate.py
homeassistant/components/eq3btsmart/const.py
homeassistant/components/eq3btsmart/entity.py
homeassistant/components/eq3btsmart/models.py
homeassistant/components/escea/__init__.py
homeassistant/components/escea/climate.py
homeassistant/components/escea/discovery.py

1
.strict-typing
View File

@ -170,6 +170,7 @@ homeassistant.components.energy.*
homeassistant.components.energyzero.*
homeassistant.components.enigma2.*
homeassistant.components.enphase_envoy.*
homeassistant.components.eq3btsmart.*
homeassistant.components.esphome.*
homeassistant.components.event.*
homeassistant.components.evil_genius_labs.*

2
CODEOWNERS
View File

@ -396,6 +396,8 @@ build.json @home-assistant/supervisor
/homeassistant/components/epson/ @pszafer
/tests/components/epson/ @pszafer
/homeassistant/components/epsonworkforce/ @ThaStealth
/homeassistant/components/eq3btsmart/ @eulemitkeule @dbuezas
/tests/components/eq3btsmart/ @eulemitkeule @dbuezas
/homeassistant/components/escea/ @lazdavila
/tests/components/escea/ @lazdavila
/homeassistant/components/esphome/ @OttoWinter @jesserockz @kbx81 @bdraco

2
homeassistant/brands/eq3.json
View File

@ -1,5 +1,5 @@
{
"domain": "eq3",
"name": "eQ-3",
"integrations": ["maxcube"]
"integrations": ["maxcube", "eq3btsmart"]
}

145
homeassistant/components/eq3btsmart/__init__.py Normal file
View File

@ -0,0 +1,145 @@
"""Support for EQ3 devices."""
import asyncio
import logging
from typing import TYPE_CHECKING
from eq3btsmart import Thermostat
from eq3btsmart.exceptions import Eq3Exception
from eq3btsmart.thermostat_config import ThermostatConfig
from homeassistant.components import bluetooth
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.dispatcher import async_dispatcher_send
from .const import DOMAIN, SIGNAL_THERMOSTAT_CONNECTED, SIGNAL_THERMOSTAT_DISCONNECTED
from .models import Eq3Config, Eq3ConfigEntryData
PLATFORMS = [
Platform.CLIMATE,
]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Handle config entry setup."""
mac_address: str | None = entry.unique_id
if TYPE_CHECKING:
assert mac_address is not None
eq3_config = Eq3Config(
mac_address=mac_address,
)
device = bluetooth.async_ble_device_from_address(
hass, mac_address.upper(), connectable=True
)
if device is None:
raise ConfigEntryNotReady(
f"[{eq3_config.mac_address}] Device could not be found"
)
thermostat = Thermostat(
thermostat_config=ThermostatConfig(
mac_address=mac_address,
),
ble_device=device,
)
eq3_config_entry = Eq3ConfigEntryData(eq3_config=eq3_config, thermostat=thermostat)
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = eq3_config_entry
entry.async_on_unload(entry.add_update_listener(update_listener))
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_create_background_task(
hass, _async_run_thermostat(hass, entry), entry.entry_id
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Handle config entry unload."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
eq3_config_entry: Eq3ConfigEntryData = hass.data[DOMAIN].pop(entry.entry_id)
await eq3_config_entry.thermostat.async_disconnect()
return unload_ok
async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle config entry update."""
await hass.config_entries.async_reload(entry.entry_id)
async def _async_run_thermostat(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Run the thermostat."""
eq3_config_entry: Eq3ConfigEntryData = hass.data[DOMAIN][entry.entry_id]
thermostat = eq3_config_entry.thermostat
mac_address = eq3_config_entry.eq3_config.mac_address
scan_interval = eq3_config_entry.eq3_config.scan_interval
await _async_reconnect_thermostat(hass, entry)
while True:
try:
await thermostat.async_get_status()
except Eq3Exception as e:
if not thermostat.is_connected:
_LOGGER.error(
"[%s] eQ-3 device disconnected",
mac_address,
)
async_dispatcher_send(
hass,
f"{SIGNAL_THERMOSTAT_DISCONNECTED}_{mac_address}",
)
await _async_reconnect_thermostat(hass, entry)
continue
_LOGGER.error(
"[%s] Error updating eQ-3 device: %s",
mac_address,
e,
)
await asyncio.sleep(scan_interval)
async def _async_reconnect_thermostat(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Reconnect the thermostat."""
eq3_config_entry: Eq3ConfigEntryData = hass.data[DOMAIN][entry.entry_id]
thermostat = eq3_config_entry.thermostat
mac_address = eq3_config_entry.eq3_config.mac_address
scan_interval = eq3_config_entry.eq3_config.scan_interval
while True:
try:
await thermostat.async_connect()
except Eq3Exception:
await asyncio.sleep(scan_interval)
continue
_LOGGER.debug(
"[%s] eQ-3 device connected",
mac_address,
)
async_dispatcher_send(
hass,
f"{SIGNAL_THERMOSTAT_CONNECTED}_{mac_address}",
)
return

306
homeassistant/components/eq3btsmart/climate.py Normal file
View File

@ -0,0 +1,306 @@
"""Platform for eQ-3 climate entities."""
import logging
from typing import Any
from eq3btsmart import Thermostat
from eq3btsmart.const import EQ3BT_MAX_TEMP, EQ3BT_OFF_TEMP, Eq3Preset, OperationMode
from eq3btsmart.exceptions import Eq3Exception
from homeassistant.components.climate import (
ATTR_HVAC_MODE,
PRESET_NONE,
ClimateEntity,
ClimateEntityFeature,
HVACAction,
HVACMode,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_TEMPERATURE, PRECISION_HALVES, UnitOfTemperature
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers.device_registry import (
CONNECTION_BLUETOOTH,
DeviceInfo,
async_get,
format_mac,
)
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util import slugify
from .const import (
DEVICE_MODEL,
DOMAIN,
EQ_TO_HA_HVAC,
HA_TO_EQ_HVAC,
MANUFACTURER,
SIGNAL_THERMOSTAT_CONNECTED,
SIGNAL_THERMOSTAT_DISCONNECTED,
CurrentTemperatureSelector,
Preset,
TargetTemperatureSelector,
)
from .entity import Eq3Entity
from .models import Eq3Config, Eq3ConfigEntryData
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Handle config entry setup."""
eq3_config_entry: Eq3ConfigEntryData = hass.data[DOMAIN][config_entry.entry_id]
async_add_entities(
[Eq3Climate(eq3_config_entry.eq3_config, eq3_config_entry.thermostat)],
)
class Eq3Climate(Eq3Entity, ClimateEntity):
"""Climate entity to represent a eQ-3 thermostat."""
_attr_name = None
_attr_supported_features = (
ClimateEntityFeature.TARGET_TEMPERATURE
| ClimateEntityFeature.PRESET_MODE
| ClimateEntityFeature.TURN_OFF
| ClimateEntityFeature.TURN_ON
)
_attr_temperature_unit = UnitOfTemperature.CELSIUS
_attr_min_temp = EQ3BT_OFF_TEMP
_attr_max_temp = EQ3BT_MAX_TEMP
_attr_precision = PRECISION_HALVES
_attr_hvac_modes = list(HA_TO_EQ_HVAC.keys())
_attr_preset_modes = list(Preset)
_attr_should_poll = False
_attr_available = False
_attr_hvac_mode: HVACMode | None = None
_attr_hvac_action: HVACAction | None = None
_attr_preset_mode: str | None = None
_target_temperature: float | None = None
def __init__(self, eq3_config: Eq3Config, thermostat: Thermostat) -> None:
"""Initialize the climate entity."""
super().__init__(eq3_config, thermostat)
self._attr_unique_id = format_mac(eq3_config.mac_address)
self._attr_device_info = DeviceInfo(
name=slugify(self._eq3_config.mac_address),
manufacturer=MANUFACTURER,
model=DEVICE_MODEL,
connections={(CONNECTION_BLUETOOTH, self._eq3_config.mac_address)},
)
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
self._thermostat.register_update_callback(self._async_on_updated)
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{SIGNAL_THERMOSTAT_DISCONNECTED}_{self._eq3_config.mac_address}",
self._async_on_disconnected,
)
)
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{SIGNAL_THERMOSTAT_CONNECTED}_{self._eq3_config.mac_address}",
self._async_on_connected,
)
)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
self._thermostat.unregister_update_callback(self._async_on_updated)
@callback
def _async_on_disconnected(self) -> None:
self._attr_available = False
self.async_write_ha_state()
@callback
def _async_on_connected(self) -> None:
self._attr_available = True
self.async_write_ha_state()
@callback
def _async_on_updated(self) -> None:
"""Handle updated data from the thermostat."""
if self._thermostat.status is not None:
self._async_on_status_updated()
if self._thermostat.device_data is not None:
self._async_on_device_updated()
self.async_write_ha_state()
@callback
def _async_on_status_updated(self) -> None:
"""Handle updated status from the thermostat."""
self._target_temperature = self._thermostat.status.target_temperature.value
self._attr_hvac_mode = EQ_TO_HA_HVAC[self._thermostat.status.operation_mode]
self._attr_current_temperature = self._get_current_temperature()
self._attr_target_temperature = self._get_target_temperature()
self._attr_preset_mode = self._get_current_preset_mode()
self._attr_hvac_action = self._get_current_hvac_action()
@callback
def _async_on_device_updated(self) -> None:
"""Handle updated device data from the thermostat."""
device_registry = async_get(self.hass)
if device := device_registry.async_get_device(
connections={(CONNECTION_BLUETOOTH, self._eq3_config.mac_address)},
):
device_registry.async_update_device(
device.id,
sw_version=self._thermostat.device_data.firmware_version,
serial_number=self._thermostat.device_data.device_serial.value,
)
def _get_current_temperature(self) -> float | None:
"""Return the current temperature."""
match self._eq3_config.current_temp_selector:
case CurrentTemperatureSelector.NOTHING:
return None
case CurrentTemperatureSelector.VALVE:
if self._thermostat.status is None:
return None
return float(self._thermostat.status.valve_temperature)
case CurrentTemperatureSelector.UI:
return self._target_temperature
case CurrentTemperatureSelector.DEVICE:
if self._thermostat.status is None:
return None
return float(self._thermostat.status.target_temperature.value)
case CurrentTemperatureSelector.ENTITY:
state = self.hass.states.get(self._eq3_config.external_temp_sensor)
if state is not None:
try:
return float(state.state)
except ValueError:
pass
return None
def _get_target_temperature(self) -> float | None:
"""Return the target temperature."""
match self._eq3_config.target_temp_selector:
case TargetTemperatureSelector.TARGET:
return self._target_temperature
case TargetTemperatureSelector.LAST_REPORTED:
if self._thermostat.status is None:
return None
return float(self._thermostat.status.target_temperature.value)
def _get_current_preset_mode(self) -> str:
"""Return the current preset mode."""
if (status := self._thermostat.status) is None:
return PRESET_NONE
if status.is_window_open:
return Preset.WINDOW_OPEN
if status.is_boost:
return Preset.BOOST
if status.is_low_battery:
return Preset.LOW_BATTERY
if status.is_away:
return Preset.AWAY
if status.operation_mode is OperationMode.ON:
return Preset.OPEN
if status.presets is None:
return PRESET_NONE
if status.target_temperature == status.presets.eco_temperature:
return Preset.ECO
if status.target_temperature == status.presets.comfort_temperature:
return Preset.COMFORT
return PRESET_NONE
def _get_current_hvac_action(self) -> HVACAction:
"""Return the current hvac action."""
if (
self._thermostat.status is None
or self._thermostat.status.operation_mode is OperationMode.OFF
):
return HVACAction.OFF
if self._thermostat.status.valve == 0:
return HVACAction.IDLE
return HVACAction.HEATING
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature."""
if ATTR_HVAC_MODE in kwargs:
mode: HVACMode | None
if (mode := kwargs.get(ATTR_HVAC_MODE)) is None:
return
if mode is not HVACMode.OFF:
await self.async_set_hvac_mode(mode)
else:
raise ServiceValidationError(
f"[{self._eq3_config.mac_address}] Can't change HVAC mode to off while changing temperature",
)
temperature: float | None
if (temperature := kwargs.get(ATTR_TEMPERATURE)) is None:
return
previous_temperature = self._target_temperature
self._target_temperature = temperature
self.async_write_ha_state()
try:
await self._thermostat.async_set_temperature(self._target_temperature)
except Eq3Exception:
_LOGGER.error(
"[%s] Failed setting temperature", self._eq3_config.mac_address
)
self._target_temperature = previous_temperature
self.async_write_ha_state()
except ValueError as ex:
raise ServiceValidationError("Invalid temperature") from ex
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set new target hvac mode."""
if hvac_mode is HVACMode.OFF:
await self.async_set_temperature(temperature=EQ3BT_OFF_TEMP)
try:
await self._thermostat.async_set_mode(HA_TO_EQ_HVAC[hvac_mode])
except Eq3Exception:
_LOGGER.error("[%s] Failed setting HVAC mode", self._eq3_config.mac_address)
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set new preset mode."""
match preset_mode:
case Preset.BOOST:
await self._thermostat.async_set_boost(True)
case Preset.AWAY:
await self._thermostat.async_set_away(True)
case Preset.ECO:
await self._thermostat.async_set_preset(Eq3Preset.ECO)
case Preset.COMFORT:
await self._thermostat.async_set_preset(Eq3Preset.COMFORT)
case Preset.OPEN:
await self._thermostat.async_set_mode(OperationMode.ON)

96
homeassistant/components/eq3btsmart/config_flow.py Normal file
View File

@ -0,0 +1,96 @@
"""Config flow for eQ-3 Bluetooth Smart thermostats."""
from typing import Any
from homeassistant import config_entries
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from homeassistant.config_entries import ConfigFlowResult
from homeassistant.const import CONF_MAC
from homeassistant.helpers.device_registry import format_mac
from homeassistant.util import slugify
from .const import DOMAIN
from .schemas import SCHEMA_MAC
class EQ3ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Config flow for eQ-3 Bluetooth Smart thermostats."""
def __init__(self) -> None:
"""Initialize the config flow."""
self.mac_address: str = ""
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a flow initialized by the user."""
errors: dict[str, str] = {}
if user_input is None:
return self.async_show_form(
step_id="user",
data_schema=SCHEMA_MAC,
errors=errors,
)
mac_address = format_mac(user_input[CONF_MAC])
if not validate_mac(mac_address):
errors[CONF_MAC] = "invalid_mac_address"
return self.async_show_form(
step_id="user",
data_schema=SCHEMA_MAC,
errors=errors,
)
await self.async_set_unique_id(mac_address)
self._abort_if_unique_id_configured(updates=user_input)
# We can not validate if this mac actually is an eQ-3 thermostat,
# since the thermostat probably is not advertising right now.
return self.async_create_entry(title=slugify(mac_address), data={})
async def async_step_bluetooth(
self, discovery_info: BluetoothServiceInfoBleak
) -> ConfigFlowResult:
"""Handle bluetooth discovery."""
self.mac_address = format_mac(discovery_info.address)
await self.async_set_unique_id(self.mac_address)
self._abort_if_unique_id_configured()
self.context.update({"title_placeholders": {CONF_MAC: self.mac_address}})
return await self.async_step_init()
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle flow start."""
if user_input is None:
return self.async_show_form(
step_id="init",
description_placeholders={CONF_MAC: self.mac_address},
)
await self.async_set_unique_id(self.mac_address)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=slugify(self.mac_address),
data={},
)
def validate_mac(mac: str) -> bool:
"""Return whether or not given value is a valid MAC address."""
return bool(
mac
and len(mac) == 17
and mac.count(":") == 5
and all(int(part, 16) < 256 for part in mac.split(":") if part)
)

73
homeassistant/components/eq3btsmart/const.py Normal file
View File

@ -0,0 +1,73 @@
"""Constants for EQ3 Bluetooth Smart Radiator Valves."""
from enum import Enum
from eq3btsmart.const import OperationMode
from homeassistant.components.climate import (
PRESET_AWAY,
PRESET_BOOST,
PRESET_COMFORT,
PRESET_ECO,
PRESET_NONE,
HVACMode,
)
DOMAIN = "eq3btsmart"
MANUFACTURER = "eQ-3 AG"
DEVICE_MODEL = "CC-RT-BLE-EQ"
GET_DEVICE_TIMEOUT = 5 # seconds
EQ_TO_HA_HVAC: dict[OperationMode, HVACMode] = {
OperationMode.OFF: HVACMode.OFF,
OperationMode.ON: HVACMode.HEAT,
OperationMode.AUTO: HVACMode.AUTO,
OperationMode.MANUAL: HVACMode.HEAT,
}
HA_TO_EQ_HVAC = {
HVACMode.OFF: OperationMode.OFF,
HVACMode.AUTO: OperationMode.AUTO,
HVACMode.HEAT: OperationMode.MANUAL,
}
class Preset(str, Enum):
"""Preset modes for the eQ-3 radiator valve."""
NONE = PRESET_NONE
ECO = PRESET_ECO
COMFORT = PRESET_COMFORT
BOOST = PRESET_BOOST
AWAY = PRESET_AWAY
OPEN = "Open"
LOW_BATTERY = "Low Battery"
WINDOW_OPEN = "Window"
class CurrentTemperatureSelector(str, Enum):
"""Selector for current temperature."""
NOTHING = "NOTHING"
UI = "UI"
DEVICE = "DEVICE"
VALVE = "VALVE"
ENTITY = "ENTITY"
class TargetTemperatureSelector(str, Enum):
"""Selector for target temperature."""
TARGET = "TARGET"
LAST_REPORTED = "LAST_REPORTED"
DEFAULT_CURRENT_TEMP_SELECTOR = CurrentTemperatureSelector.DEVICE
DEFAULT_TARGET_TEMP_SELECTOR = TargetTemperatureSelector.TARGET
DEFAULT_SCAN_INTERVAL = 10 # seconds
SIGNAL_THERMOSTAT_DISCONNECTED = f"{DOMAIN}.thermostat_disconnected"
SIGNAL_THERMOSTAT_CONNECTED = f"{DOMAIN}.thermostat_connected"

19
homeassistant/components/eq3btsmart/entity.py Normal file
View File

@ -0,0 +1,19 @@
"""Base class for all eQ-3 entities."""
from eq3btsmart.thermostat import Thermostat
from homeassistant.helpers.entity import Entity
from .models import Eq3Config
class Eq3Entity(Entity):
"""Base class for all eQ-3 entities."""
_attr_has_entity_name = True
def __init__(self, eq3_config: Eq3Config, thermostat: Thermostat) -> None:
"""Initialize the eq3 entity."""
self._eq3_config = eq3_config
self._thermostat = thermostat

27
homeassistant/components/eq3btsmart/manifest.json Normal file
View File

@ -0,0 +1,27 @@
{
"domain": "eq3btsmart",
"name": "eQ-3 Bluetooth Smart Thermostats",
"bluetooth": [
{
"local_name": "CC-RT-BLE",
"connectable": true
},
{
"local_name": "CC-RT-M-BLE",
"connectable": true
},
{
"local_name": "CC-RT-BLE-EQ",
"connectable": true
}
],
"codeowners": ["@eulemitkeule", "@dbuezas"],
"config_flow": true,
"dependencies": ["bluetooth", "bluetooth_adapters"],
"documentation": "https://www.home-assistant.io/integrations/eq3btsmart",
"integration_type": "device",
"iot_class": "local_polling",
"loggers": ["eq3btsmart"],
"quality_scale": "silver",
"requirements": ["eq3btsmart==1.1.6", "bleak-esphome==1.0.0"]
}

35
homeassistant/components/eq3btsmart/models.py Normal file
View File

@ -0,0 +1,35 @@
"""Models for eq3btsmart integration."""
from dataclasses import dataclass
from eq3btsmart.const import DEFAULT_AWAY_HOURS, DEFAULT_AWAY_TEMP
from eq3btsmart.thermostat import Thermostat
from .const import (
DEFAULT_CURRENT_TEMP_SELECTOR,
DEFAULT_SCAN_INTERVAL,
DEFAULT_TARGET_TEMP_SELECTOR,
CurrentTemperatureSelector,
TargetTemperatureSelector,
)
@dataclass(slots=True)
class Eq3Config:
"""Config for a single eQ-3 device."""
mac_address: str
current_temp_selector: CurrentTemperatureSelector = DEFAULT_CURRENT_TEMP_SELECTOR
target_temp_selector: TargetTemperatureSelector = DEFAULT_TARGET_TEMP_SELECTOR
external_temp_sensor: str = ""
scan_interval: int = DEFAULT_SCAN_INTERVAL
default_away_hours: float = DEFAULT_AWAY_HOURS
default_away_temperature: float = DEFAULT_AWAY_TEMP
@dataclass(slots=True)
class Eq3ConfigEntryData:
"""Config entry for a single eQ-3 device."""
eq3_config: Eq3Config
thermostat: Thermostat

15
homeassistant/components/eq3btsmart/schemas.py Normal file
View File

@ -0,0 +1,15 @@
"""Voluptuous schemas for eq3btsmart."""
from eq3btsmart.const import EQ3BT_MAX_TEMP, EQ3BT_MIN_TEMP
import voluptuous as vol
from homeassistant.const import CONF_MAC
from homeassistant.helpers import config_validation as cv
SCHEMA_TEMPERATURE = vol.Range(min=EQ3BT_MIN_TEMP, max=EQ3BT_MAX_TEMP)
SCHEMA_DEVICE = vol.Schema({vol.Required(CONF_MAC): cv.string})
SCHEMA_MAC = vol.Schema(
{
vol.Required(CONF_MAC): str,
}
)

19
homeassistant/components/eq3btsmart/strings.json Normal file
View File

@ -0,0 +1,19 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
},
"flow_title": "eQ-3 Device [{mac}]",
"step": {
"user": {
"title": "Configure new eQ-3 device",
"data": {
"mac": "MAC address"
}
},
"init": {
"title": "Configure new eQ-3 device"
}
}
}
}

15
homeassistant/generated/bluetooth.py
View File

@ -66,6 +66,21 @@ BLUETOOTH: list[dict[str, bool | str | int | list[int]]] = [
"domain": "dormakaba_dkey",
"service_uuid": "e7a60001-6639-429f-94fd-86de8ea26897",
},
{
"connectable": True,
"domain": "eq3btsmart",
"local_name": "CC-RT-BLE",
},
{
"connectable": True,
"domain": "eq3btsmart",
"local_name": "CC-RT-M-BLE",
},
{
"connectable": True,
"domain": "eq3btsmart",
"local_name": "CC-RT-BLE-EQ",
},
{
"domain": "eufylife_ble",
"local_name": "eufy T9140",

1
homeassistant/generated/config_flows.py
View File

@ -150,6 +150,7 @@ FLOWS = {
"environment_canada",
"epion",
"epson",
"eq3btsmart",
"escea",
"esphome",
"eufylife_ble",

6
homeassistant/generated/integrations.json
View File

@ -1660,6 +1660,12 @@
"config_flow": false,
"iot_class": "local_polling",
"name": "eQ-3 MAX!"
},
"eq3btsmart": {
"integration_type": "device",
"config_flow": true,
"iot_class": "local_polling",
"name": "eQ-3 Bluetooth Smart Thermostats"
}
}
},

10
mypy.ini
View File

@ -1461,6 +1461,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.eq3btsmart.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.esphome.*]
check_untyped_defs = true
disallow_incomplete_defs = true

4
requirements_all.txt
View File

@ -552,6 +552,7 @@ bimmer-connected[china]==0.14.6
# homeassistant.components.bizkaibus
bizkaibus==0.1.1
# homeassistant.components.eq3btsmart
# homeassistant.components.esphome
bleak-esphome==1.0.0
@ -820,6 +821,9 @@ epson-projector==0.5.1
# homeassistant.components.epsonworkforce
epsonprinter==0.0.9
# homeassistant.components.eq3btsmart
eq3btsmart==1.1.6
# homeassistant.components.esphome
esphome-dashboard-api==1.2.3

4
requirements_test_all.txt
View File

@ -474,6 +474,7 @@ bellows==0.38.1
# homeassistant.components.bmw_connected_drive
bimmer-connected[china]==0.14.6
# homeassistant.components.eq3btsmart
# homeassistant.components.esphome
bleak-esphome==1.0.0
@ -671,6 +672,9 @@ epion==0.0.3
# homeassistant.components.epson
epson-projector==0.5.1
# homeassistant.components.eq3btsmart
eq3btsmart==1.1.6
# homeassistant.components.esphome
esphome-dashboard-api==1.2.3

1
tests/components/eq3btsmart/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the eq3btsmart component."""

41
tests/components/eq3btsmart/conftest.py Normal file
View File

@ -0,0 +1,41 @@
"""Fixtures for eq3btsmart tests."""
from bleak.backends.scanner import AdvertisementData
import pytest
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from .const import MAC
from tests.components.bluetooth import generate_ble_device
@pytest.fixture(autouse=True)
def mock_bluetooth(enable_bluetooth):
"""Auto mock bluetooth."""
@pytest.fixture
def fake_service_info():
"""Return a BluetoothServiceInfoBleak for use in testing."""
return BluetoothServiceInfoBleak(
name="CC-RT-BLE",
address=MAC,
rssi=0,
manufacturer_data={},
service_data={},
service_uuids=[],
source="local",
connectable=False,
time=0,
device=generate_ble_device(address=MAC, name="CC-RT-BLE", rssi=0),
advertisement=AdvertisementData(
local_name="CC-RT-BLE",
manufacturer_data={},
service_data={},
service_uuids=[],
rssi=0,
tx_power=-127,
platform_data=(),
),
)

4
tests/components/eq3btsmart/const.py Normal file
View File

@ -0,0 +1,4 @@
"""Constants for the eq3btsmart tests."""
MAC = "aa:bb:cc:dd:ee:ff"
RSSI = -60

135
tests/components/eq3btsmart/test_config_flow.py Normal file
View File

@ -0,0 +1,135 @@
"""Test the eq3btsmart config flow."""
from unittest.mock import patch
from homeassistant import config_entries
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from homeassistant.components.eq3btsmart.const import DOMAIN
from homeassistant.const import CONF_MAC
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.device_registry import format_mac
from homeassistant.util import slugify
from .const import MAC
from tests.common import MockConfigEntry
async def test_user_flow(hass: HomeAssistant) -> None:
"""Test we can handle a regular successflow setup flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.eq3btsmart.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_MAC: MAC},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == slugify(MAC)
assert result["data"] == {}
assert result["context"]["unique_id"] == MAC
assert len(mock_setup_entry.mock_calls) == 1
async def test_user_flow_invalid_mac(hass: HomeAssistant) -> None:
"""Test we handle invalid mac address."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.eq3btsmart.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_MAC: "invalid"},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {CONF_MAC: "invalid_mac_address"}
assert len(mock_setup_entry.mock_calls) == 0
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_MAC: MAC},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == slugify(MAC)
assert result["data"] == {}
assert result["context"]["unique_id"] == MAC
assert len(mock_setup_entry.mock_calls) == 1
async def test_bluetooth_flow(
hass: HomeAssistant, fake_service_info: BluetoothServiceInfoBleak
) -> None:
"""Test we can handle a bluetooth discovery flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=fake_service_info,
)
with patch(
"homeassistant.components.eq3btsmart.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == slugify(MAC)
assert result["data"] == {}
assert result["context"]["unique_id"] == MAC
assert len(mock_setup_entry.mock_calls) == 1
async def test_duplicate_entry(hass: HomeAssistant) -> None:
"""Test duplicate setup handling."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_MAC: MAC,
},
unique_id=format_mac(MAC),
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.eq3btsmart.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_MAC: MAC,
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert mock_setup_entry.call_count == 0