evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Renew Smappee (sensors and switches) (#36445)

Browse source 
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
bsmappee 2020-06-17 13:28:28 +02:00 • committed by GitHub
parent fd67a079db
commit 5228282f69
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
16 changed files with 709 additions and 571 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
.coveragerc
View file

@ -728,7 +728,10 @@ omit =
homeassistant/components/sinch/*
homeassistant/components/slide/*
homeassistant/components/sma/sensor.py
homeassistant/components/smappee/*
homeassistant/components/smappee/__init__.py
homeassistant/components/smappee/api.py
homeassistant/components/smappee/sensor.py
homeassistant/components/smappee/switch.py
homeassistant/components/smarty/*
homeassistant/components/smarthab/*
homeassistant/components/sms/*

1
CODEOWNERS
View file

@ -366,6 +366,7 @@ homeassistant/components/sinch/* @bendikrb
homeassistant/components/sisyphus/* @jkeljo
homeassistant/components/slide/* @ualex73
homeassistant/components/sma/* @kellerza
homeassistant/components/smappee/* @bsmappee
homeassistant/components/smarthab/* @outadoc
homeassistant/components/smartthings/* @andrewsayre
homeassistant/components/smarty/* @z0mbieprocess

383
homeassistant/components/smappee/__init__.py
View file

@ -1,334 +1,109 @@
"""Support for Smappee energy monitor."""
from datetime import datetime, timedelta
import logging
import re
"""The Smappee integration."""
import asyncio
from requests.exceptions import RequestException
import smappy
from pysmappee import Smappee
import voluptuous as vol
from homeassistant.const import (
CONF_CLIENT_ID,
CONF_CLIENT_SECRET,
CONF_HOST,
CONF_PASSWORD,
CONF_USERNAME,
)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.discovery import load_platform
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow, config_validation as cv
from homeassistant.util import Throttle
_LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = "Smappee"
DEFAULT_HOST_PASSWORD = "admin"
CONF_HOST_PASSWORD = "host_password"
DOMAIN = "smappee"
DATA_SMAPPEE = "SMAPPEE"
_SENSOR_REGEX = re.compile(r"(?P<key>([A-Za-z]+))\=(?P<value>([0-9\.]+))")
from . import api, config_flow
from .const import (
AUTHORIZE_URL,
BASE,
DOMAIN,
MIN_TIME_BETWEEN_UPDATES,
SMAPPEE_PLATFORMS,
TOKEN_URL,
)
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Inclusive(CONF_CLIENT_ID, "Server credentials"): cv.string,
vol.Inclusive(CONF_CLIENT_SECRET, "Server credentials"): cv.string,
vol.Inclusive(CONF_USERNAME, "Server credentials"): cv.string,
vol.Inclusive(CONF_PASSWORD, "Server credentials"): cv.string,
vol.Optional(CONF_HOST): cv.string,
vol.Optional(
CONF_HOST_PASSWORD, default=DEFAULT_HOST_PASSWORD
): cv.string,
vol.Required(CONF_CLIENT_ID): cv.string,
vol.Required(CONF_CLIENT_SECRET): cv.string,
}
)
},
extra=vol.ALLOW_EXTRA,
)
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=30)
async def async_setup(hass: HomeAssistant, config: dict):
"""Set up the Smappee component."""
hass.data[DOMAIN] = {}
def setup(hass, config):
"""Set up the Smapee component."""
client_id = config.get(DOMAIN).get(CONF_CLIENT_ID)
client_secret = config.get(DOMAIN).get(CONF_CLIENT_SECRET)
username = config.get(DOMAIN).get(CONF_USERNAME)
password = config.get(DOMAIN).get(CONF_PASSWORD)
host = config.get(DOMAIN).get(CONF_HOST)
host_password = config.get(DOMAIN).get(CONF_HOST_PASSWORD)
if DOMAIN not in config:
return True
smappee = Smappee(client_id, client_secret, username, password, host, host_password)
config_flow.SmappeeFlowHandler.async_register_implementation(
hass,
config_entry_oauth2_flow.LocalOAuth2Implementation(
hass,
DOMAIN,
config[DOMAIN][CONF_CLIENT_ID],
config[DOMAIN][CONF_CLIENT_SECRET],
AUTHORIZE_URL,
TOKEN_URL,
),
)
if not smappee.is_local_active and not smappee.is_remote_active:
_LOGGER.error("Neither Smappee server or local integration enabled.")
return False
hass.data[DATA_SMAPPEE] = smappee
load_platform(hass, "switch", DOMAIN, {}, config)
load_platform(hass, "sensor", DOMAIN, {}, config)
return True
class Smappee:
"""Stores data retrieved from Smappee sensor."""
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Set up Smappee from a config entry."""
implementation = await config_entry_oauth2_flow.async_get_config_entry_implementation(
hass, entry
)
def __init__(
self, client_id, client_secret, username, password, host, host_password
):
"""Initialize the data."""
smappee_api = api.ConfigEntrySmappeeApi(hass, entry, implementation)
self._remote_active = False
self._local_active = False
if client_id is not None:
try:
self._smappy = smappy.Smappee(client_id, client_secret)
self._smappy.authenticate(username, password)
self._remote_active = True
except RequestException as error:
self._smappy = None
_LOGGER.exception("Smappee server authentication failed (%s)", error)
else:
_LOGGER.warning("Smappee server integration init skipped.")
smappee = Smappee(smappee_api)
await hass.async_add_executor_job(smappee.load_service_locations)
if host is not None:
try:
self._localsmappy = smappy.LocalSmappee(host)
self._localsmappy.logon(host_password)
self._local_active = True
except RequestException as error:
self._localsmappy = None
_LOGGER.exception(
"Local Smappee device authentication failed (%s)", error
)
else:
_LOGGER.warning("Smappee local integration init skipped.")
hass.data[DOMAIN][BASE] = SmappeeBase(hass, smappee)
self.locations = {}
self.info = {}
self.consumption = {}
self.sensor_consumption = {}
self.instantaneous = {}
for component in SMAPPEE_PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
if self._remote_active or self._local_active:
self.update()
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(entry, component)
for component in SMAPPEE_PLATFORMS
]
)
)
if unload_ok:
hass.data[DOMAIN].pop(BASE, None)
return unload_ok
class SmappeeBase:
"""An object to hold the PySmappee instance."""
def __init__(self, hass, smappee):
"""Initialize the Smappee API wrapper class."""
self.hass = hass
self.smappee = smappee
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
"""Update data from Smappee API."""
if self.is_remote_active:
service_locations = self._smappy.get_service_locations().get(
"serviceLocations"
)
for location in service_locations:
location_id = location.get("serviceLocationId")
if location_id is not None:
self.sensor_consumption[location_id] = {}
self.locations[location_id] = location.get("name")
self.info[location_id] = self._smappy.get_service_location_info(
location_id
)
_LOGGER.debug(
"Remote info %s %s", self.locations, self.info[location_id]
)
for sensors in self.info[location_id].get("sensors"):
sensor_id = sensors.get("id")
self.sensor_consumption[location_id].update(
{
sensor_id: self.get_sensor_consumption(
location_id, sensor_id, aggregation=3, delta=1440
)
}
)
_LOGGER.debug(
"Remote sensors %s %s",
self.locations,
self.sensor_consumption[location_id],
)
self.consumption[location_id] = self.get_consumption(
location_id, aggregation=3, delta=1440
)
_LOGGER.debug(
"Remote consumption %s %s",
self.locations,
self.consumption[location_id],
)
if self.is_local_active:
self.local_devices = self.get_switches()
_LOGGER.debug("Local switches %s", self.local_devices)
self.instantaneous = self.load_instantaneous()
_LOGGER.debug("Local values %s", self.instantaneous)
@property
def is_remote_active(self):
"""Return true if Smappe server is configured and working."""
return self._remote_active
@property
def is_local_active(self):
"""Return true if Smappe local device is configured and working."""
return self._local_active
def get_switches(self):
"""Get switches from local Smappee."""
if not self.is_local_active:
return
try:
return self._localsmappy.load_command_control_config()
except RequestException as error:
_LOGGER.error("Error getting switches from local Smappee. (%s)", error)
def get_consumption(self, location_id, aggregation, delta):
"""Update data from Smappee."""
# Start & End accept epoch (in milliseconds),
# datetime and pandas timestamps
# Aggregation:
# 1 = 5 min values (only available for the last 14 days),
# 2 = hourly values,
# 3 = daily values,
# 4 = monthly values,
# 5 = quarterly values
if not self.is_remote_active:
return
end = datetime.utcnow()
start = end - timedelta(minutes=delta)
try:
return self._smappy.get_consumption(location_id, start, end, aggregation)
except RequestException as error:
_LOGGER.error("Error getting consumption from Smappee cloud. (%s)", error)
def get_sensor_consumption(self, location_id, sensor_id, aggregation, delta):
"""Update data from Smappee."""
# Start & End accept epoch (in milliseconds),
# datetime and pandas timestamps
# Aggregation:
# 1 = 5 min values (only available for the last 14 days),
# 2 = hourly values,
# 3 = daily values,
# 4 = monthly values,
# 5 = quarterly values
if not self.is_remote_active:
return
end = datetime.utcnow()
start = end - timedelta(minutes=delta)
try:
return self._smappy.get_sensor_consumption(
location_id, sensor_id, start, end, aggregation
)
except RequestException as error:
_LOGGER.error("Error getting consumption from Smappee cloud. (%s)", error)
def actuator_on(self, location_id, actuator_id, is_remote_switch, duration=None):
"""Turn on actuator."""
# Duration = 300,900,1800,3600
# or any other value for an undetermined period of time.
#
# The comport plugs have a tendency to ignore the on/off signal.
# And because you can't read the status of a plug, it's more
# reliable to execute the command twice.
try:
if is_remote_switch:
self._smappy.actuator_on(location_id, actuator_id, duration)
self._smappy.actuator_on(location_id, actuator_id, duration)
else:
self._localsmappy.on_command_control(actuator_id)
self._localsmappy.on_command_control(actuator_id)
except RequestException as error:
_LOGGER.error("Error turning actuator on. (%s)", error)
return False
return True
def actuator_off(self, location_id, actuator_id, is_remote_switch, duration=None):
"""Turn off actuator."""
# Duration = 300,900,1800,3600
# or any other value for an undetermined period of time.
#
# The comport plugs have a tendency to ignore the on/off signal.
# And because you can't read the status of a plug, it's more
# reliable to execute the command twice.
try:
if is_remote_switch:
self._smappy.actuator_off(location_id, actuator_id, duration)
self._smappy.actuator_off(location_id, actuator_id, duration)
else:
self._localsmappy.off_command_control(actuator_id)
self._localsmappy.off_command_control(actuator_id)
except RequestException as error:
_LOGGER.error("Error turning actuator on. (%s)", error)
return False
return True
def active_power(self):
"""Get sum of all instantaneous active power values from local hub."""
if not self.is_local_active:
return
try:
return self._localsmappy.active_power()
except RequestException as error:
_LOGGER.error("Error getting data from Local Smappee unit. (%s)", error)
def active_cosfi(self):
"""Get the average of all instantaneous cosfi values."""
if not self.is_local_active:
return
try:
return self._localsmappy.active_cosfi()
except RequestException as error:
_LOGGER.error("Error getting data from Local Smappee unit. (%s)", error)
def instantaneous_values(self):
"""ReportInstantaneousValues."""
if not self.is_local_active:
return
report_instantaneous_values = self._localsmappy.report_instantaneous_values()
report_result = report_instantaneous_values["report"].split("<BR>")
properties = {}
for lines in report_result:
lines_result = lines.split(",")
for prop in lines_result:
match = _SENSOR_REGEX.search(prop)
if match:
properties[match.group("key")] = match.group("value")
_LOGGER.debug(properties)
return properties
def active_current(self):
"""Get current active Amps."""
if not self.is_local_active:
return
properties = self.instantaneous_values()
return float(properties["current"])
def active_voltage(self):
"""Get current active Voltage."""
if not self.is_local_active:
return
properties = self.instantaneous_values()
return float(properties["voltage"])
def load_instantaneous(self):
"""LoadInstantaneous."""
if not self.is_local_active:
return
try:
return self._localsmappy.load_instantaneous()
except RequestException as error:
_LOGGER.error("Error getting data from Local Smappee unit. (%s)", error)
async def async_update(self):
"""Update all Smappee trends and appliance states."""
await self.hass.async_add_executor_job(
self.smappee.update_trends_and_appliance_states
)

33
homeassistant/components/smappee/api.py Normal file
View file

@ -0,0 +1,33 @@
"""API for Smappee bound to Home Assistant OAuth."""
from asyncio import run_coroutine_threadsafe
from pysmappee import api
from homeassistant import config_entries, core
from homeassistant.helpers import config_entry_oauth2_flow
class ConfigEntrySmappeeApi(api.SmappeeApi):
"""Provide Smappee authentication tied to an OAuth2 based config entry."""
def __init__(
self,
hass: core.HomeAssistant,
config_entry: config_entries.ConfigEntry,
implementation: config_entry_oauth2_flow.AbstractOAuth2Implementation,
):
"""Initialize Smappee Auth."""
self.hass = hass
self.config_entry = config_entry
self.session = config_entry_oauth2_flow.OAuth2Session(
hass, config_entry, implementation
)
super().__init__(None, None, token=self.session.token)
def refresh_tokens(self) -> dict:
"""Refresh and return new Smappee tokens using Home Assistant OAuth2 session."""
run_coroutine_threadsafe(
self.session.async_ensure_token_valid(), self.hass.loop
).result()
return self.session.token

30
homeassistant/components/smappee/config_flow.py Normal file
View file

@ -0,0 +1,30 @@
"""Config flow for Smappee."""
import logging
from homeassistant import config_entries
from homeassistant.helpers import config_entry_oauth2_flow
from .const import DOMAIN # pylint: disable=unused-import
_LOGGER = logging.getLogger(__name__)
class SmappeeFlowHandler(
config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN
):
"""Config flow to handle Smappee OAuth2 authentication."""
DOMAIN = DOMAIN
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
@property
def logger(self) -> logging.Logger:
"""Return logger."""
return logging.getLogger(__name__)
async def async_step_user(self, user_input=None):
"""Handle a flow start."""
if self.hass.config_entries.async_entries(DOMAIN):
return self.async_abort(reason="single_instance_allowed")
return await super().async_step_user(user_input)

15
homeassistant/components/smappee/const.py Normal file
View file

@ -0,0 +1,15 @@
"""Constants for the Smappee integration."""
from datetime import timedelta
DOMAIN = "smappee"
DATA_CLIENT = "smappee_data"
BASE = "BASE"
SMAPPEE_PLATFORMS = ["sensor", "switch"]
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=5)
AUTHORIZE_URL = "https://app1pub.smappee.net/dev/v1/oauth2/authorize"
TOKEN_URL = "https://app1pub.smappee.net/dev/v3/oauth2/token"

10
homeassistant/components/smappee/manifest.json
View file

@ -1,7 +1,13 @@
{
"domain": "smappee",
"name": "Smappee",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/smappee",
"requirements": ["smappy==0.2.16"],
"codeowners": []
"dependencies": ["http"],
"requirements": [
"pysmappee==0.1.0"
],
"codeowners": [
"@bsmappee"
]
}

499
homeassistant/components/smappee/sensor.py
View file

@ -1,170 +1,248 @@
"""Support for monitoring a Smappee energy sensor."""
from datetime import timedelta
import logging
from homeassistant.const import (
DEGREE,
ELECTRICAL_CURRENT_AMPERE,
ENERGY_KILO_WATT_HOUR,
POWER_WATT,
UNIT_PERCENTAGE,
VOLT,
VOLUME_CUBIC_METERS,
)
from homeassistant.const import DEVICE_CLASS_POWER, ENERGY_WATT_HOUR, POWER_WATT, VOLT
from homeassistant.helpers.entity import Entity
from . import DATA_SMAPPEE
from .const import BASE, DOMAIN
_LOGGER = logging.getLogger(__name__)
SENSOR_PREFIX = "Smappee"
SENSOR_TYPES = {
"solar": ["Solar", "mdi:white-balance-sunny", "local", POWER_WATT, "solar"],
"active_power": [
"Active Power",
"mdi:power-plug",
"local",
TREND_SENSORS = {
"total_power": [
"Total consumption - Active power",
None,
POWER_WATT,
"active_power",
"total_power",
DEVICE_CLASS_POWER,
],
"current": ["Current", "mdi:gauge", "local", ELECTRICAL_CURRENT_AMPERE, "current"],
"voltage": ["Voltage", "mdi:gauge", "local", VOLT, "voltage"],
"active_cosfi": [
"Power Factor",
"mdi:gauge",
"local",
UNIT_PERCENTAGE,
"active_cosfi",
"total_reactive_power": [
"Total consumption - Reactive power",
None,
POWER_WATT,
"total_reactive_power",
DEVICE_CLASS_POWER,
],
"alwayson_today": [
"Always On Today",
"mdi:gauge",
"remote",
ENERGY_KILO_WATT_HOUR,
"alwaysOn",
],
"solar_today": [
"Solar Today",
"mdi:white-balance-sunny",
"remote",
ENERGY_KILO_WATT_HOUR,
"solar",
"alwayson": [
"Always on - Active power",
None,
POWER_WATT,
"alwayson",
DEVICE_CLASS_POWER,
],
"power_today": [
"Power Today",
"Total consumption - Today",
"mdi:power-plug",
"remote",
ENERGY_KILO_WATT_HOUR,
"consumption",
ENERGY_WATT_HOUR,
"power_today",
None,
],
"water_sensor_1": [
"Water Sensor 1",
"mdi:water",
"water",
VOLUME_CUBIC_METERS,
"value1",
"power_current_hour": [
"Total consumption - Current hour",
"mdi:power-plug",
ENERGY_WATT_HOUR,
"power_current_hour",
None,
],
"water_sensor_2": [
"Water Sensor 2",
"mdi:water",
"water",
VOLUME_CUBIC_METERS,
"value2",
"power_last_5_minutes": [
"Total consumption - Last 5 minutes",
"mdi:power-plug",
ENERGY_WATT_HOUR,
"power_last_5_minutes",
None,
],
"water_sensor_temperature": [
"Water Sensor Temperature",
"mdi:temperature-celsius",
"water",
DEGREE,
"temperature",
"alwayson_today": [
"Always on - Today",
"mdi:sleep",
ENERGY_WATT_HOUR,
"alwayson_today",
None,
],
"water_sensor_humidity": [
"Water Sensor Humidity",
"mdi:water-percent",
"water",
UNIT_PERCENTAGE,
"humidity",
}
SOLAR_SENSORS = {
"solar_power": [
"Total production - Active power",
None,
POWER_WATT,
"solar_power",
DEVICE_CLASS_POWER,
],
"water_sensor_battery": [
"Water Sensor Battery",
"mdi:battery",
"water",
UNIT_PERCENTAGE,
"battery",
"solar_today": [
"Total production - Today",
"mdi:white-balance-sunny",
ENERGY_WATT_HOUR,
"solar_today",
None,
],
"solar_current_hour": [
"Total production - Current hour",
"mdi:white-balance-sunny",
ENERGY_WATT_HOUR,
"solar_current_hour",
None,
],
}
VOLTAGE_SENSORS = {
"phase_voltages_a": [
"Phase voltages - A",
"mdi:flash",
VOLT,
"phase_voltage_a",
["ONE", "TWO", "THREE_STAR", "THREE_DELTA"],
None,
],
"phase_voltages_b": [
"Phase voltages - B",
"mdi:flash",
VOLT,
"phase_voltage_b",
["TWO", "THREE_STAR", "THREE_DELTA"],
None,
],
"phase_voltages_c": [
"Phase voltages - C",
"mdi:flash",
VOLT,
"phase_voltage_c",
["THREE_STAR"],
None,
],
"line_voltages_a": [
"Line voltages - A",
"mdi:flash",
VOLT,
"line_voltage_a",
["ONE", "TWO", "THREE_STAR", "THREE_DELTA"],
None,
],
"line_voltages_b": [
"Line voltages - B",
"mdi:flash",
VOLT,
"line_voltage_b",
["TWO", "THREE_STAR", "THREE_DELTA"],
None,
],
"line_voltages_c": [
"Line voltages - C",
"mdi:flash",
VOLT,
"line_voltage_c",
["THREE_STAR", "THREE_DELTA"],
None,
],
}
SCAN_INTERVAL = timedelta(seconds=30)
def setup_platform(hass, config, add_entities, discovery_info=None):
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Smappee sensor."""
smappee = hass.data[DATA_SMAPPEE]
smappee_base = hass.data[DOMAIN][BASE]
dev = []
if smappee.is_remote_active:
for location_id in smappee.locations.keys():
for sensor in SENSOR_TYPES:
if "remote" in SENSOR_TYPES[sensor]:
dev.append(
SmappeeSensor(
smappee, location_id, sensor, SENSOR_TYPES[sensor]
)
entities = []
for service_location in smappee_base.smappee.service_locations.values():
# Add all basic sensors (realtime values and aggregators)
for sensor in TREND_SENSORS:
entities.append(
SmappeeSensor(
smappee_base=smappee_base,
service_location=service_location,
sensor=sensor,
attributes=TREND_SENSORS[sensor],
)
)
# Add solar sensors
if service_location.has_solar_production:
for sensor in SOLAR_SENSORS:
entities.append(
SmappeeSensor(
smappee_base=smappee_base,
service_location=service_location,
sensor=sensor,
attributes=SOLAR_SENSORS[sensor],
)
elif "water" in SENSOR_TYPES[sensor]:
for items in smappee.info[location_id].get("sensors"):
dev.append(
SmappeeSensor(
smappee,
location_id,
"{}:{}".format(sensor, items.get("id")),
SENSOR_TYPES[sensor],
)
)
)
if smappee.is_local_active:
if smappee.is_remote_active:
location_keys = smappee.locations.keys()
else:
location_keys = [None]
for location_id in location_keys:
for sensor in SENSOR_TYPES:
if "local" in SENSOR_TYPES[sensor]:
dev.append(
SmappeeSensor(
smappee, location_id, sensor, SENSOR_TYPES[sensor]
)
# Add all CT measurements
for measurement_id, measurement in service_location.measurements.items():
entities.append(
SmappeeSensor(
smappee_base=smappee_base,
service_location=service_location,
sensor="load",
attributes=[
measurement.name,
None,
POWER_WATT,
measurement_id,
DEVICE_CLASS_POWER,
],
)
)
# Add phase- and line voltages
for sensor_name, sensor in VOLTAGE_SENSORS.items():
if service_location.phase_type in sensor[4]:
entities.append(
SmappeeSensor(
smappee_base=smappee_base,
service_location=service_location,
sensor=sensor_name,
attributes=sensor,
)
)
add_entities(dev, True)
# Add Gas and Water sensors
for sensor_id, sensor in service_location.sensors.items():
for channel in sensor.channels:
gw_icon = "mdi:gas-cylinder"
if channel.get("type") == "water":
gw_icon = "mdi:water"
entities.append(
SmappeeSensor(
smappee_base=smappee_base,
service_location=service_location,
sensor="sensor",
attributes=[
channel.get("name"),
gw_icon,
channel.get("uom"),
f"{sensor_id}-{channel.get('channel')}",
None,
],
)
)
async_add_entities(entities, True)
class SmappeeSensor(Entity):
"""Implementation of a Smappee sensor."""
def __init__(self, smappee, location_id, sensor, attributes):
def __init__(self, smappee_base, service_location, sensor, attributes):
"""Initialize the Smappee sensor."""
self._smappee = smappee
self._location_id = location_id
self._attributes = attributes
self._smappee_base = smappee_base
self._service_location = service_location
self._sensor = sensor
self.data = None
self._state = None
self._name = self._attributes[0]
self._icon = self._attributes[1]
self._type = self._attributes[2]
self._unit_of_measurement = self._attributes[3]
self._smappe_name = self._attributes[4]
self._name = attributes[0]
self._icon = attributes[1]
self._unit_of_measurement = attributes[2]
self._sensor_id = attributes[3]
self._device_class = attributes[4]
@property
def name(self):
"""Return the name of the sensor."""
if self._location_id:
location_name = self._smappee.locations[self._location_id]
else:
location_name = "Local"
"""Return the name for this sensor."""
if self._sensor in ["sensor", "load"]:
return (
f"{self._service_location.service_location_name} - "
f"{self._sensor.title()} - {self._name}"
)
return f"{SENSOR_PREFIX} {location_name} {self._name}"
return f"{self._service_location.service_location_name} - {self._name}"
@property
def icon(self):
@ -176,97 +254,94 @@ class SmappeeSensor(Entity):
"""Return the state of the sensor."""
return self._state
@property
def device_class(self):
"""Return the class of this device, from component DEVICE_CLASSES."""
return self._device_class
@property
def unit_of_measurement(self):
"""Return the unit of measurement of this entity, if any."""
return self._unit_of_measurement
@property
def device_state_attributes(self):
"""Return the state attributes of the device."""
attr = {}
if self._location_id:
attr["Location Id"] = self._location_id
attr["Location Name"] = self._smappee.locations[self._location_id]
return attr
def update(self):
"""Get the latest data from Smappee and update the state."""
self._smappee.update()
if self._sensor in ["alwayson_today", "solar_today", "power_today"]:
data = self._smappee.consumption[self._location_id]
if data:
consumption = data.get("consumptions")[-1]
_LOGGER.debug("%s %s", self._sensor, consumption)
value = consumption.get(self._smappe_name)
self._state = round(value / 1000, 2)
elif self._sensor == "active_cosfi":
cosfi = self._smappee.active_cosfi()
_LOGGER.debug("%s %s", self._sensor, cosfi)
if cosfi:
self._state = round(cosfi, 2)
elif self._sensor == "current":
current = self._smappee.active_current()
_LOGGER.debug("%s %s", self._sensor, current)
if current:
self._state = round(current, 2)
elif self._sensor == "voltage":
voltage = self._smappee.active_voltage()
_LOGGER.debug("%s %s", self._sensor, voltage)
if voltage:
self._state = round(voltage, 3)
elif self._sensor == "active_power":
data = self._smappee.instantaneous
_LOGGER.debug("%s %s", self._sensor, data)
if data:
value1 = [
float(i["value"])
for i in data
if i["key"].endswith("phase0ActivePower")
]
value2 = [
float(i["value"])
for i in data
if i["key"].endswith("phase1ActivePower")
]
value3 = [
float(i["value"])
for i in data
if i["key"].endswith("phase2ActivePower")
]
active_power = sum(value1 + value2 + value3) / 1000
self._state = round(active_power, 2)
elif self._sensor == "solar":
data = self._smappee.instantaneous
_LOGGER.debug("%s %s", self._sensor, data)
if data:
value1 = [
float(i["value"])
for i in data
if i["key"].endswith("phase3ActivePower")
]
value2 = [
float(i["value"])
for i in data
if i["key"].endswith("phase4ActivePower")
]
value3 = [
float(i["value"])
for i in data
if i["key"].endswith("phase5ActivePower")
]
power = sum(value1 + value2 + value3) / 1000
self._state = round(power, 2)
elif self._type == "water":
sensor_name, sensor_id = self._sensor.split(":")
data = self._smappee.sensor_consumption[self._location_id].get(
int(sensor_id)
def unique_id(self,):
"""Return the unique ID for this sensor."""
if self._sensor in ["load", "sensor"]:
return (
f"{self._service_location.device_serial_number}-"
f"{self._service_location.service_location_id}-"
f"{self._sensor}-{self._sensor_id}"
)
if data:
tempdata = data.get("records")
if tempdata:
consumption = tempdata[-1]
_LOGGER.debug("%s (%s) %s", sensor_name, sensor_id, consumption)
value = consumption.get(self._smappe_name)
self._state = value
return (
f"{self._service_location.device_serial_number}-"
f"{self._service_location.service_location_id}-"
f"{self._sensor}"
)
@property
def device_info(self):
"""Return the device info for this sensor."""
return {
"identifiers": {(DOMAIN, self._service_location.device_serial_number)},
"name": self._service_location.service_location_name,
"manufacturer": "Smappee",
"model": self._service_location.device_model,
"sw_version": self._service_location.firmware_version,
}
async def async_update(self):
"""Get the latest data from Smappee and update the state."""
await self._smappee_base.async_update()
if self._sensor == "total_power":
self._state = self._service_location.total_power
elif self._sensor == "total_reactive_power":
self._state = self._service_location.total_reactive_power
elif self._sensor == "solar_power":
self._state = self._service_location.solar_power
elif self._sensor == "alwayson":
self._state = self._service_location.alwayson
elif self._sensor in [
"phase_voltages_a",
"phase_voltages_b",
"phase_voltages_c",
]:
phase_voltages = self._service_location.phase_voltages
if phase_voltages is not None:
if self._sensor == "phase_voltages_a":
self._state = phase_voltages[0]
elif self._sensor == "phase_voltages_b":
self._state = phase_voltages[1]
elif self._sensor == "phase_voltages_c":
self._state = phase_voltages[2]
elif self._sensor in ["line_voltages_a", "line_voltages_b", "line_voltages_c"]:
line_voltages = self._service_location.line_voltages
if line_voltages is not None:
if self._sensor == "line_voltages_a":
self._state = line_voltages[0]
elif self._sensor == "line_voltages_b":
self._state = line_voltages[1]
elif self._sensor == "line_voltages_c":
self._state = line_voltages[2]
elif self._sensor in [
"power_today",
"power_current_hour",
"power_last_5_minutes",
"solar_today",
"solar_current_hour",
"alwayson_today",
]:
trend_value = self._service_location.aggregated_values.get(self._sensor)
self._state = round(trend_value) if trend_value is not None else None
elif self._sensor == "load":
self._state = self._service_location.measurements.get(
self._sensor_id
).active_total
elif self._sensor == "sensor":
sensor_id, channel_id = self._sensor_id.split("-")
sensor = self._service_location.sensors.get(int(sensor_id))
for channel in sensor.channels:
if channel.get("channel") == int(channel_id):
self._state = channel.get("value_today")

14
homeassistant/components/smappee/strings.json Normal file
View file

@ -0,0 +1,14 @@
{
"config": {
"step": {
"pick_implementation": {
"title": "Pick Authentication Method"
}
},
"abort": {
"authorize_url_timeout": "Timeout generating authorize url.",
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]",
"missing_configuration": "The component is not configured. Please follow the documentation."
}
}
}

197
homeassistant/components/smappee/switch.py
View file

@ -1,62 +1,105 @@
"""Support for interacting with Smappee Comport Plugs."""
"""Support for interacting with Smappee Comport Plugs, Switches and Output Modules."""
from datetime import timedelta
import logging
from homeassistant.components.switch import SwitchEntity
from . import DATA_SMAPPEE
from .const import BASE, DOMAIN
_LOGGER = logging.getLogger(__name__)
ICON = "mdi:power-plug"
SWITCH_PREFIX = "Switch"
ICON = "mdi:toggle-switch"
SCAN_INTERVAL = timedelta(seconds=5)
def setup_platform(hass, config, add_entities, discovery_info=None):
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Smappee Comfort Plugs."""
smappee = hass.data[DATA_SMAPPEE]
smappee_base = hass.data[DOMAIN][BASE]
dev = []
if smappee.is_remote_active:
for location_id in smappee.locations.keys():
for items in smappee.info[location_id].get("actuators"):
if items.get("name") != "":
_LOGGER.debug("Remote actuator %s", items)
dev.append(
SmappeeSwitch(
smappee, items.get("name"), location_id, items.get("id")
entities = []
for service_location in smappee_base.smappee.service_locations.values():
for actuator_id, actuator in service_location.actuators.items():
if actuator.type in ["SWITCH", "COMFORT_PLUG"]:
entities.append(
SmappeeActuator(
smappee_base,
service_location,
actuator.name,
actuator_id,
actuator.type,
actuator.serialnumber,
)
)
elif actuator.type == "INFINITY_OUTPUT_MODULE":
for option in actuator.state_options:
entities.append(
SmappeeActuator(
smappee_base,
service_location,
actuator.name,
actuator_id,
actuator.type,
actuator.serialnumber,
actuator_state_option=option,
)
)
elif smappee.is_local_active:
for items in smappee.local_devices:
_LOGGER.debug("Local actuator %s", items)
dev.append(
SmappeeSwitch(smappee, items.get("value"), None, items.get("key"))
)
add_entities(dev)
async_add_entities(entities, True)
class SmappeeSwitch(SwitchEntity):
class SmappeeActuator(SwitchEntity):
"""Representation of a Smappee Comport Plug."""
def __init__(self, smappee, name, location_id, switch_id):
def __init__(
self,
smappee_base,
service_location,
name,
actuator_id,
actuator_type,
actuator_serialnumber,
actuator_state_option=None,
):
"""Initialize a new Smappee Comfort Plug."""
self._name = name
self._state = False
self._smappee = smappee
self._location_id = location_id
self._switch_id = switch_id
self._remoteswitch = True
if location_id is None:
self._remoteswitch = False
self._smappee_base = smappee_base
self._service_location = service_location
self._actuator_name = name
self._actuator_id = actuator_id
self._actuator_type = actuator_type
self._actuator_serialnumber = actuator_serialnumber
self._actuator_state_option = actuator_state_option
self._state = self._service_location.actuators.get(actuator_id).state
self._connection_state = self._service_location.actuators.get(
actuator_id
).connection_state
@property
def name(self):
"""Return the name of the switch."""
return self._name
if self._actuator_type == "INFINITY_OUTPUT_MODULE":
return (
f"{self._service_location.service_location_name} - "
f"Output module - {self._actuator_name} - {self._actuator_state_option}"
)
# Switch or comfort plug
return (
f"{self._service_location.service_location_name} - "
f"{self._actuator_type.title()} - {self._actuator_name}"
)
@property
def is_on(self):
"""Return true if switch is on."""
return self._state
if self._actuator_type == "INFINITY_OUTPUT_MODULE":
return (
self._service_location.actuators.get(self._actuator_id).state
== self._actuator_state_option
)
# Switch or comfort plug
return self._state == "ON_ON"
@property
def icon(self):
@ -65,24 +108,80 @@ class SmappeeSwitch(SwitchEntity):
def turn_on(self, **kwargs):
"""Turn on Comport Plug."""
if self._smappee.actuator_on(
self._location_id, self._switch_id, self._remoteswitch
):
self._state = True
if self._actuator_type in ["SWITCH", "COMFORT_PLUG"]:
self._service_location.set_actuator_state(self._actuator_id, state="ON_ON")
elif self._actuator_type == "INFINITY_OUTPUT_MODULE":
self._service_location.set_actuator_state(
self._actuator_id, state=self._actuator_state_option
)
def turn_off(self, **kwargs):
"""Turn off Comport Plug."""
if self._smappee.actuator_off(
self._location_id, self._switch_id, self._remoteswitch
):
self._state = False
if self._actuator_type in ["SWITCH", "COMFORT_PLUG"]:
self._service_location.set_actuator_state(
self._actuator_id, state="OFF_OFF"
)
elif self._actuator_type == "INFINITY_OUTPUT_MODULE":
self._service_location.set_actuator_state(
self._actuator_id, state="PLACEHOLDER", api=False
)
@property
def device_state_attributes(self):
"""Return the state attributes of the device."""
attr = {}
if self._remoteswitch:
attr["Location Id"] = self._location_id
attr["Location Name"] = self._smappee.locations[self._location_id]
attr["Switch Id"] = self._switch_id
return attr
def available(self):
"""Return True if entity is available. Unavailable for COMFORT_PLUGS."""
return (
self._connection_state == "CONNECTED"
or self._actuator_type == "COMFORT_PLUG"
)
@property
def today_energy_kwh(self):
"""Return the today total energy usage in kWh."""
if self._actuator_type == "SWITCH":
cons = self._service_location.actuators.get(
self._actuator_id
).consumption_today
if cons is not None:
return round(cons / 1000.0, 2)
return None
@property
def unique_id(self,):
"""Return the unique ID for this switch."""
if self._actuator_type == "INFINITY_OUTPUT_MODULE":
return (
f"{self._service_location.device_serial_number}-"
f"{self._service_location.service_location_id}-actuator-"
f"{self._actuator_id}-{self._actuator_state_option}"
)
# Switch or comfort plug
return (
f"{self._service_location.device_serial_number}-"
f"{self._service_location.service_location_id}-actuator-"
f"{self._actuator_id}"
)
@property
def device_info(self):
"""Return the device info for this switch."""
return {
"identifiers": {(DOMAIN, self._service_location.device_serial_number)},
"name": self._service_location.service_location_name,
"manufacturer": "Smappee",
"model": self._service_location.device_model,
"sw_version": self._service_location.firmware_version,
}
async def async_update(self):
"""Get the latest data from Smappee and update the state."""
await self._smappee_base.async_update()
new_state = self._service_location.actuators.get(self._actuator_id).state
if new_state != self._state:
self._state = new_state
self.async_write_ha_state()
self._connection_state = self._service_location.actuators.get(
self._actuator_id
).connection_state

14
homeassistant/components/smappee/translations/en.json Normal file
View file

@ -0,0 +1,14 @@
{
"config": {
"step": {
"pick_implementation": {
"title": "Pick Authentication Method"
}
},
"abort": {
"authorize_url_timeout": "Timeout generating authorize url.",
"single_instance_allowed": "Already configured. Only a single configuration possible.",
"missing_configuration": "The component is not configured. Please follow the documentation."
}
}
}

1
homeassistant/generated/config_flows.py
View file

@ -135,6 +135,7 @@ FLOWS = [
"sentry",
"shopping_list",
"simplisafe",
"smappee",
"smartthings",
"smhi",
"solaredge",

6
requirements_all.txt
View file

@ -1609,6 +1609,9 @@ pysignalclirestapi==0.3.4
# homeassistant.components.sma
pysma==0.3.5
# homeassistant.components.smappee
pysmappee==0.1.0
# homeassistant.components.smartthings
pysmartapp==0.3.2
@ -1964,9 +1967,6 @@ sleepyq==0.7
# homeassistant.components.xmpp
slixmpp==1.5.1
# homeassistant.components.smappee
smappy==0.2.16
# homeassistant.components.smarthab
smarthab==0.20

3
requirements_test_all.txt
View file

@ -702,6 +702,9 @@ pysignalclirestapi==0.3.4
# homeassistant.components.sma
pysma==0.3.5
# homeassistant.components.smappee
pysmappee==0.1.0
# homeassistant.components.smartthings
pysmartapp==0.3.2

1
tests/components/smappee/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Tests for the Smappee integration."""

68
tests/components/smappee/test_config_flow.py Normal file
View file

@ -0,0 +1,68 @@
"""Test the Smappee config flow."""
from homeassistant import config_entries, data_entry_flow, setup
from homeassistant.components.smappee.const import AUTHORIZE_URL, DOMAIN, TOKEN_URL
from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET
from homeassistant.helpers import config_entry_oauth2_flow
from tests.async_mock import patch
from tests.common import MockConfigEntry
CLIENT_ID = "1234"
CLIENT_SECRET = "5678"
async def test_abort_if_existing_entry(hass):
"""Check flow abort when an entry already exist."""
MockConfigEntry(domain=DOMAIN).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "single_instance_allowed"
async def test_full_flow(hass, aiohttp_client, aioclient_mock):
"""Check full flow."""
assert await setup.async_setup_component(
hass,
DOMAIN,
{
DOMAIN: {CONF_CLIENT_ID: CLIENT_ID, CONF_CLIENT_SECRET: CLIENT_SECRET},
"http": {"base_url": "https://example.com"},
},
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
state = config_entry_oauth2_flow._encode_jwt(hass, {"flow_id": result["flow_id"]})
assert result["url"] == (
f"{AUTHORIZE_URL}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}"
)
client = await aiohttp_client(hass.http.app)
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
aioclient_mock.post(
TOKEN_URL,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
with patch(
"homeassistant.components.smappee.async_setup_entry", return_value=True
) as mock_setup:
await hass.config_entries.flow.async_configure(result["flow_id"])
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert len(mock_setup.mock_calls) == 1