Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Aurora abb (solar) configflow (#36300) 

Co-authored-by: J. Nick Koston <nick@koston.org>
pull/58450/head
Dave T 2021-10-26 02:04:42 +01:00 committed by GitHub
parent 44aa1fdc66
commit e22aaea5b2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 786 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -83,7 +83,6 @@ omit =
homeassistant/components/aurora/binary_sensor.py
homeassistant/components/aurora/const.py
homeassistant/components/aurora/sensor.py
homeassistant/components/aurora_abb_powerone/sensor.py
homeassistant/components/avea/light.py
homeassistant/components/avion/light.py
homeassistant/components/azure_devops/__init__.py

48
homeassistant/components/aurora_abb_powerone/__init__.py
View File

@ -1 +1,49 @@
"""The Aurora ABB Powerone PV inverter sensor integration."""
# Reference info:
# https://s1.solacity.com/docs/PVI-3.0-3.6-4.2-OUTD-US%20Manual.pdf
# http://www.drhack.it/images/PDF/AuroraCommunicationProtocol_4_2.pdf
#
# Developer note:
# vscode devcontainer: use the following to access USB device:
# "runArgs": ["-e", "GIT_EDITOR=code --wait", "--device=/dev/ttyUSB0"],
import logging
from aurorapy.client import AuroraSerialClient
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ADDRESS, CONF_PORT
from homeassistant.core import HomeAssistant
from .const import DOMAIN
PLATFORMS = ["sensor"]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Set up Aurora ABB PowerOne from a config entry."""
comport = entry.data[CONF_PORT]
address = entry.data[CONF_ADDRESS]
serclient = AuroraSerialClient(address, comport, parity="N", timeout=1)
hass.data.setdefault(DOMAIN, {})[entry.unique_id] = serclient
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
# It should not be necessary to close the serial port because we close
# it after every use in sensor.py, i.e. no need to do entry["client"].close()
if unload_ok:
hass.data[DOMAIN].pop(entry.unique_id)
return unload_ok

51
homeassistant/components/aurora_abb_powerone/aurora_device.py Normal file
View File

@ -0,0 +1,51 @@
"""Top level class for AuroraABBPowerOneSolarPV inverters and sensors."""
import logging
from aurorapy.client import AuroraSerialClient
from homeassistant.helpers.entity import Entity
from .const import (
ATTR_DEVICE_NAME,
ATTR_FIRMWARE,
ATTR_MODEL,
ATTR_SERIAL_NUMBER,
DEFAULT_DEVICE_NAME,
DOMAIN,
MANUFACTURER,
)
_LOGGER = logging.getLogger(__name__)
class AuroraDevice(Entity):
"""Representation of an Aurora ABB PowerOne device."""
def __init__(self, client: AuroraSerialClient, data) -> None:
"""Initialise the basic device."""
self._data = data
self.type = "device"
self.client = client
self._available = True
@property
def unique_id(self) -> str:
"""Return the unique id for this device."""
serial = self._data[ATTR_SERIAL_NUMBER]
return f"{serial}_{self.type}"
@property
def available(self) -> bool:
"""Return True if entity is available."""
return self._available
@property
def device_info(self):
"""Return device specific attributes."""
return {
"identifiers": {(DOMAIN, self._data[ATTR_SERIAL_NUMBER])},
"manufacturer": MANUFACTURER,
"model": self._data[ATTR_MODEL],
"name": self._data.get(ATTR_DEVICE_NAME, DEFAULT_DEVICE_NAME),
"sw_version": self._data[ATTR_FIRMWARE],
}

147
homeassistant/components/aurora_abb_powerone/config_flow.py Normal file
View File

@ -0,0 +1,147 @@
"""Config flow for Aurora ABB PowerOne integration."""
import logging
from aurorapy.client import AuroraError, AuroraSerialClient
import serial.tools.list_ports
import voluptuous as vol
from homeassistant import config_entries, core
from homeassistant.const import CONF_ADDRESS, CONF_PORT
from .const import (
ATTR_FIRMWARE,
ATTR_MODEL,
ATTR_SERIAL_NUMBER,
DEFAULT_ADDRESS,
DEFAULT_INTEGRATION_TITLE,
DOMAIN,
MAX_ADDRESS,
MIN_ADDRESS,
)
_LOGGER = logging.getLogger(__name__)
def validate_and_connect(hass: core.HomeAssistant, data):
"""Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA with values provided by the user.
"""
comport = data[CONF_PORT]
address = data[CONF_ADDRESS]
_LOGGER.debug("Intitialising com port=%s", comport)
ret = {}
ret["title"] = DEFAULT_INTEGRATION_TITLE
try:
client = AuroraSerialClient(address, comport, parity="N", timeout=1)
client.connect()
ret[ATTR_SERIAL_NUMBER] = client.serial_number()
ret[ATTR_MODEL] = f"{client.version()} ({client.pn()})"
ret[ATTR_FIRMWARE] = client.firmware(1)
_LOGGER.info("Returning device info=%s", ret)
except AuroraError as err:
_LOGGER.warning("Could not connect to device=%s", comport)
raise err
finally:
if client.serline.isOpen():
client.close()
# Return info we want to store in the config entry.
return ret
def scan_comports():
"""Find and store available com ports for the GUI dropdown."""
comports = serial.tools.list_ports.comports(include_links=True)
comportslist = []
for port in comports:
comportslist.append(port.device)
_LOGGER.debug("COM port option: %s", port.device)
if len(comportslist) > 0:
return comportslist, comportslist[0]
_LOGGER.warning("No com ports found. Need a valid RS485 device to communicate")
return None, None
class AuroraABBConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Aurora ABB PowerOne."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
def __init__(self):
"""Initialise the config flow."""
self.config = None
self._comportslist = None
self._defaultcomport = None
async def async_step_import(self, config: dict):
"""Import a configuration from config.yaml."""
if self.hass.config_entries.async_entries(DOMAIN):
return self.async_abort(reason="already_setup")
conf = {}
conf[ATTR_SERIAL_NUMBER] = "sn_unknown_yaml"
conf[ATTR_MODEL] = "model_unknown_yaml"
conf[ATTR_FIRMWARE] = "fw_unknown_yaml"
conf[CONF_PORT] = config["device"]
conf[CONF_ADDRESS] = config["address"]
# config["name"] from yaml is ignored.
await self.async_set_unique_id(self.flow_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(title=DEFAULT_INTEGRATION_TITLE, data=conf)
async def async_step_user(self, user_input=None):
"""Handle a flow initialised by the user."""
errors = {}
if self._comportslist is None:
result = await self.hass.async_add_executor_job(scan_comports)
self._comportslist, self._defaultcomport = result
if self._defaultcomport is None:
return self.async_abort(reason="no_serial_ports")
# Handle the initial step.
if user_input is not None:
try:
info = await self.hass.async_add_executor_job(
validate_and_connect, self.hass, user_input
)
info.update(user_input)
# Bomb out early if someone has already set up this device.
device_unique_id = info["serial_number"]
await self.async_set_unique_id(device_unique_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(title=info["title"], data=info)
except OSError as error:
if error.errno == 19: # No such device.
errors["base"] = "invalid_serial_port"
except AuroraError as error:
if "could not open port" in str(error):
errors["base"] = "cannot_open_serial_port"
elif "No response after" in str(error):
errors["base"] = "cannot_connect" # could be dark
else:
_LOGGER.error(
"Unable to communicate with Aurora ABB Inverter at %s: %s %s",
user_input[CONF_PORT],
type(error),
error,
)
errors["base"] = "cannot_connect"
# If no user input, must be first pass through the config. Show initial form.
config_options = {
vol.Required(CONF_PORT, default=self._defaultcomport): vol.In(
self._comportslist
),
vol.Required(CONF_ADDRESS, default=DEFAULT_ADDRESS): vol.In(
range(MIN_ADDRESS, MAX_ADDRESS + 1)
),
}
schema = vol.Schema(config_options)
return self.async_show_form(step_id="user", data_schema=schema, errors=errors)

22
homeassistant/components/aurora_abb_powerone/const.py Normal file
View File

@ -0,0 +1,22 @@
"""Constants for the Aurora ABB PowerOne integration."""
DOMAIN = "aurora_abb_powerone"
# Min max addresses and default according to here:
# https://library.e.abb.com/public/e57212c407344a16b4644cee73492b39/PVI-3.0_3.6_4.2-TL-OUTD-Product%20manual%20EN-RevB(M000016BG).pdf
MIN_ADDRESS = 2
MAX_ADDRESS = 63
DEFAULT_ADDRESS = 2
DEFAULT_INTEGRATION_TITLE = "PhotoVoltaic Inverters"
DEFAULT_DEVICE_NAME = "Solar Inverter"
DEVICES = "devices"
MANUFACTURER = "ABB"
ATTR_DEVICE_NAME = "device_name"
ATTR_DEVICE_ID = "device_id"
ATTR_SERIAL_NUMBER = "serial_number"
ATTR_MODEL = "model"
ATTR_FIRMWARE = "firmware"

9
homeassistant/components/aurora_abb_powerone/manifest.json
View File

@ -1,8 +1,11 @@
{
"domain": "aurora_abb_powerone",
"name": "Aurora ABB Solar PV",
"documentation": "https://www.home-assistant.io/integrations/aurora_abb_powerone/",
"codeowners": ["@davet2001"],
"name": "Aurora ABB PowerOne Solar PV",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/aurora_abb_powerone",
"requirements": ["aurorapy==0.2.6"],
"codeowners": [
"@davet2001"
],
"iot_class": "local_polling"
}

96
homeassistant/components/aurora_abb_powerone/sensor.py
View File

@ -10,54 +10,85 @@ from homeassistant.components.sensor import (
STATE_CLASS_MEASUREMENT,
SensorEntity,
)
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import (
CONF_ADDRESS,
CONF_DEVICE,
CONF_NAME,
DEVICE_CLASS_POWER,
DEVICE_CLASS_TEMPERATURE,
POWER_WATT,
TEMP_CELSIUS,
)
from homeassistant.exceptions import InvalidStateError
import homeassistant.helpers.config_validation as cv
from .aurora_device import AuroraDevice
from .const import DEFAULT_ADDRESS, DOMAIN
_LOGGER = logging.getLogger(__name__)
DEFAULT_ADDRESS = 2
DEFAULT_NAME = "Solar PV"
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_DEVICE): cv.string,
vol.Optional(CONF_ADDRESS, default=DEFAULT_ADDRESS): cv.positive_int,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_NAME, default="Solar PV"): cv.string,
}
)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Aurora ABB PowerOne device."""
devices = []
comport = config[CONF_DEVICE]
address = config[CONF_ADDRESS]
name = config[CONF_NAME]
_LOGGER.debug("Intitialising com port=%s address=%s", comport, address)
client = AuroraSerialClient(address, comport, parity="N", timeout=1)
devices.append(AuroraABBSolarPVMonitorSensor(client, name, "Power"))
add_entities(devices, True)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up based on configuration.yaml (DEPRECATED)."""
_LOGGER.warning(
"Loading aurora_abb_powerone via platform config is deprecated; The configuration"
" has been migrated to a config entry and can be safely removed from configuration.yaml"
)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
)
)
class AuroraABBSolarPVMonitorSensor(SensorEntity):
"""Representation of a Sensor."""
async def async_setup_entry(hass, config_entry, async_add_entities) -> None:
"""Set up aurora_abb_powerone sensor based on a config entry."""
entities = []
sensortypes = [
{"parameter": "instantaneouspower", "name": "Power Output"},
{"parameter": "temperature", "name": "Temperature"},
]
client = hass.data[DOMAIN][config_entry.unique_id]
data = config_entry.data
for sens in sensortypes:
entities.append(AuroraSensor(client, data, sens["name"], sens["parameter"]))
_LOGGER.debug("async_setup_entry adding %d entities", len(entities))
async_add_entities(entities, True)
class AuroraSensor(AuroraDevice, SensorEntity):
"""Representation of a Sensor on a Aurora ABB PowerOne Solar inverter."""
_attr_state_class = STATE_CLASS_MEASUREMENT
_attr_native_unit_of_measurement = POWER_WATT
_attr_device_class = DEVICE_CLASS_POWER
def __init__(self, client, name, typename):
def __init__(self, client: AuroraSerialClient, data, name, typename):
"""Initialize the sensor."""
self._attr_name = f"{name} {typename}"
self.client = client
super().__init__(client, data)
if typename == "instantaneouspower":
self.type = typename
self._attr_unit_of_measurement = POWER_WATT
self._attr_device_class = DEVICE_CLASS_POWER
elif typename == "temperature":
self.type = typename
self._attr_native_unit_of_measurement = TEMP_CELSIUS
self._attr_device_class = DEVICE_CLASS_TEMPERATURE
else:
raise InvalidStateError(f"Unrecognised typename '{typename}'")
self._attr_name = f"{name}"
self.availableprev = True
def update(self):
"""Fetch new state data for the sensor.
@ -65,11 +96,21 @@ class AuroraABBSolarPVMonitorSensor(SensorEntity):
This is the only method that should fetch new data for Home Assistant.
"""
try:
self.availableprev = self._attr_available
self.client.connect()
if self.type == "instantaneouspower":
# read ADC channel 3 (grid power output)
power_watts = self.client.measure(3, True)
self._attr_native_value = round(power_watts, 1)
self._attr_state = round(power_watts, 1)
elif self.type == "temperature":
temperature_c = self.client.measure(21)
self._attr_native_value = round(temperature_c, 1)
self._attr_available = True
except AuroraError as error:
self._attr_state = None
self._attr_native_value = None
self._attr_available = False
# aurorapy does not have different exceptions (yet) for dealing
# with timeout vs other comms errors.
# This means the (normal) situation of no response during darkness
@ -82,7 +123,14 @@ class AuroraABBSolarPVMonitorSensor(SensorEntity):
_LOGGER.debug("No response from inverter (could be dark)")
else:
raise error
self._attr_native_value = None
finally:
if self._attr_available != self.availableprev:
if self._attr_available:
_LOGGER.info("Communication with %s back online", self.name)
else:
_LOGGER.warning(
"Communication with %s lost",
self.name,
)
if self.client.serline.isOpen():
self.client.close()

23
homeassistant/components/aurora_abb_powerone/strings.json Normal file
View File

@ -0,0 +1,23 @@
{
"config": {
"step": {
"user": {
"description": "The inverter must be connected via an RS485 adaptor, please select serial port and the inverter's address as configured on the LCD panel",
"data": {
"port": "RS485 or USB-RS485 Adaptor Port",
"address": "Inverter Address"
}
}
},
"error": {
"cannot_connect": "Unable to connect, please check serial port, address, electrical connection and that inverter is on (in daylight)",
"invalid_serial_port": "Serial port is not a valid device or could not be openned",
"cannot_open_serial_port": "Cannot open serial port, please check and try again",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "Device is already configured",
"no_serial_ports": "No com ports found. Need a valid RS485 device to communicate."
}
}
}

23
homeassistant/components/aurora_abb_powerone/translations/en.json Normal file
View File

@ -0,0 +1,23 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured",
"no_serial_ports": "No com ports found. Need a valid RS485 device to communicate."
},
"error": {
"cannot_connect": "Unable to connect, please check serial port, address, electrical connection and that inverter is on (in daylight)",
"cannot_open_serial_port": "Cannot open serial port, please check and try again",
"invalid_serial_port": "Serial port is not a valid device or could not be openned",
"unknown": "Unexpected error"
},
"step": {
"user": {
"data": {
"address": "Inverter Address",
"port": "RS485 or USB-RS485 Adaptor Port"
},
"description": "The inverter must be connected via an RS485 adaptor, please select serial port and the inverter's address as configured on the LCD panel"
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -31,6 +31,7 @@ FLOWS = [
"atag",
"august",
"aurora",
"aurora_abb_powerone",
"awair",
"axis",
"azure_devops",

3
requirements_test_all.txt
View File

@ -235,6 +235,9 @@ async-upnp-client==0.22.10
# homeassistant.components.aurora
auroranoaa==0.0.2
# homeassistant.components.aurora_abb_powerone
aurorapy==0.2.6
# homeassistant.components.stream
av==8.0.3

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

@ -0,0 +1 @@
"""Tests for the Aurora ABB PowerOne Solar PV integration."""

165
tests/components/aurora_abb_powerone/test_config_flow.py Normal file
View File

@ -0,0 +1,165 @@
"""Test the Aurora ABB PowerOne Solar PV config flow."""
from logging import INFO
from unittest.mock import patch
from aurorapy.client import AuroraError
from serial.tools import list_ports_common
from homeassistant import config_entries, data_entry_flow, setup
from homeassistant.components.aurora_abb_powerone.const import (
ATTR_FIRMWARE,
ATTR_MODEL,
ATTR_SERIAL_NUMBER,
DOMAIN,
)
from homeassistant.const import CONF_ADDRESS, CONF_PORT
async def test_form(hass):
"""Test we get the form."""
await setup.async_setup_component(hass, "persistent_notification", {})
fakecomports = []
fakecomports.append(list_ports_common.ListPortInfo("/dev/ttyUSB7"))
with patch(
"serial.tools.list_ports.comports",
return_value=fakecomports,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["errors"] == {}
with patch("aurorapy.client.AuroraSerialClient.connect", return_value=None,), patch(
"aurorapy.client.AuroraSerialClient.serial_number",
return_value="9876543",
), patch(
"aurorapy.client.AuroraSerialClient.version",
return_value="9.8.7.6",
), patch(
"aurorapy.client.AuroraSerialClient.pn",
return_value="A.B.C",
), patch(
"aurorapy.client.AuroraSerialClient.firmware",
return_value="1.234",
), patch(
"homeassistant.components.aurora_abb_powerone.config_flow._LOGGER.getEffectiveLevel",
return_value=INFO,
) as mock_setup, patch(
"homeassistant.components.aurora_abb_powerone.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PORT: "/dev/ttyUSB7", CONF_ADDRESS: 7},
)
assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result2["data"] == {
CONF_PORT: "/dev/ttyUSB7",
CONF_ADDRESS: 7,
ATTR_FIRMWARE: "1.234",
ATTR_MODEL: "9.8.7.6 (A.B.C)",
ATTR_SERIAL_NUMBER: "9876543",
"title": "PhotoVoltaic Inverters",
}
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_no_comports(hass):
"""Test we display correct info when there are no com ports.."""
fakecomports = []
with patch(
"serial.tools.list_ports.comports",
return_value=fakecomports,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "abort"
assert result["reason"] == "no_serial_ports"
async def test_form_invalid_com_ports(hass):
"""Test we display correct info when the comport is invalid.."""
fakecomports = []
fakecomports.append(list_ports_common.ListPortInfo("/dev/ttyUSB7"))
with patch(
"serial.tools.list_ports.comports",
return_value=fakecomports,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["errors"] == {}
with patch(
"aurorapy.client.AuroraSerialClient.connect",
side_effect=OSError(19, "...no such device..."),
return_value=None,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PORT: "/dev/ttyUSB7", CONF_ADDRESS: 7},
)
assert result2["errors"] == {"base": "invalid_serial_port"}
with patch(
"aurorapy.client.AuroraSerialClient.connect",
side_effect=AuroraError("..could not open port..."),
return_value=None,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PORT: "/dev/ttyUSB7", CONF_ADDRESS: 7},
)
assert result2["errors"] == {"base": "cannot_open_serial_port"}
with patch(
"aurorapy.client.AuroraSerialClient.connect",
side_effect=AuroraError("...No response after..."),
return_value=None,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PORT: "/dev/ttyUSB7", CONF_ADDRESS: 7},
)
assert result2["errors"] == {"base": "cannot_connect"}
with patch(
"aurorapy.client.AuroraSerialClient.connect",
side_effect=AuroraError("...Some other message!!!123..."),
return_value=None,
), patch("serial.Serial.isOpen", return_value=True,), patch(
"aurorapy.client.AuroraSerialClient.close",
) as mock_clientclose:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PORT: "/dev/ttyUSB7", CONF_ADDRESS: 7},
)
assert result2["errors"] == {"base": "cannot_connect"}
assert len(mock_clientclose.mock_calls) == 1
# Tests below can be deleted after deprecation period is finished.
async def test_import(hass):
"""Test configuration.yaml import used during migration."""
TESTDATA = {"device": "/dev/ttyUSB7", "address": 3, "name": "MyAuroraPV"}
with patch(
"homeassistant.components.generic.camera.GenericCamera.async_camera_image",
return_value=None,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=TESTDATA
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"][CONF_PORT] == "/dev/ttyUSB7"
assert result["data"][CONF_ADDRESS] == 3

37
tests/components/aurora_abb_powerone/test_init.py Normal file
View File

@ -0,0 +1,37 @@
"""Pytest modules for Aurora ABB Powerone PV inverter sensor integration."""
from unittest.mock import patch
from homeassistant.components.aurora_abb_powerone.const import (
ATTR_FIRMWARE,
ATTR_MODEL,
ATTR_SERIAL_NUMBER,
DOMAIN,
)
from homeassistant.const import CONF_ADDRESS, CONF_PORT
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
async def test_unload_entry(hass):
"""Test unloading the aurora_abb_powerone entry."""
with patch("aurorapy.client.AuroraSerialClient.connect", return_value=None), patch(
"homeassistant.components.aurora_abb_powerone.sensor.AuroraSensor.update",
return_value=None,
):
mock_entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_PORT: "/dev/ttyUSB7",
CONF_ADDRESS: 7,
ATTR_MODEL: "model123",
ATTR_SERIAL_NUMBER: "876",
ATTR_FIRMWARE: "1.2.3.4",
},
)
mock_entry.add_to_hass(hass)
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
assert await hass.config_entries.async_unload(mock_entry.entry_id)
await hass.async_block_till_done()

185
tests/components/aurora_abb_powerone/test_sensor.py Normal file
View File

@ -0,0 +1,185 @@
"""Test the Aurora ABB PowerOne Solar PV sensors."""
from datetime import timedelta
from unittest.mock import patch
from aurorapy.client import AuroraError
import pytest
from homeassistant.components.aurora_abb_powerone.const import (
ATTR_DEVICE_NAME,
ATTR_FIRMWARE,
ATTR_MODEL,
ATTR_SERIAL_NUMBER,
DEFAULT_INTEGRATION_TITLE,
DOMAIN,
)
from homeassistant.components.aurora_abb_powerone.sensor import AuroraSensor
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import CONF_ADDRESS, CONF_PORT
from homeassistant.exceptions import InvalidStateError
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from tests.common import (
MockConfigEntry,
assert_setup_component,
async_fire_time_changed,
)
TEST_CONFIG = {
"sensor": {
"platform": "aurora_abb_powerone",
"device": "/dev/fakedevice0",
"address": 2,
}
}
def _simulated_returns(index, global_measure=None):
returns = {
3: 45.678, # power
21: 9.876, # temperature
}
return returns[index]
def _mock_config_entry():
return MockConfigEntry(
version=1,
domain=DOMAIN,
title=DEFAULT_INTEGRATION_TITLE,
data={
CONF_PORT: "/dev/usb999",
CONF_ADDRESS: 3,
ATTR_DEVICE_NAME: "mydevicename",
ATTR_MODEL: "mymodel",
ATTR_SERIAL_NUMBER: "123456",
ATTR_FIRMWARE: "1.2.3.4",
},
source="dummysource",
entry_id="13579",
)
async def test_setup_platform_valid_config(hass):
"""Test that (deprecated) yaml import still works."""
with patch("aurorapy.client.AuroraSerialClient.connect", return_value=None), patch(
"aurorapy.client.AuroraSerialClient.measure",
side_effect=_simulated_returns,
), assert_setup_component(1, "sensor"):
assert await async_setup_component(hass, "sensor", TEST_CONFIG)
await hass.async_block_till_done()
power = hass.states.get("sensor.power_output")
assert power
assert power.state == "45.7"
# try to set up a second time - should abort.
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=TEST_CONFIG,
context={"source": SOURCE_IMPORT},
)
assert result["type"] == "abort"
assert result["reason"] == "already_setup"
async def test_sensors(hass):
"""Test data coming back from inverter."""
mock_entry = _mock_config_entry()
with patch("aurorapy.client.AuroraSerialClient.connect", return_value=None), patch(
"aurorapy.client.AuroraSerialClient.measure",
side_effect=_simulated_returns,
):
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
power = hass.states.get("sensor.power_output")
assert power
assert power.state == "45.7"
temperature = hass.states.get("sensor.temperature")
assert temperature
assert temperature.state == "9.9"
async def test_sensor_invalid_type(hass):
"""Test invalid sensor type during setup."""
entities = []
mock_entry = _mock_config_entry()
with patch("aurorapy.client.AuroraSerialClient.connect", return_value=None), patch(
"aurorapy.client.AuroraSerialClient.measure",
side_effect=_simulated_returns,
):
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
client = hass.data[DOMAIN][mock_entry.unique_id]
data = mock_entry.data
with pytest.raises(InvalidStateError):
entities.append(AuroraSensor(client, data, "WrongSensor", "wrongparameter"))
async def test_sensor_dark(hass):
"""Test that darkness (no comms) is handled correctly."""
mock_entry = _mock_config_entry()
utcnow = dt_util.utcnow()
# sun is up
with patch("aurorapy.client.AuroraSerialClient.connect", return_value=None), patch(
"aurorapy.client.AuroraSerialClient.measure", side_effect=_simulated_returns
):
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
power = hass.states.get("sensor.power_output")
assert power is not None
assert power.state == "45.7"
# sunset
with patch("aurorapy.client.AuroraSerialClient.connect", return_value=None), patch(
"aurorapy.client.AuroraSerialClient.measure",
side_effect=AuroraError("No response after 10 seconds"),
):
async_fire_time_changed(hass, utcnow + timedelta(seconds=60))
await hass.async_block_till_done()
power = hass.states.get("sensor.power_output")
assert power.state == "unknown"
# sun rose again
with patch("aurorapy.client.AuroraSerialClient.connect", return_value=None), patch(
"aurorapy.client.AuroraSerialClient.measure", side_effect=_simulated_returns
):
async_fire_time_changed(hass, utcnow + timedelta(seconds=60))
await hass.async_block_till_done()
power = hass.states.get("sensor.power_output")
assert power is not None
assert power.state == "45.7"
# sunset
with patch("aurorapy.client.AuroraSerialClient.connect", return_value=None), patch(
"aurorapy.client.AuroraSerialClient.measure",
side_effect=AuroraError("No response after 10 seconds"),
):
async_fire_time_changed(hass, utcnow + timedelta(seconds=60))
await hass.async_block_till_done()
power = hass.states.get("sensor.power_output")
assert power.state == "unknown" # should this be 'available'?
async def test_sensor_unknown_error(hass):
"""Test other comms error is handled correctly."""
mock_entry = _mock_config_entry()
with patch("aurorapy.client.AuroraSerialClient.connect", return_value=None), patch(
"aurorapy.client.AuroraSerialClient.measure",
side_effect=AuroraError("another error"),
):
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
power = hass.states.get("sensor.power_output")
assert power is None