Move TP-Link power and energy switch attributes to sensors (#53596)

Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
This commit is contained in:
Michael 2021-07-29 20:02:47 +02:00 committed by GitHub
parent 24a589961a
commit bedb9550f5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 634 additions and 196 deletions

View File

@ -1,37 +1,57 @@
"""Component to embed TP-Link smart home devices."""
import logging
from __future__ import annotations
from datetime import datetime, timedelta
import logging
import time
from pyHS100.smartdevice import SmartDevice, SmartDeviceException
from pyHS100.smartplug import SmartPlug
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.sensor import ATTR_LAST_RESET
from homeassistant.components.switch import ATTR_CURRENT_POWER_W, ATTR_TODAY_ENERGY_KWH
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.const import (
ATTR_VOLTAGE,
CONF_ALIAS,
CONF_DEVICE_ID,
CONF_HOST,
CONF_MAC,
CONF_STATE,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from homeassistant.util.dt import utc_from_timestamp
from .common import (
from .common import SmartDevices, async_discover_devices, get_static_devices
from .const import (
ATTR_CONFIG,
ATTR_CURRENT_A,
ATTR_TOTAL_ENERGY_KWH,
CONF_DIMMER,
CONF_DISCOVERY,
CONF_EMETER_PARAMS,
CONF_LIGHT,
CONF_MODEL,
CONF_STRIP,
CONF_SW_VERSION,
CONF_SWITCH,
SmartDevices,
async_discover_devices,
get_static_devices,
COORDINATORS,
PLATFORMS,
)
_LOGGER = logging.getLogger(__name__)
DOMAIN = "tplink"
PLATFORMS = [CONF_LIGHT, CONF_SWITCH]
TPLINK_HOST_SCHEMA = vol.Schema({vol.Required(CONF_HOST): cv.string})
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
@ -82,8 +102,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
device_count = len(tplink_devices)
# These will contain the initialized devices
lights = hass.data[DOMAIN][CONF_LIGHT] = []
switches = hass.data[DOMAIN][CONF_SWITCH] = []
hass.data[DOMAIN][CONF_LIGHT] = []
hass.data[DOMAIN][CONF_SWITCH] = []
lights: list[SmartDevice] = hass.data[DOMAIN][CONF_LIGHT]
switches: list[SmartPlug] = hass.data[DOMAIN][CONF_SWITCH]
# Add static devices
static_devices = SmartDevices()
@ -102,14 +124,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
lights.extend(discovered_devices.lights)
switches.extend(discovered_devices.switches)
forward_setup = hass.config_entries.async_forward_entry_setup
if lights:
_LOGGER.debug(
"Got %s lights: %s", len(lights), ", ".join(d.host for d in lights)
)
hass.async_create_task(forward_setup(entry, "light"))
if switches:
_LOGGER.debug(
"Got %s switches: %s",
@ -117,7 +136,23 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
", ".join(d.host for d in switches),
)
hass.async_create_task(forward_setup(entry, "switch"))
# prepare DataUpdateCoordinators
hass.data[DOMAIN][COORDINATORS] = {}
for switch in switches:
try:
await hass.async_add_executor_job(switch.get_sysinfo)
except SmartDeviceException as ex:
_LOGGER.debug(ex)
raise ConfigEntryNotReady from ex
hass.data[DOMAIN][COORDINATORS][
switch.mac
] = coordinator = SmartPlugDataUpdateCoordinator(hass, switch)
await coordinator.async_config_entry_first_refresh()
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
@ -130,3 +165,65 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
hass.data[DOMAIN].clear()
return unload_ok
class SmartPlugDataUpdateCoordinator(DataUpdateCoordinator):
"""DataUpdateCoordinator to gather data for specific SmartPlug."""
def __init__(
self,
hass: HomeAssistant,
smartplug: SmartPlug,
) -> None:
"""Initialize DataUpdateCoordinator to gather data for specific SmartPlug."""
self.smartplug = smartplug
update_interval = timedelta(seconds=30)
super().__init__(
hass, _LOGGER, name=smartplug.alias, update_interval=update_interval
)
async def _async_update_data(self) -> dict:
"""Fetch all device and sensor data from api."""
info = self.smartplug.sys_info
data = {
CONF_HOST: self.smartplug.host,
CONF_MAC: info["mac"],
CONF_MODEL: info["model"],
CONF_SW_VERSION: info["sw_ver"],
}
if self.smartplug.context is None:
data[CONF_ALIAS] = info["alias"]
data[CONF_DEVICE_ID] = info["mac"]
data[CONF_STATE] = self.smartplug.state == self.smartplug.SWITCH_STATE_ON
else:
plug_from_context = next(
c
for c in self.smartplug.sys_info["children"]
if c["id"] == self.smartplug.context
)
data[CONF_ALIAS] = plug_from_context["alias"]
data[CONF_DEVICE_ID] = self.smartplug.context
data[CONF_STATE] = plug_from_context["state"] == 1
if self.smartplug.has_emeter:
emeter_readings = self.smartplug.get_emeter_realtime()
data[CONF_EMETER_PARAMS] = {
ATTR_CURRENT_POWER_W: round(float(emeter_readings["power"]), 2),
ATTR_TOTAL_ENERGY_KWH: round(float(emeter_readings["total"]), 3),
ATTR_VOLTAGE: round(float(emeter_readings["voltage"]), 1),
ATTR_CURRENT_A: round(float(emeter_readings["current"]), 2),
ATTR_LAST_RESET: {ATTR_TOTAL_ENERGY_KWH: utc_from_timestamp(0)},
}
emeter_statics = self.smartplug.get_emeter_daily()
data[CONF_EMETER_PARAMS][ATTR_LAST_RESET][
ATTR_TODAY_ENERGY_KWH
] = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
if emeter_statics.get(int(time.strftime("%e"))):
data[CONF_EMETER_PARAMS][ATTR_TODAY_ENERGY_KWH] = round(
float(emeter_statics[int(time.strftime("%e"))]), 3
)
else:
# today's consumption not available, when device was off all the day
data[CONF_EMETER_PARAMS][ATTR_TODAY_ENERGY_KWH] = 0.0
return data

View File

@ -14,21 +14,20 @@ from pyHS100 import (
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from .const import DOMAIN as TPLINK_DOMAIN
from .const import (
CONF_DIMMER,
CONF_LIGHT,
CONF_STRIP,
CONF_SWITCH,
DOMAIN as TPLINK_DOMAIN,
MAX_DISCOVERY_RETRIES,
)
_LOGGER = logging.getLogger(__name__)
ATTR_CONFIG = "config"
CONF_DIMMER = "dimmer"
CONF_DISCOVERY = "discovery"
CONF_LIGHT = "light"
CONF_STRIP = "strip"
CONF_SWITCH = "switch"
MAX_DISCOVERY_RETRIES = 4
class SmartDevices:
"""Hold different kinds of devices."""
@ -98,7 +97,7 @@ async def async_discover_devices(
else:
_LOGGER.error("Unknown smart device type: %s", type(dev))
devices = {}
devices: dict[str, SmartDevice] = {}
for attempt in range(1, MAX_DISCOVERY_RETRIES + 1):
_LOGGER.debug(
"Discovering tplink devices, attempt %s of %s",
@ -159,16 +158,18 @@ def get_static_devices(config_data) -> SmartDevices:
def add_available_devices(
hass: HomeAssistant, device_type: str, device_class: Callable
) -> list:
) -> list[Entity]:
"""Get sysinfo for all devices."""
devices = hass.data[TPLINK_DOMAIN][device_type]
devices: list[SmartDevice] = hass.data[TPLINK_DOMAIN][device_type]
if f"{device_type}_remaining" in hass.data[TPLINK_DOMAIN]:
devices = hass.data[TPLINK_DOMAIN][f"{device_type}_remaining"]
devices: list[SmartDevice] = hass.data[TPLINK_DOMAIN][
f"{device_type}_remaining"
]
entities_ready = []
devices_unavailable = []
entities_ready: list[Entity] = []
devices_unavailable: list[SmartDevice] = []
for device in devices:
try:
device.get_sysinfo()

View File

@ -1,5 +1,81 @@
"""Const for TP-Link."""
from __future__ import annotations
import datetime
from homeassistant.components.sensor import (
STATE_CLASS_MEASUREMENT,
SensorEntityDescription,
)
from homeassistant.components.switch import ATTR_CURRENT_POWER_W, ATTR_TODAY_ENERGY_KWH
from homeassistant.const import (
ATTR_VOLTAGE,
DEVICE_CLASS_CURRENT,
DEVICE_CLASS_ENERGY,
DEVICE_CLASS_POWER,
DEVICE_CLASS_VOLTAGE,
ELECTRIC_CURRENT_AMPERE,
ELECTRIC_POTENTIAL_VOLT,
ENERGY_KILO_WATT_HOUR,
POWER_WATT,
)
DOMAIN = "tplink"
COORDINATORS = "coordinators"
MIN_TIME_BETWEEN_UPDATES = datetime.timedelta(seconds=8)
MAX_DISCOVERY_RETRIES = 4
ATTR_CONFIG = "config"
ATTR_TOTAL_ENERGY_KWH = "total_energy_kwh"
ATTR_CURRENT_A = "current_a"
CONF_MODEL = "model"
CONF_SW_VERSION = "sw_ver"
CONF_EMETER_PARAMS = "emeter_params"
CONF_DIMMER = "dimmer"
CONF_DISCOVERY = "discovery"
CONF_LIGHT = "light"
CONF_STRIP = "strip"
CONF_SWITCH = "switch"
CONF_SENSOR = "sensor"
PLATFORMS = [CONF_LIGHT, CONF_SENSOR, CONF_SWITCH]
ENERGY_SENSORS: list[SensorEntityDescription] = [
SensorEntityDescription(
key=ATTR_CURRENT_POWER_W,
unit_of_measurement=POWER_WATT,
device_class=DEVICE_CLASS_POWER,
state_class=STATE_CLASS_MEASUREMENT,
name="Current Consumption",
),
SensorEntityDescription(
key=ATTR_TOTAL_ENERGY_KWH,
unit_of_measurement=ENERGY_KILO_WATT_HOUR,
device_class=DEVICE_CLASS_ENERGY,
state_class=STATE_CLASS_MEASUREMENT,
name="Total Consumption",
),
SensorEntityDescription(
key=ATTR_TODAY_ENERGY_KWH,
unit_of_measurement=ENERGY_KILO_WATT_HOUR,
device_class=DEVICE_CLASS_ENERGY,
state_class=STATE_CLASS_MEASUREMENT,
name="Today's Consumption",
),
SensorEntityDescription(
key=ATTR_VOLTAGE,
unit_of_measurement=ELECTRIC_POTENTIAL_VOLT,
device_class=DEVICE_CLASS_VOLTAGE,
state_class=STATE_CLASS_MEASUREMENT,
name="Voltage",
),
SensorEntityDescription(
key=ATTR_CURRENT_A,
unit_of_measurement=ELECTRIC_CURRENT_AMPERE,
device_class=DEVICE_CLASS_CURRENT,
state_class=STATE_CLASS_MEASUREMENT,
name="Current",
),
]

View File

@ -0,0 +1,100 @@
"""Support for TPLink HS100/HS110/HS200 smart switch energy sensors."""
from __future__ import annotations
from typing import Any
from pyHS100 import SmartPlug
from homeassistant.components.sensor import (
ATTR_LAST_RESET,
SensorEntity,
SensorEntityDescription,
)
from homeassistant.components.tplink import SmartPlugDataUpdateCoordinator
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ALIAS, CONF_DEVICE_ID, CONF_MAC
from homeassistant.core import HomeAssistant
import homeassistant.helpers.device_registry as dr
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import (
CONF_EMETER_PARAMS,
CONF_MODEL,
CONF_SW_VERSION,
CONF_SWITCH,
COORDINATORS,
DOMAIN as TPLINK_DOMAIN,
ENERGY_SENSORS,
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up switches."""
entities: list[SmartPlugSensor] = []
coordinators: list[SmartPlugDataUpdateCoordinator] = hass.data[TPLINK_DOMAIN][
COORDINATORS
]
switches: list[SmartPlug] = hass.data[TPLINK_DOMAIN][CONF_SWITCH]
for switch in switches:
coordinator: SmartPlugDataUpdateCoordinator = coordinators[switch.mac]
if not switch.has_emeter and coordinator.data.get(CONF_EMETER_PARAMS) is None:
continue
for description in ENERGY_SENSORS:
if coordinator.data[CONF_EMETER_PARAMS].get(description.key) is not None:
entities.append(SmartPlugSensor(switch, coordinator, description))
async_add_entities(entities)
class SmartPlugSensor(CoordinatorEntity, SensorEntity):
"""Representation of a TPLink Smart Plug energy sensor."""
def __init__(
self,
smartplug: SmartPlug,
coordinator: DataUpdateCoordinator,
description: SensorEntityDescription,
) -> None:
"""Initialize the switch."""
super().__init__(coordinator)
self.smartplug = smartplug
self.entity_description = description
self._attr_name = f"{coordinator.data[CONF_ALIAS]} {description.name}"
self._attr_last_reset = coordinator.data[CONF_EMETER_PARAMS][
ATTR_LAST_RESET
].get(description.key)
@property
def data(self) -> dict[str, Any]:
"""Return data from DataUpdateCoordinator."""
return self.coordinator.data
@property
def state(self) -> float | None:
"""Return the sensors state."""
return self.data[CONF_EMETER_PARAMS][self.entity_description.key]
@property
def unique_id(self) -> str | None:
"""Return a unique ID."""
return f"{self.data[CONF_DEVICE_ID]}_{self.entity_description.key}"
@property
def device_info(self) -> DeviceInfo:
"""Return information about the device."""
return {
"name": self.data[CONF_ALIAS],
"model": self.data[CONF_MODEL],
"manufacturer": "TP-Link",
"connections": {(dr.CONNECTION_NETWORK_MAC, self.data[CONF_MAC])},
"sw_version": self.data[CONF_SW_VERSION],
}

View File

@ -1,40 +1,30 @@
"""Support for TPLink HS100/HS110/HS200 smart switch."""
from __future__ import annotations
import asyncio
from collections.abc import Mapping
from contextlib import suppress
import logging
import time
from typing import Any
from pyHS100 import SmartDeviceException, SmartPlug
from pyHS100 import SmartPlug
from homeassistant.components.switch import (
ATTR_CURRENT_POWER_W,
ATTR_TODAY_ENERGY_KWH,
SwitchEntity,
)
from homeassistant.components.switch import SwitchEntity
from homeassistant.components.tplink import SmartPlugDataUpdateCoordinator
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_VOLTAGE
from homeassistant.const import CONF_ALIAS, CONF_DEVICE_ID, CONF_MAC, CONF_STATE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady
import homeassistant.helpers.device_registry as dr
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from . import CONF_SWITCH, DOMAIN as TPLINK_DOMAIN
from .common import add_available_devices
PARALLEL_UPDATES = 0
_LOGGER = logging.getLogger(__name__)
ATTR_TOTAL_ENERGY_KWH = "total_energy_kwh"
ATTR_CURRENT_A = "current_a"
MAX_ATTEMPTS = 300
SLEEP_TIME = 2
from .const import (
CONF_MODEL,
CONF_SW_VERSION,
CONF_SWITCH,
COORDINATORS,
DOMAIN as TPLINK_DOMAIN,
)
async def async_setup_entry(
@ -43,164 +33,65 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up switches."""
entities = await hass.async_add_executor_job(
add_available_devices, hass, CONF_SWITCH, SmartPlugSwitch
)
entities: list[SmartPlugSwitch] = []
coordinators: list[SmartPlugDataUpdateCoordinator] = hass.data[TPLINK_DOMAIN][
COORDINATORS
]
switches: list[SmartPlug] = hass.data[TPLINK_DOMAIN][CONF_SWITCH]
for switch in switches:
coordinator = coordinators[switch.mac]
entities.append(SmartPlugSwitch(switch, coordinator))
if entities:
async_add_entities(entities, update_before_add=True)
if hass.data[TPLINK_DOMAIN][f"{CONF_SWITCH}_remaining"]:
raise PlatformNotReady
async_add_entities(entities)
class SmartPlugSwitch(SwitchEntity):
class SmartPlugSwitch(CoordinatorEntity, SwitchEntity):
"""Representation of a TPLink Smart Plug switch."""
def __init__(self, smartplug: SmartPlug) -> None:
def __init__(
self, smartplug: SmartPlug, coordinator: DataUpdateCoordinator
) -> None:
"""Initialize the switch."""
super().__init__(coordinator)
self.smartplug = smartplug
self._sysinfo = None
self._state = None
self._is_available = False
# Set up emeter cache
self._emeter_params = {}
self._mac = None
self._alias = None
self._model = None
self._device_id = None
self._host = None
@property
def data(self) -> dict[str, Any]:
"""Return data from DataUpdateCoordinator."""
return self.coordinator.data
@property
def unique_id(self) -> str | None:
"""Return a unique ID."""
return self._device_id
return self.data[CONF_DEVICE_ID]
@property
def name(self) -> str | None:
"""Return the name of the Smart Plug."""
return self._alias
return self.data[CONF_ALIAS]
@property
def device_info(self) -> DeviceInfo:
"""Return information about the device."""
return {
"name": self._alias,
"model": self._model,
"name": self.data[CONF_ALIAS],
"model": self.data[CONF_MODEL],
"manufacturer": "TP-Link",
"connections": {(dr.CONNECTION_NETWORK_MAC, self._mac)},
"sw_version": self._sysinfo["sw_ver"],
"connections": {(dr.CONNECTION_NETWORK_MAC, self.data[CONF_MAC])},
"sw_version": self.data[CONF_SW_VERSION],
}
@property
def available(self) -> bool:
"""Return if switch is available."""
return self._is_available
@property
def is_on(self) -> bool | None:
"""Return true if switch is on."""
return self._state
return self.data[CONF_STATE]
def turn_on(self, **kwargs: Any) -> None:
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
self.smartplug.turn_on()
await self.hass.async_add_job(self.smartplug.turn_on)
await self.coordinator.async_refresh()
def turn_off(self, **kwargs: Any) -> None:
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off."""
self.smartplug.turn_off()
@property
def extra_state_attributes(self) -> Mapping[str, Any] | None:
"""Return the state attributes of the device."""
return self._emeter_params
@property
def _plug_from_context(self) -> Any:
"""Return the plug from the context."""
children = self.smartplug.sys_info["children"]
return next(c for c in children if c["id"] == self.smartplug.context)
def update_state(self) -> None:
"""Update the TP-Link switch's state."""
if self.smartplug.context is None:
self._state = self.smartplug.state == self.smartplug.SWITCH_STATE_ON
else:
self._state = self._plug_from_context["state"] == 1
def attempt_update(self, update_attempt: int) -> bool:
"""Attempt to get details from the TP-Link switch."""
try:
if not self._sysinfo:
self._sysinfo = self.smartplug.sys_info
self._mac = self._sysinfo["mac"]
self._model = self._sysinfo["model"]
self._host = self.smartplug.host
if self.smartplug.context is None:
self._alias = self._sysinfo["alias"]
self._device_id = self._mac
else:
self._alias = self._plug_from_context["alias"]
self._device_id = self.smartplug.context
self.update_state()
if self.smartplug.has_emeter:
emeter_readings = self.smartplug.get_emeter_realtime()
self._emeter_params[ATTR_CURRENT_POWER_W] = round(
float(emeter_readings["power"]), 2
)
self._emeter_params[ATTR_TOTAL_ENERGY_KWH] = round(
float(emeter_readings["total"]), 3
)
self._emeter_params[ATTR_VOLTAGE] = round(
float(emeter_readings["voltage"]), 1
)
self._emeter_params[ATTR_CURRENT_A] = round(
float(emeter_readings["current"]), 2
)
emeter_statics = self.smartplug.get_emeter_daily()
with suppress(KeyError): # Device returned no daily history
self._emeter_params[ATTR_TODAY_ENERGY_KWH] = round(
float(emeter_statics[int(time.strftime("%e"))]), 3
)
return True
except (SmartDeviceException, OSError) as ex:
if update_attempt == 0:
_LOGGER.debug(
"Retrying in %s seconds for %s|%s due to: %s",
SLEEP_TIME,
self._host,
self._alias,
ex,
)
return False
async def async_update(self) -> None:
"""Update the TP-Link switch's state."""
for update_attempt in range(MAX_ATTEMPTS):
is_ready = await self.hass.async_add_executor_job(
self.attempt_update, update_attempt
)
if is_ready:
self._is_available = True
if update_attempt > 0:
_LOGGER.debug(
"Device %s|%s responded after %s attempts",
self._host,
self._alias,
update_attempt,
)
break
await asyncio.sleep(SLEEP_TIME)
else:
if self._is_available:
_LOGGER.warning(
"Could not read state for %s|%s", self.smartplug.host, self._alias
)
self._is_available = False
await self.hass.async_add_job(self.smartplug.turn_off)
await self.coordinator.async_refresh()

View File

@ -0,0 +1,72 @@
"""Constants for the TP-Link component tests."""
SMARTPLUGSWITCH_DATA = {
"sysinfo": {
"sw_ver": "1.0.4 Build 191111 Rel.143500",
"hw_ver": "4.0",
"model": "HS110(EU)",
"deviceId": "4C56447B395BB7A2FAC68C9DFEE2E84163222581",
"oemId": "40F54B43071E9436B6395611E9D91CEA",
"hwId": "A6C77E4FDD238B53D824AC8DA361F043",
"rssi": -24,
"longitude_i": 130793,
"latitude_i": 480582,
"alias": "SmartPlug",
"status": "new",
"mic_type": "IOT.SMARTPLUGSWITCH",
"feature": "TIM:ENE",
"mac": "69:F2:3C:8E:E3:47",
"updating": 0,
"led_off": 0,
"relay_state": 0,
"on_time": 0,
"active_mode": "none",
"icon_hash": "",
"dev_name": "Smart Wi-Fi Plug With Energy Monitoring",
"next_action": {"type": -1},
"err_code": 0,
},
"realtime": {
"voltage_mv": 233957,
"current_ma": 21,
"power_mw": 0,
"total_wh": 1793,
"err_code": 0,
},
}
SMARTSTRIPWITCH_DATA = {
"sysinfo": {
"sw_ver": "1.0.4 Build 191111 Rel.143500",
"hw_ver": "4.0",
"model": "HS110(EU)",
"deviceId": "4C56447B395BB7A2FAC68C9DFEE2E84163222581",
"oemId": "40F54B43071E9436B6395611E9D91CEA",
"hwId": "A6C77E4FDD238B53D824AC8DA361F043",
"rssi": -24,
"longitude_i": 130793,
"latitude_i": 480582,
"alias": "SmartPlug",
"status": "new",
"mic_type": "IOT.SMARTPLUGSWITCH",
"feature": "TIM",
"mac": "69:F2:3C:8E:E3:47",
"updating": 0,
"led_off": 0,
"relay_state": 0,
"on_time": 0,
"active_mode": "none",
"icon_hash": "",
"dev_name": "Smart Wi-Fi Plug With Energy Monitoring",
"next_action": {"type": -1},
"children": [{"id": "1", "state": 1, "alias": "SmartPlug#1"}],
"err_code": 0,
},
"realtime": {
"voltage_mv": 233957,
"current_ma": 21,
"power_mw": 0,
"total_wh": 1793,
"err_code": 0,
},
"context": "1",
}

View File

@ -1,24 +1,45 @@
"""Tests for the TP-Link component."""
from __future__ import annotations
from datetime import datetime
import time
from typing import Any
from unittest.mock import MagicMock, patch
from pyHS100 import SmartBulb, SmartDevice, SmartDeviceException, SmartPlug
from pyHS100 import SmartBulb, SmartDevice, SmartDeviceException, SmartPlug, smartstrip
from pyHS100.smartdevice import EmeterStatus
import pytest
from homeassistant import config_entries, data_entry_flow
from homeassistant.components import tplink
from homeassistant.components.tplink.common import (
from homeassistant.components.sensor import ATTR_LAST_RESET
from homeassistant.components.switch import ATTR_CURRENT_POWER_W, ATTR_TODAY_ENERGY_KWH
from homeassistant.components.tplink.common import SmartDevices
from homeassistant.components.tplink.const import (
ATTR_CURRENT_A,
ATTR_TOTAL_ENERGY_KWH,
CONF_DIMMER,
CONF_DISCOVERY,
CONF_EMETER_PARAMS,
CONF_LIGHT,
CONF_MODEL,
CONF_SW_VERSION,
CONF_SWITCH,
COORDINATORS,
)
from homeassistant.const import CONF_HOST
from homeassistant.const import (
ATTR_VOLTAGE,
CONF_ALIAS,
CONF_DEVICE_ID,
CONF_HOST,
CONF_MAC,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from homeassistant.util.dt import utc_from_timestamp
from tests.common import MockConfigEntry, mock_coro
from tests.components.tplink.consts import SMARTPLUGSWITCH_DATA, SMARTSTRIPWITCH_DATA
async def test_creating_entry_tries_discover(hass):
@ -186,7 +207,7 @@ async def test_configuring_discovery_disabled(hass):
assert mock_setup.call_count == 1
async def test_platforms_are_initialized(hass):
async def test_platforms_are_initialized(hass: HomeAssistant):
"""Test that platforms are initialized per configuration array."""
config = {
tplink.DOMAIN: {
@ -199,6 +220,8 @@ async def test_platforms_are_initialized(hass):
with patch(
"homeassistant.components.tplink.common.Discover.discover"
) as discover, patch(
"homeassistant.components.tplink.get_static_devices"
) as get_static_devices, patch(
"homeassistant.components.tplink.common.SmartDevice._query_helper"
), patch(
"homeassistant.components.tplink.light.async_setup_entry",
@ -209,13 +232,141 @@ async def test_platforms_are_initialized(hass):
) as switch_setup, patch(
"homeassistant.components.tplink.common.SmartPlug.is_dimmable", False
):
light = SmartBulb("123.123.123.123")
switch = SmartPlug("321.321.321.321")
switch.get_sysinfo = MagicMock(return_value=SMARTPLUGSWITCH_DATA["sysinfo"])
switch.get_emeter_realtime = MagicMock(
return_value=EmeterStatus(SMARTPLUGSWITCH_DATA["realtime"])
)
switch.get_emeter_daily = MagicMock(
return_value={int(time.strftime("%e")): 1.123}
)
get_static_devices.return_value = SmartDevices([light], [switch])
# patching is_dimmable is necessray to avoid misdetection as light.
await async_setup_component(hass, tplink.DOMAIN, config)
await hass.async_block_till_done()
assert discover.call_count == 0
assert light_setup.call_count == 1
assert switch_setup.call_count == 1
assert hass.data.get(tplink.DOMAIN)
assert hass.data[tplink.DOMAIN].get(COORDINATORS)
assert hass.data[tplink.DOMAIN][COORDINATORS].get(switch.mac)
assert isinstance(
hass.data[tplink.DOMAIN][COORDINATORS][switch.mac],
tplink.SmartPlugDataUpdateCoordinator,
)
data = hass.data[tplink.DOMAIN][COORDINATORS][switch.mac].data
assert data[CONF_HOST] == switch.host
assert data[CONF_MAC] == switch.sys_info["mac"]
assert data[CONF_MODEL] == switch.sys_info["model"]
assert data[CONF_SW_VERSION] == switch.sys_info["sw_ver"]
assert data[CONF_ALIAS] == switch.sys_info["alias"]
assert data[CONF_DEVICE_ID] == switch.sys_info["mac"]
emeter_readings = switch.get_emeter_realtime()
assert data[CONF_EMETER_PARAMS][ATTR_VOLTAGE] == round(
float(emeter_readings["voltage"]), 1
)
assert data[CONF_EMETER_PARAMS][ATTR_CURRENT_A] == round(
float(emeter_readings["current"]), 2
)
assert data[CONF_EMETER_PARAMS][ATTR_CURRENT_POWER_W] == round(
float(emeter_readings["power"]), 2
)
assert data[CONF_EMETER_PARAMS][ATTR_TOTAL_ENERGY_KWH] == round(
float(emeter_readings["total"]), 3
)
assert data[CONF_EMETER_PARAMS][ATTR_LAST_RESET][
ATTR_TOTAL_ENERGY_KWH
] == utc_from_timestamp(0)
assert data[CONF_EMETER_PARAMS][ATTR_TODAY_ENERGY_KWH] == 1.123
assert data[CONF_EMETER_PARAMS][ATTR_LAST_RESET][
ATTR_TODAY_ENERGY_KWH
] == datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
assert discover.call_count == 0
assert get_static_devices.call_count == 1
assert light_setup.call_count == 1
assert switch_setup.call_count == 1
async def test_smartplug_without_consumption_sensors(hass: HomeAssistant):
"""Test that platforms are initialized per configuration array."""
config = {
tplink.DOMAIN: {
CONF_DISCOVERY: False,
CONF_SWITCH: [{CONF_HOST: "321.321.321.321"}],
}
}
with patch("homeassistant.components.tplink.common.Discover.discover"), patch(
"homeassistant.components.tplink.get_static_devices"
) as get_static_devices, patch(
"homeassistant.components.tplink.common.SmartDevice._query_helper"
), patch(
"homeassistant.components.tplink.light.async_setup_entry",
return_value=mock_coro(True),
), patch(
"homeassistant.components.tplink.switch.async_setup_entry",
return_value=mock_coro(True),
), patch(
"homeassistant.components.tplink.sensor.SmartPlugSensor.__init__"
) as SmartPlugSensor, patch(
"homeassistant.components.tplink.common.SmartPlug.is_dimmable", False
):
switch = SmartPlug("321.321.321.321")
switch.get_sysinfo = MagicMock(return_value=SMARTPLUGSWITCH_DATA["sysinfo"])
get_static_devices.return_value = SmartDevices([], [switch])
await async_setup_component(hass, tplink.DOMAIN, config)
await hass.async_block_till_done()
assert SmartPlugSensor.call_count == 0
async def test_smartstrip_device(hass: HomeAssistant):
"""Test discover a SmartStrip devices."""
config = {
tplink.DOMAIN: {
CONF_DISCOVERY: True,
}
}
class SmartStrip(smartstrip.SmartStrip):
"""Moked SmartStrip class."""
def get_sysinfo(self):
return SMARTSTRIPWITCH_DATA["sysinfo"]
with patch(
"homeassistant.components.tplink.common.Discover.discover"
) as discover, patch(
"homeassistant.components.tplink.common.SmartDevice._query_helper"
), patch(
"homeassistant.components.tplink.common.SmartPlug.get_sysinfo",
return_value=SMARTSTRIPWITCH_DATA["sysinfo"],
), patch(
"homeassistant.config_entries.ConfigEntries.async_forward_entry_setup"
):
strip = SmartStrip("123.123.123.123")
discover.return_value = {"123.123.123.123": strip}
assert await async_setup_component(hass, tplink.DOMAIN, config)
await hass.async_block_till_done()
assert hass.data.get(tplink.DOMAIN)
assert hass.data[tplink.DOMAIN].get(COORDINATORS)
assert hass.data[tplink.DOMAIN][COORDINATORS].get(strip.mac)
assert isinstance(
hass.data[tplink.DOMAIN][COORDINATORS][strip.mac],
tplink.SmartPlugDataUpdateCoordinator,
)
data = hass.data[tplink.DOMAIN][COORDINATORS][strip.mac].data
assert data[CONF_ALIAS] == strip.sys_info["children"][0]["alias"]
assert data[CONF_DEVICE_ID] == "1"
async def test_no_config_creates_no_entry(hass):
@ -230,6 +381,42 @@ async def test_no_config_creates_no_entry(hass):
assert mock_setup.call_count == 0
async def test_not_ready(hass: HomeAssistant):
"""Test for not ready when configured devices are not available."""
config = {
tplink.DOMAIN: {
CONF_DISCOVERY: False,
CONF_SWITCH: [{CONF_HOST: "321.321.321.321"}],
}
}
with patch("homeassistant.components.tplink.common.Discover.discover"), patch(
"homeassistant.components.tplink.get_static_devices"
) as get_static_devices, patch(
"homeassistant.components.tplink.common.SmartDevice._query_helper"
), patch(
"homeassistant.components.tplink.light.async_setup_entry",
return_value=mock_coro(True),
), patch(
"homeassistant.components.tplink.switch.async_setup_entry",
return_value=mock_coro(True),
), patch(
"homeassistant.components.tplink.common.SmartPlug.is_dimmable", False
):
switch = SmartPlug("321.321.321.321")
switch.get_sysinfo = MagicMock(side_effect=SmartDeviceException())
get_static_devices.return_value = SmartDevices([], [switch])
await async_setup_component(hass, tplink.DOMAIN, config)
await hass.async_block_till_done()
entries = hass.config_entries.async_entries(tplink.DOMAIN)
assert len(entries) == 1
assert entries[0].state is config_entries.ConfigEntryState.SETUP_RETRY
@pytest.mark.parametrize("platform", ["switch", "light"])
async def test_unload(hass, platform):
"""Test that the async_unload_entry works."""
@ -238,21 +425,35 @@ async def test_unload(hass, platform):
entry.add_to_hass(hass)
with patch(
"homeassistant.components.tplink.get_static_devices"
) as get_static_devices, patch(
"homeassistant.components.tplink.common.SmartDevice._query_helper"
), patch(
f"homeassistant.components.tplink.{platform}.async_setup_entry",
return_value=mock_coro(True),
) as light_setup:
) as async_setup_entry:
config = {
tplink.DOMAIN: {
platform: [{CONF_HOST: "123.123.123.123"}],
CONF_DISCOVERY: False,
}
}
light = SmartBulb("123.123.123.123")
switch = SmartPlug("321.321.321.321")
switch.get_sysinfo = MagicMock(return_value=SMARTPLUGSWITCH_DATA["sysinfo"])
switch.get_emeter_realtime = MagicMock(
return_value=EmeterStatus(SMARTPLUGSWITCH_DATA["realtime"])
)
if platform == "light":
get_static_devices.return_value = SmartDevices([light], [])
elif platform == "switch":
get_static_devices.return_value = SmartDevices([], [switch])
assert await async_setup_component(hass, tplink.DOMAIN, config)
await hass.async_block_till_done()
assert len(light_setup.mock_calls) == 1
assert len(async_setup_entry.mock_calls) == 1
assert tplink.DOMAIN in hass.data
assert await tplink.async_unload_entry(hass, entry)

View File

@ -18,7 +18,7 @@ from homeassistant.components.light import (
ATTR_HS_COLOR,
DOMAIN as LIGHT_DOMAIN,
)
from homeassistant.components.tplink.common import (
from homeassistant.components.tplink.const import (
CONF_DIMMER,
CONF_DISCOVERY,
CONF_LIGHT,