mirror of
https://github.com/home-assistant/core.git
synced 2025-07-23 13:17:32 +00:00
Sun WEG integration (#88272)
* feat(sunweg): initial support * chore: removed commented out code * chore: removed warning * fix: set never_resets for total sensors * test: some tests * fix(sunweg): default plantid type * fix(sunweg): return first plant id * test(sunweg): improved code coverage * chore(sunweg): missing FlowResult return type * chore(sunweg): removed unused strings * perf(sunweg): using only one api instance * chore(sunweg): removed uneeded atribute * refact(sunweg): small refactoring * refact(sunweg): typing * chore(sunweg): comments * chore(sunweg): bump version * chore(sunweg): bump lib version * test(sunweg): different mocking and coverage * test: fixed setup component parameter * feat: dynamic metrics * fix(sunweg): ruff * fix(sunweg): mypy * refact(sunweg): codereview suggestions * chore(sunweg): removed unused string * chore(sunweg): typehint and code formatting
This commit is contained in:
parent
906aa14b43
commit
f567bf6dfe
@ -1253,6 +1253,8 @@ build.json @home-assistant/supervisor
|
||||
/homeassistant/components/suez_water/ @ooii
|
||||
/homeassistant/components/sun/ @Swamp-Ig
|
||||
/tests/components/sun/ @Swamp-Ig
|
||||
/homeassistant/components/sunweg/ @rokam
|
||||
/tests/components/sunweg/ @rokam
|
||||
/homeassistant/components/supla/ @mwegrzynek
|
||||
/homeassistant/components/surepetcare/ @benleb @danielhiversen
|
||||
/tests/components/surepetcare/ @benleb @danielhiversen
|
||||
|
193
homeassistant/components/sunweg/__init__.py
Normal file
193
homeassistant/components/sunweg/__init__.py
Normal file
@ -0,0 +1,193 @@
|
||||
"""The Sun WEG inverter sensor integration."""
|
||||
import datetime
|
||||
import json
|
||||
import logging
|
||||
|
||||
from sunweg.api import APIHelper
|
||||
from sunweg.plant import Plant
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.typing import StateType
|
||||
from homeassistant.util import Throttle
|
||||
|
||||
from .const import CONF_PLANT_ID, DOMAIN, PLATFORMS
|
||||
from .sensor_types.sensor_entity_description import SunWEGSensorEntityDescription
|
||||
|
||||
SCAN_INTERVAL = datetime.timedelta(minutes=5)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant, entry: config_entries.ConfigEntry
|
||||
) -> bool:
|
||||
"""Load the saved entities."""
|
||||
api = APIHelper(entry.data[CONF_USERNAME], entry.data[CONF_PASSWORD])
|
||||
if not await hass.async_add_executor_job(api.authenticate):
|
||||
_LOGGER.error("Username or Password may be incorrect!")
|
||||
return False
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = SunWEGData(
|
||||
api, entry.data[CONF_PLANT_ID]
|
||||
)
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
if len(hass.data[DOMAIN]) == 0:
|
||||
hass.data.pop(DOMAIN)
|
||||
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
|
||||
|
||||
class SunWEGData:
|
||||
"""The class for handling data retrieval."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
api: APIHelper,
|
||||
plant_id: int,
|
||||
) -> None:
|
||||
"""Initialize the probe."""
|
||||
|
||||
self.api = api
|
||||
self.plant_id = plant_id
|
||||
self.data: Plant = None
|
||||
self.previous_values: dict = {}
|
||||
|
||||
@Throttle(SCAN_INTERVAL)
|
||||
def update(self) -> None:
|
||||
"""Update probe data."""
|
||||
_LOGGER.debug("Updating data for plant %s", self.plant_id)
|
||||
try:
|
||||
self.data = self.api.plant(self.plant_id)
|
||||
for inverter in self.data.inverters:
|
||||
self.api.complete_inverter(inverter)
|
||||
except json.decoder.JSONDecodeError:
|
||||
_LOGGER.error("Unable to fetch data from SunWEG server")
|
||||
_LOGGER.debug("Finished updating data for plant %s", self.plant_id)
|
||||
|
||||
def get_api_value(
|
||||
self,
|
||||
variable: str,
|
||||
device_type: str,
|
||||
inverter_id: int = 0,
|
||||
deep_name: str | None = None,
|
||||
):
|
||||
"""Retrieve from a Plant the desired variable value."""
|
||||
if device_type == "total":
|
||||
return self.data.__dict__.get(variable)
|
||||
|
||||
inverter_list = [i for i in self.data.inverters if i.id == inverter_id]
|
||||
if len(inverter_list) == 0:
|
||||
return None
|
||||
inverter = inverter_list[0]
|
||||
|
||||
if device_type == "inverter":
|
||||
return inverter.__dict__.get(variable)
|
||||
if device_type == "phase":
|
||||
for phase in inverter.phases:
|
||||
if phase.name == deep_name:
|
||||
return phase.__dict__.get(variable)
|
||||
elif device_type == "string":
|
||||
for mppt in inverter.mppts:
|
||||
for string in mppt.strings:
|
||||
if string.name == deep_name:
|
||||
return string.__dict__.get(variable)
|
||||
return None
|
||||
|
||||
def get_data(
|
||||
self,
|
||||
entity_description: SunWEGSensorEntityDescription,
|
||||
device_type: str,
|
||||
inverter_id: int = 0,
|
||||
deep_name: str | None = None,
|
||||
) -> StateType | datetime.datetime:
|
||||
"""Get the data."""
|
||||
_LOGGER.debug(
|
||||
"Data request for: %s",
|
||||
entity_description.name,
|
||||
)
|
||||
variable = entity_description.api_variable_key
|
||||
previous_metric = entity_description.native_unit_of_measurement
|
||||
api_value = self.get_api_value(variable, device_type, inverter_id, deep_name)
|
||||
previous_value = self.previous_values.get(variable)
|
||||
return_value = api_value
|
||||
if entity_description.api_variable_metric is not None:
|
||||
entity_description.native_unit_of_measurement = self.get_api_value(
|
||||
entity_description.api_variable_metric,
|
||||
device_type,
|
||||
inverter_id,
|
||||
deep_name,
|
||||
)
|
||||
|
||||
# If we have a 'drop threshold' specified, then check it and correct if needed
|
||||
if (
|
||||
entity_description.previous_value_drop_threshold is not None
|
||||
and previous_value is not None
|
||||
and api_value is not None
|
||||
and previous_metric == entity_description.native_unit_of_measurement
|
||||
):
|
||||
_LOGGER.debug(
|
||||
(
|
||||
"%s - Drop threshold specified (%s), checking for drop... API"
|
||||
" Value: %s, Previous Value: %s"
|
||||
),
|
||||
entity_description.name,
|
||||
entity_description.previous_value_drop_threshold,
|
||||
api_value,
|
||||
previous_value,
|
||||
)
|
||||
diff = float(api_value) - float(previous_value)
|
||||
|
||||
# Check if the value has dropped (negative value i.e. < 0) and it has only
|
||||
# dropped by a small amount, if so, use the previous value.
|
||||
# Note - The energy dashboard takes care of drops within 10%
|
||||
# of the current value, however if the value is low e.g. 0.2
|
||||
# and drops by 0.1 it classes as a reset.
|
||||
if -(entity_description.previous_value_drop_threshold) <= diff < 0:
|
||||
_LOGGER.debug(
|
||||
(
|
||||
"Diff is negative, but only by a small amount therefore not a"
|
||||
" nightly reset, using previous value (%s) instead of api value"
|
||||
" (%s)"
|
||||
),
|
||||
previous_value,
|
||||
api_value,
|
||||
)
|
||||
return_value = previous_value
|
||||
else:
|
||||
_LOGGER.debug(
|
||||
"%s - No drop detected, using API value", entity_description.name
|
||||
)
|
||||
|
||||
# Lifetime total values should always be increasing, they will never reset,
|
||||
# however the API sometimes returns 0 values when the clock turns to 00:00
|
||||
# local time in that scenario we should just return the previous value
|
||||
# Scenarios:
|
||||
# 1 - System has a genuine 0 value when it it first commissioned:
|
||||
# - will return 0 until a non-zero value is registered
|
||||
# 2 - System has been running fine but temporarily resets to 0 briefly
|
||||
# at midnight:
|
||||
# - will return the previous value
|
||||
# 3 - HA is restarted during the midnight 'outage' - Not handled:
|
||||
# - Previous value will not exist meaning 0 will be returned
|
||||
# - This is an edge case that would be better handled by looking
|
||||
# up the previous value of the entity from the recorder
|
||||
if entity_description.never_resets and api_value == 0 and previous_value:
|
||||
_LOGGER.debug(
|
||||
(
|
||||
"API value is 0, but this value should never reset, returning"
|
||||
" previous value (%s) instead"
|
||||
),
|
||||
previous_value,
|
||||
)
|
||||
return_value = previous_value
|
||||
|
||||
self.previous_values[variable] = return_value
|
||||
|
||||
return return_value
|
74
homeassistant/components/sunweg/config_flow.py
Normal file
74
homeassistant/components/sunweg/config_flow.py
Normal file
@ -0,0 +1,74 @@
|
||||
"""Config flow for Sun WEG integration."""
|
||||
from sunweg.api import APIHelper
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.const import CONF_NAME, CONF_PASSWORD, CONF_USERNAME
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
|
||||
from .const import CONF_PLANT_ID, DOMAIN
|
||||
|
||||
|
||||
class SunWEGConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Config flow class."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialise sun weg server flow."""
|
||||
self.api: APIHelper = None
|
||||
self.data: dict = {}
|
||||
|
||||
@callback
|
||||
def _async_show_user_form(self, errors=None) -> FlowResult:
|
||||
"""Show the form to the user."""
|
||||
data_schema = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_USERNAME): str,
|
||||
vol.Required(CONF_PASSWORD): str,
|
||||
}
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user", data_schema=data_schema, errors=errors
|
||||
)
|
||||
|
||||
async def async_step_user(self, user_input=None) -> FlowResult:
|
||||
"""Handle the start of the config flow."""
|
||||
if not user_input:
|
||||
return self._async_show_user_form()
|
||||
|
||||
# Initialise the library with the username & password
|
||||
self.api = APIHelper(user_input[CONF_USERNAME], user_input[CONF_PASSWORD])
|
||||
login_response = await self.hass.async_add_executor_job(self.api.authenticate)
|
||||
|
||||
if not login_response:
|
||||
return self._async_show_user_form({"base": "invalid_auth"})
|
||||
|
||||
# Store authentication info
|
||||
self.data = user_input
|
||||
return await self.async_step_plant()
|
||||
|
||||
async def async_step_plant(self, user_input=None) -> FlowResult:
|
||||
"""Handle adding a "plant" to Home Assistant."""
|
||||
plant_list = await self.hass.async_add_executor_job(self.api.listPlants)
|
||||
|
||||
if len(plant_list) == 0:
|
||||
return self.async_abort(reason="no_plants")
|
||||
|
||||
plants = {plant.id: plant.name for plant in plant_list}
|
||||
|
||||
if user_input is None and len(plant_list) > 1:
|
||||
data_schema = vol.Schema({vol.Required(CONF_PLANT_ID): vol.In(plants)})
|
||||
|
||||
return self.async_show_form(step_id="plant", data_schema=data_schema)
|
||||
|
||||
if user_input is None and len(plant_list) == 1:
|
||||
user_input = {CONF_PLANT_ID: plant_list[0].id}
|
||||
|
||||
user_input[CONF_NAME] = plants[user_input[CONF_PLANT_ID]]
|
||||
await self.async_set_unique_id(user_input[CONF_PLANT_ID])
|
||||
self._abort_if_unique_id_configured()
|
||||
self.data.update(user_input)
|
||||
return self.async_create_entry(title=self.data[CONF_NAME], data=self.data)
|
12
homeassistant/components/sunweg/const.py
Normal file
12
homeassistant/components/sunweg/const.py
Normal file
@ -0,0 +1,12 @@
|
||||
"""Define constants for the Sun WEG component."""
|
||||
from homeassistant.const import Platform
|
||||
|
||||
CONF_PLANT_ID = "plant_id"
|
||||
|
||||
DEFAULT_PLANT_ID = 0
|
||||
|
||||
DEFAULT_NAME = "Sun WEG"
|
||||
|
||||
DOMAIN = "sunweg"
|
||||
|
||||
PLATFORMS = [Platform.SENSOR]
|
10
homeassistant/components/sunweg/manifest.json
Normal file
10
homeassistant/components/sunweg/manifest.json
Normal file
@ -0,0 +1,10 @@
|
||||
{
|
||||
"domain": "sunweg",
|
||||
"name": "Sun WEG",
|
||||
"codeowners": ["@rokam"],
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/sunweg/",
|
||||
"iot_class": "cloud_polling",
|
||||
"loggers": ["sunweg"],
|
||||
"requirements": ["sunweg==2.0.0"]
|
||||
}
|
177
homeassistant/components/sunweg/sensor.py
Normal file
177
homeassistant/components/sunweg/sensor.py
Normal file
@ -0,0 +1,177 @@
|
||||
"""Read status of SunWEG inverters."""
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import logging
|
||||
from types import MappingProxyType
|
||||
from typing import Any
|
||||
|
||||
from sunweg.api import APIHelper
|
||||
from sunweg.device import Inverter
|
||||
from sunweg.plant import Plant
|
||||
|
||||
from homeassistant.components.sensor import SensorEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_NAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.device_registry import DeviceInfo
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.typing import StateType
|
||||
|
||||
from . import SunWEGData
|
||||
from .const import CONF_PLANT_ID, DEFAULT_PLANT_ID, DOMAIN
|
||||
from .sensor_types.inverter import INVERTER_SENSOR_TYPES
|
||||
from .sensor_types.phase import PHASE_SENSOR_TYPES
|
||||
from .sensor_types.sensor_entity_description import SunWEGSensorEntityDescription
|
||||
from .sensor_types.string import STRING_SENSOR_TYPES
|
||||
from .sensor_types.total import TOTAL_SENSOR_TYPES
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def get_device_list(
|
||||
api: APIHelper, config: MappingProxyType[str, Any]
|
||||
) -> tuple[list[Inverter], int]:
|
||||
"""Retrieve the device list for the selected plant."""
|
||||
plant_id = int(config[CONF_PLANT_ID])
|
||||
|
||||
if plant_id == DEFAULT_PLANT_ID:
|
||||
plant_info: list[Plant] = api.listPlants()
|
||||
plant_id = plant_info[0].id
|
||||
|
||||
devices: list[Inverter] = []
|
||||
# Get a list of devices for specified plant to add sensors for.
|
||||
for inverter in api.plant(plant_id).inverters:
|
||||
api.complete_inverter(inverter)
|
||||
devices.append(inverter)
|
||||
return (devices, plant_id)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
config_entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the SunWEG sensor."""
|
||||
name = config_entry.data[CONF_NAME]
|
||||
|
||||
probe: SunWEGData = hass.data[DOMAIN][config_entry.entry_id]
|
||||
|
||||
devices, plant_id = await hass.async_add_executor_job(
|
||||
get_device_list, probe.api, config_entry.data
|
||||
)
|
||||
|
||||
entities = [
|
||||
SunWEGInverter(
|
||||
probe,
|
||||
name=f"{name} Total",
|
||||
unique_id=f"{plant_id}-{description.key}",
|
||||
description=description,
|
||||
device_type="total",
|
||||
)
|
||||
for description in TOTAL_SENSOR_TYPES
|
||||
]
|
||||
|
||||
# Add sensors for each device in the specified plant.
|
||||
entities.extend(
|
||||
[
|
||||
SunWEGInverter(
|
||||
probe,
|
||||
name=f"{device.name}",
|
||||
unique_id=f"{device.sn}-{description.key}",
|
||||
description=description,
|
||||
device_type="inverter",
|
||||
inverter_id=device.id,
|
||||
)
|
||||
for device in devices
|
||||
for description in INVERTER_SENSOR_TYPES
|
||||
]
|
||||
)
|
||||
|
||||
entities.extend(
|
||||
[
|
||||
SunWEGInverter(
|
||||
probe,
|
||||
name=f"{device.name} {phase.name}",
|
||||
unique_id=f"{device.sn}-{phase.name}-{description.key}",
|
||||
description=description,
|
||||
inverter_id=device.id,
|
||||
device_type="phase",
|
||||
deep_name=phase.name,
|
||||
)
|
||||
for device in devices
|
||||
for phase in device.phases
|
||||
for description in PHASE_SENSOR_TYPES
|
||||
]
|
||||
)
|
||||
|
||||
entities.extend(
|
||||
[
|
||||
SunWEGInverter(
|
||||
probe,
|
||||
name=f"{device.name} {string.name}",
|
||||
unique_id=f"{device.sn}-{string.name}-{description.key}",
|
||||
description=description,
|
||||
inverter_id=device.id,
|
||||
device_type="string",
|
||||
deep_name=string.name,
|
||||
)
|
||||
for device in devices
|
||||
for mppt in device.mppts
|
||||
for string in mppt.strings
|
||||
for description in STRING_SENSOR_TYPES
|
||||
]
|
||||
)
|
||||
|
||||
async_add_entities(entities, True)
|
||||
|
||||
|
||||
class SunWEGInverter(SensorEntity):
|
||||
"""Representation of a SunWEG Sensor."""
|
||||
|
||||
entity_description: SunWEGSensorEntityDescription
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
probe: SunWEGData,
|
||||
name: str,
|
||||
unique_id: str,
|
||||
description: SunWEGSensorEntityDescription,
|
||||
device_type: str,
|
||||
inverter_id: int = 0,
|
||||
deep_name: str | None = None,
|
||||
) -> None:
|
||||
"""Initialize a sensor."""
|
||||
self.probe = probe
|
||||
self.entity_description = description
|
||||
self.device_type = device_type
|
||||
self.inverter_id = inverter_id
|
||||
self.deep_name = deep_name
|
||||
|
||||
self._attr_name = f"{name} {description.name}"
|
||||
self._attr_unique_id = unique_id
|
||||
self._attr_icon = (
|
||||
description.icon if description.icon is not None else "mdi:solar-power"
|
||||
)
|
||||
|
||||
self._attr_device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, str(probe.plant_id))},
|
||||
manufacturer="SunWEG",
|
||||
name=name,
|
||||
)
|
||||
|
||||
@property
|
||||
def native_value(
|
||||
self,
|
||||
) -> StateType | datetime.datetime:
|
||||
"""Return the state of the sensor."""
|
||||
return self.probe.get_data(
|
||||
self.entity_description,
|
||||
device_type=self.device_type,
|
||||
inverter_id=self.inverter_id,
|
||||
deep_name=self.deep_name,
|
||||
)
|
||||
|
||||
def update(self) -> None:
|
||||
"""Get the latest data from the Sun WEG API and updates the state."""
|
||||
self.probe.update()
|
1
homeassistant/components/sunweg/sensor_types/__init__.py
Normal file
1
homeassistant/components/sunweg/sensor_types/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
"""Sensor types for supported Sun WEG systems."""
|
69
homeassistant/components/sunweg/sensor_types/inverter.py
Normal file
69
homeassistant/components/sunweg/sensor_types/inverter.py
Normal file
@ -0,0 +1,69 @@
|
||||
"""SunWEG Sensor definitions for the Inverter type."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass
|
||||
from homeassistant.const import (
|
||||
UnitOfEnergy,
|
||||
UnitOfFrequency,
|
||||
UnitOfPower,
|
||||
UnitOfTemperature,
|
||||
)
|
||||
|
||||
from .sensor_entity_description import SunWEGSensorEntityDescription
|
||||
|
||||
INVERTER_SENSOR_TYPES: tuple[SunWEGSensorEntityDescription, ...] = (
|
||||
SunWEGSensorEntityDescription(
|
||||
key="inverter_energy_today",
|
||||
name="Energy today",
|
||||
api_variable_key="_today_energy",
|
||||
api_variable_metric="_today_energy_metric",
|
||||
native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
|
||||
device_class=SensorDeviceClass.ENERGY,
|
||||
state_class=SensorStateClass.TOTAL_INCREASING,
|
||||
suggested_display_precision=1,
|
||||
),
|
||||
SunWEGSensorEntityDescription(
|
||||
key="inverter_energy_total",
|
||||
name="Lifetime energy output",
|
||||
api_variable_key="_total_energy",
|
||||
api_variable_metric="_total_energy_metric",
|
||||
native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
|
||||
device_class=SensorDeviceClass.ENERGY,
|
||||
suggested_display_precision=1,
|
||||
state_class=SensorStateClass.TOTAL,
|
||||
never_resets=True,
|
||||
),
|
||||
SunWEGSensorEntityDescription(
|
||||
key="inverter_frequency",
|
||||
name="AC frequency",
|
||||
api_variable_key="_frequency",
|
||||
native_unit_of_measurement=UnitOfFrequency.HERTZ,
|
||||
device_class=SensorDeviceClass.FREQUENCY,
|
||||
suggested_display_precision=1,
|
||||
),
|
||||
SunWEGSensorEntityDescription(
|
||||
key="inverter_current_wattage",
|
||||
name="Output power",
|
||||
api_variable_key="_power",
|
||||
api_variable_metric="_power_metric",
|
||||
native_unit_of_measurement=UnitOfPower.WATT,
|
||||
device_class=SensorDeviceClass.POWER,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=1,
|
||||
),
|
||||
SunWEGSensorEntityDescription(
|
||||
key="inverter_temperature",
|
||||
name="Temperature",
|
||||
api_variable_key="_temperature",
|
||||
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
|
||||
device_class=SensorDeviceClass.TEMPERATURE,
|
||||
icon="mdi:temperature-celsius",
|
||||
suggested_display_precision=1,
|
||||
),
|
||||
SunWEGSensorEntityDescription(
|
||||
key="inverter_power_factor",
|
||||
name="Power Factor",
|
||||
api_variable_key="_power_factor",
|
||||
suggested_display_precision=1,
|
||||
),
|
||||
)
|
26
homeassistant/components/sunweg/sensor_types/phase.py
Normal file
26
homeassistant/components/sunweg/sensor_types/phase.py
Normal file
@ -0,0 +1,26 @@
|
||||
"""SunWEG Sensor definitions for the Phase type."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.components.sensor import SensorDeviceClass
|
||||
from homeassistant.const import UnitOfElectricCurrent, UnitOfElectricPotential
|
||||
|
||||
from .sensor_entity_description import SunWEGSensorEntityDescription
|
||||
|
||||
PHASE_SENSOR_TYPES: tuple[SunWEGSensorEntityDescription, ...] = (
|
||||
SunWEGSensorEntityDescription(
|
||||
key="voltage",
|
||||
name="Voltage",
|
||||
api_variable_key="_voltage",
|
||||
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
|
||||
device_class=SensorDeviceClass.VOLTAGE,
|
||||
suggested_display_precision=2,
|
||||
),
|
||||
SunWEGSensorEntityDescription(
|
||||
key="amperage",
|
||||
name="Amperage",
|
||||
api_variable_key="_amperage",
|
||||
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
|
||||
device_class=SensorDeviceClass.CURRENT,
|
||||
suggested_display_precision=1,
|
||||
),
|
||||
)
|
@ -0,0 +1,23 @@
|
||||
"""Sensor Entity Description for the SunWEG integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
from homeassistant.components.sensor import SensorEntityDescription
|
||||
|
||||
|
||||
@dataclass
|
||||
class SunWEGRequiredKeysMixin:
|
||||
"""Mixin for required keys."""
|
||||
|
||||
api_variable_key: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class SunWEGSensorEntityDescription(SensorEntityDescription, SunWEGRequiredKeysMixin):
|
||||
"""Describes SunWEG sensor entity."""
|
||||
|
||||
api_variable_metric: str | None = None
|
||||
previous_value_drop_threshold: float | None = None
|
||||
never_resets: bool = False
|
||||
icon: str | None = None
|
26
homeassistant/components/sunweg/sensor_types/string.py
Normal file
26
homeassistant/components/sunweg/sensor_types/string.py
Normal file
@ -0,0 +1,26 @@
|
||||
"""SunWEG Sensor definitions for the String type."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.components.sensor import SensorDeviceClass
|
||||
from homeassistant.const import UnitOfElectricCurrent, UnitOfElectricPotential
|
||||
|
||||
from .sensor_entity_description import SunWEGSensorEntityDescription
|
||||
|
||||
STRING_SENSOR_TYPES: tuple[SunWEGSensorEntityDescription, ...] = (
|
||||
SunWEGSensorEntityDescription(
|
||||
key="voltage",
|
||||
name="Voltage",
|
||||
api_variable_key="_voltage",
|
||||
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
|
||||
device_class=SensorDeviceClass.VOLTAGE,
|
||||
suggested_display_precision=2,
|
||||
),
|
||||
SunWEGSensorEntityDescription(
|
||||
key="amperage",
|
||||
name="Amperage",
|
||||
api_variable_key="_amperage",
|
||||
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
|
||||
device_class=SensorDeviceClass.CURRENT,
|
||||
suggested_display_precision=1,
|
||||
),
|
||||
)
|
54
homeassistant/components/sunweg/sensor_types/total.py
Normal file
54
homeassistant/components/sunweg/sensor_types/total.py
Normal file
@ -0,0 +1,54 @@
|
||||
"""SunWEG Sensor definitions for Totals."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass
|
||||
from homeassistant.const import UnitOfEnergy, UnitOfPower
|
||||
|
||||
from .sensor_entity_description import SunWEGSensorEntityDescription
|
||||
|
||||
TOTAL_SENSOR_TYPES: tuple[SunWEGSensorEntityDescription, ...] = (
|
||||
SunWEGSensorEntityDescription(
|
||||
key="total_money_total",
|
||||
name="Money lifetime",
|
||||
api_variable_key="_saving",
|
||||
icon="mdi:cash",
|
||||
native_unit_of_measurement="R$",
|
||||
suggested_display_precision=2,
|
||||
),
|
||||
SunWEGSensorEntityDescription(
|
||||
key="total_energy_today",
|
||||
name="Energy Today",
|
||||
api_variable_key="_today_energy",
|
||||
api_variable_metric="_today_energy_metric",
|
||||
native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
|
||||
device_class=SensorDeviceClass.ENERGY,
|
||||
state_class=SensorStateClass.TOTAL_INCREASING,
|
||||
),
|
||||
SunWEGSensorEntityDescription(
|
||||
key="total_output_power",
|
||||
name="Output Power",
|
||||
api_variable_key="_total_power",
|
||||
native_unit_of_measurement=UnitOfPower.KILO_WATT,
|
||||
device_class=SensorDeviceClass.POWER,
|
||||
),
|
||||
SunWEGSensorEntityDescription(
|
||||
key="total_energy_output",
|
||||
name="Lifetime energy output",
|
||||
api_variable_key="_total_energy",
|
||||
native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
|
||||
device_class=SensorDeviceClass.ENERGY,
|
||||
state_class=SensorStateClass.TOTAL,
|
||||
never_resets=True,
|
||||
),
|
||||
SunWEGSensorEntityDescription(
|
||||
key="kwh_per_kwp",
|
||||
name="kWh por kWp",
|
||||
api_variable_key="_kwh_per_kwp",
|
||||
),
|
||||
SunWEGSensorEntityDescription(
|
||||
key="last_update",
|
||||
name="Last Update",
|
||||
api_variable_key="_last_update",
|
||||
device_class=SensorDeviceClass.DATE,
|
||||
),
|
||||
)
|
25
homeassistant/components/sunweg/strings.json
Normal file
25
homeassistant/components/sunweg/strings.json
Normal file
@ -0,0 +1,25 @@
|
||||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"no_plants": "No plants have been found on this account"
|
||||
},
|
||||
"error": {
|
||||
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
|
||||
},
|
||||
"step": {
|
||||
"plant": {
|
||||
"data": {
|
||||
"plant_id": "Plant"
|
||||
},
|
||||
"title": "Select your plant"
|
||||
},
|
||||
"user": {
|
||||
"data": {
|
||||
"password": "[%key:common::config_flow::data::password%]",
|
||||
"username": "[%key:common::config_flow::data::username%]"
|
||||
},
|
||||
"title": "Enter your Sun WEG information"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
@ -472,6 +472,7 @@ FLOWS = {
|
||||
"stookwijzer",
|
||||
"subaru",
|
||||
"sun",
|
||||
"sunweg",
|
||||
"surepetcare",
|
||||
"switchbee",
|
||||
"switchbot",
|
||||
|
@ -5545,6 +5545,12 @@
|
||||
"config_flow": true,
|
||||
"iot_class": "calculated"
|
||||
},
|
||||
"sunweg": {
|
||||
"name": "Sun WEG",
|
||||
"integration_type": "hub",
|
||||
"config_flow": true,
|
||||
"iot_class": "cloud_polling"
|
||||
},
|
||||
"supervisord": {
|
||||
"name": "Supervisord",
|
||||
"integration_type": "hub",
|
||||
|
@ -2552,6 +2552,9 @@ subarulink==0.7.9
|
||||
# homeassistant.components.solarlog
|
||||
sunwatcher==0.2.1
|
||||
|
||||
# homeassistant.components.sunweg
|
||||
sunweg==2.0.0
|
||||
|
||||
# homeassistant.components.surepetcare
|
||||
surepy==0.8.0
|
||||
|
||||
|
@ -1913,6 +1913,9 @@ subarulink==0.7.9
|
||||
# homeassistant.components.solarlog
|
||||
sunwatcher==0.2.1
|
||||
|
||||
# homeassistant.components.sunweg
|
||||
sunweg==2.0.0
|
||||
|
||||
# homeassistant.components.surepetcare
|
||||
surepy==0.8.0
|
||||
|
||||
|
1
tests/components/sunweg/__init__.py
Normal file
1
tests/components/sunweg/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
"""Tests for the sunweg component."""
|
63
tests/components/sunweg/common.py
Normal file
63
tests/components/sunweg/common.py
Normal file
@ -0,0 +1,63 @@
|
||||
"""Common functions needed to setup tests for Sun WEG."""
|
||||
from datetime import datetime
|
||||
|
||||
from sunweg.device import MPPT, Inverter, Phase, String
|
||||
from sunweg.plant import Plant
|
||||
|
||||
from homeassistant.components.sunweg.const import CONF_PLANT_ID, DOMAIN
|
||||
from homeassistant.const import CONF_NAME, CONF_PASSWORD, CONF_USERNAME
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
FIXTURE_USER_INPUT = {
|
||||
CONF_USERNAME: "username",
|
||||
CONF_PASSWORD: "password",
|
||||
}
|
||||
|
||||
SUNWEG_PLANT_RESPONSE = Plant(
|
||||
123456,
|
||||
"Plant #123",
|
||||
29.5,
|
||||
0.5,
|
||||
0,
|
||||
12.786912,
|
||||
24.0,
|
||||
"kWh",
|
||||
332.2,
|
||||
0.012296,
|
||||
datetime(2023, 2, 16, 14, 22, 37),
|
||||
)
|
||||
|
||||
SUNWEG_INVERTER_RESPONSE = Inverter(
|
||||
21255,
|
||||
"INVERSOR01",
|
||||
"J63T233018RE074",
|
||||
23.2,
|
||||
0.0,
|
||||
0.0,
|
||||
"MWh",
|
||||
0,
|
||||
"kWh",
|
||||
0.0,
|
||||
1,
|
||||
0,
|
||||
"kW",
|
||||
)
|
||||
|
||||
SUNWEG_PHASE_RESPONSE = Phase("PhaseA", 120.0, 3.2, 0, 0)
|
||||
|
||||
SUNWEG_MPPT_RESPONSE = MPPT("MPPT1")
|
||||
|
||||
SUNWEG_STRING_RESPONSE = String("STR1", 450.3, 23.4, 0)
|
||||
|
||||
SUNWEG_LOGIN_RESPONSE = True
|
||||
|
||||
SUNWEG_MOCK_ENTRY = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={
|
||||
CONF_USERNAME: "user@email.com",
|
||||
CONF_PASSWORD: "password",
|
||||
CONF_PLANT_ID: 0,
|
||||
CONF_NAME: "Name",
|
||||
},
|
||||
)
|
135
tests/components/sunweg/test_config_flow.py
Normal file
135
tests/components/sunweg/test_config_flow.py
Normal file
@ -0,0 +1,135 @@
|
||||
"""Tests for the Sun WEG server config flow."""
|
||||
from copy import deepcopy
|
||||
from unittest.mock import patch
|
||||
|
||||
from sunweg.api import APIHelper
|
||||
|
||||
from homeassistant import config_entries, data_entry_flow
|
||||
from homeassistant.components.sunweg.const import CONF_PLANT_ID, DOMAIN
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .common import FIXTURE_USER_INPUT, SUNWEG_LOGIN_RESPONSE, SUNWEG_PLANT_RESPONSE
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_show_authenticate_form(hass: HomeAssistant) -> None:
|
||||
"""Test that the setup form is served."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
|
||||
async def test_incorrect_login(hass: HomeAssistant) -> None:
|
||||
"""Test that it shows the appropriate error when an incorrect username/password/server is entered."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch.object(APIHelper, "authenticate", return_value=False):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], FIXTURE_USER_INPUT
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {"base": "invalid_auth"}
|
||||
|
||||
|
||||
async def test_no_plants_on_account(hass: HomeAssistant) -> None:
|
||||
"""Test registering an integration and finishing flow with an entered plant_id."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
user_input = FIXTURE_USER_INPUT.copy()
|
||||
|
||||
with patch.object(
|
||||
APIHelper, "authenticate", return_value=SUNWEG_LOGIN_RESPONSE
|
||||
), patch.object(APIHelper, "listPlants", return_value=[]):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input
|
||||
)
|
||||
|
||||
assert result["type"] == "abort"
|
||||
assert result["reason"] == "no_plants"
|
||||
|
||||
|
||||
async def test_multiple_plant_ids(hass: HomeAssistant) -> None:
|
||||
"""Test registering an integration and finishing flow with an entered plant_id."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
user_input = FIXTURE_USER_INPUT.copy()
|
||||
plant_list = [deepcopy(SUNWEG_PLANT_RESPONSE), deepcopy(SUNWEG_PLANT_RESPONSE)]
|
||||
|
||||
with patch.object(
|
||||
APIHelper, "authenticate", return_value=SUNWEG_LOGIN_RESPONSE
|
||||
), patch.object(APIHelper, "listPlants", return_value=plant_list):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input
|
||||
)
|
||||
assert result["type"] == data_entry_flow.FlowResultType.FORM
|
||||
assert result["step_id"] == "plant"
|
||||
|
||||
user_input = {CONF_PLANT_ID: 123456}
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
|
||||
assert result["data"][CONF_USERNAME] == FIXTURE_USER_INPUT[CONF_USERNAME]
|
||||
assert result["data"][CONF_PASSWORD] == FIXTURE_USER_INPUT[CONF_PASSWORD]
|
||||
assert result["data"][CONF_PLANT_ID] == 123456
|
||||
|
||||
|
||||
async def test_one_plant_on_account(hass: HomeAssistant) -> None:
|
||||
"""Test registering an integration and finishing flow with an entered plant_id."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
user_input = FIXTURE_USER_INPUT.copy()
|
||||
|
||||
with patch.object(
|
||||
APIHelper, "authenticate", return_value=SUNWEG_LOGIN_RESPONSE
|
||||
), patch.object(
|
||||
APIHelper,
|
||||
"listPlants",
|
||||
return_value=[deepcopy(SUNWEG_PLANT_RESPONSE)],
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
|
||||
assert result["data"][CONF_USERNAME] == FIXTURE_USER_INPUT[CONF_USERNAME]
|
||||
assert result["data"][CONF_PASSWORD] == FIXTURE_USER_INPUT[CONF_PASSWORD]
|
||||
assert result["data"][CONF_PLANT_ID] == 123456
|
||||
|
||||
|
||||
async def test_existing_plant_configured(hass: HomeAssistant) -> None:
|
||||
"""Test entering an existing plant_id."""
|
||||
entry = MockConfigEntry(domain=DOMAIN, unique_id=123456)
|
||||
entry.add_to_hass(hass)
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
user_input = FIXTURE_USER_INPUT.copy()
|
||||
|
||||
with patch.object(
|
||||
APIHelper, "authenticate", return_value=SUNWEG_LOGIN_RESPONSE
|
||||
), patch.object(
|
||||
APIHelper,
|
||||
"listPlants",
|
||||
return_value=[deepcopy(SUNWEG_PLANT_RESPONSE)],
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input
|
||||
)
|
||||
|
||||
assert result["type"] == "abort"
|
||||
assert result["reason"] == "already_configured"
|
146
tests/components/sunweg/test_init.py
Normal file
146
tests/components/sunweg/test_init.py
Normal file
@ -0,0 +1,146 @@
|
||||
"""Tests for the Sun WEG init."""
|
||||
|
||||
from copy import deepcopy
|
||||
import json
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from sunweg.api import APIHelper
|
||||
from sunweg.device import MPPT, Inverter
|
||||
from sunweg.plant import Plant
|
||||
|
||||
from homeassistant.components.sunweg import SunWEGData
|
||||
from homeassistant.components.sunweg.const import DOMAIN
|
||||
from homeassistant.components.sunweg.sensor_types.sensor_entity_description import (
|
||||
SunWEGSensorEntityDescription,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from .common import (
|
||||
SUNWEG_INVERTER_RESPONSE,
|
||||
SUNWEG_LOGIN_RESPONSE,
|
||||
SUNWEG_MOCK_ENTRY,
|
||||
SUNWEG_MPPT_RESPONSE,
|
||||
SUNWEG_PHASE_RESPONSE,
|
||||
SUNWEG_PLANT_RESPONSE,
|
||||
SUNWEG_STRING_RESPONSE,
|
||||
)
|
||||
|
||||
|
||||
async def test_methods(hass: HomeAssistant) -> None:
|
||||
"""Test methods."""
|
||||
mock_entry = SUNWEG_MOCK_ENTRY
|
||||
mock_entry.add_to_hass(hass)
|
||||
mppt: MPPT = deepcopy(SUNWEG_MPPT_RESPONSE)
|
||||
mppt.strings.append(SUNWEG_STRING_RESPONSE)
|
||||
inverter: Inverter = deepcopy(SUNWEG_INVERTER_RESPONSE)
|
||||
inverter.phases.append(SUNWEG_PHASE_RESPONSE)
|
||||
inverter.mppts.append(mppt)
|
||||
plant: Plant = deepcopy(SUNWEG_PLANT_RESPONSE)
|
||||
plant.inverters.append(inverter)
|
||||
|
||||
with patch.object(
|
||||
APIHelper, "authenticate", return_value=SUNWEG_LOGIN_RESPONSE
|
||||
), patch.object(APIHelper, "listPlants", return_value=[plant]), patch.object(
|
||||
APIHelper, "plant", return_value=plant
|
||||
), patch.object(
|
||||
APIHelper, "inverter", return_value=inverter
|
||||
), patch.object(
|
||||
APIHelper, "complete_inverter"
|
||||
):
|
||||
assert await async_setup_component(hass, DOMAIN, mock_entry.data)
|
||||
await hass.async_block_till_done()
|
||||
assert await hass.config_entries.async_unload(mock_entry.entry_id)
|
||||
|
||||
|
||||
async def test_setup_wrongpass(hass: HomeAssistant) -> None:
|
||||
"""Test setup with wrong pass."""
|
||||
mock_entry = SUNWEG_MOCK_ENTRY
|
||||
mock_entry.add_to_hass(hass)
|
||||
with patch.object(APIHelper, "authenticate", return_value=False):
|
||||
assert await async_setup_component(hass, DOMAIN, mock_entry.data)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
async def test_sunwegdata_update_exception() -> None:
|
||||
"""Test SunWEGData exception on update."""
|
||||
api = MagicMock()
|
||||
api.plant = MagicMock(side_effect=json.decoder.JSONDecodeError("Message", "Doc", 1))
|
||||
data = SunWEGData(api, 0)
|
||||
data.update()
|
||||
assert data.data is None
|
||||
|
||||
|
||||
async def test_sunwegdata_update_success() -> None:
|
||||
"""Test SunWEGData success on update."""
|
||||
inverter: Inverter = deepcopy(SUNWEG_INVERTER_RESPONSE)
|
||||
plant: Plant = deepcopy(SUNWEG_PLANT_RESPONSE)
|
||||
plant.inverters.append(inverter)
|
||||
api = MagicMock()
|
||||
api.plant = MagicMock(return_value=plant)
|
||||
api.complete_inverter = MagicMock()
|
||||
data = SunWEGData(api, 0)
|
||||
data.update()
|
||||
assert data.data.id == plant.id
|
||||
assert data.data.name == plant.name
|
||||
assert data.data.kwh_per_kwp == plant.kwh_per_kwp
|
||||
assert data.data.last_update == plant.last_update
|
||||
assert data.data.performance_rate == plant.performance_rate
|
||||
assert data.data.saving == plant.saving
|
||||
assert len(data.data.inverters) == 1
|
||||
|
||||
|
||||
async def test_sunwegdata_get_api_value_none() -> None:
|
||||
"""Test SunWEGData none return on get_api_value."""
|
||||
api = MagicMock()
|
||||
data = SunWEGData(api, 123456)
|
||||
data.data = deepcopy(SUNWEG_PLANT_RESPONSE)
|
||||
assert data.get_api_value("variable", "inverter", 0, "deep_name") is None
|
||||
data.data.inverters.append(deepcopy(SUNWEG_INVERTER_RESPONSE))
|
||||
assert data.get_api_value("variable", "invalid type", 21255, "deep_name") is None
|
||||
|
||||
|
||||
async def test_sunwegdata_get_data_drop_threshold() -> None:
|
||||
"""Test SunWEGData get_data with drop threshold."""
|
||||
api = MagicMock()
|
||||
data = SunWEGData(api, 123456)
|
||||
data.get_api_value = MagicMock()
|
||||
entity_description = SunWEGSensorEntityDescription(
|
||||
api_variable_key="variable", key="key"
|
||||
)
|
||||
entity_description.previous_value_drop_threshold = 0.1
|
||||
data.get_api_value.return_value = 3.0
|
||||
assert (
|
||||
data.get_data(entity_description=entity_description, device_type="total") == 3.0
|
||||
)
|
||||
data.get_api_value.return_value = 2.91
|
||||
assert (
|
||||
data.get_data(entity_description=entity_description, device_type="total") == 3.0
|
||||
)
|
||||
data.get_api_value.return_value = 2.8
|
||||
assert (
|
||||
data.get_data(entity_description=entity_description, device_type="total") == 2.8
|
||||
)
|
||||
|
||||
|
||||
async def test_sunwegdata_get_data_never_reset() -> None:
|
||||
"""Test SunWEGData get_data with never reset."""
|
||||
api = MagicMock()
|
||||
data = SunWEGData(api, 123456)
|
||||
data.get_api_value = MagicMock()
|
||||
entity_description = SunWEGSensorEntityDescription(
|
||||
api_variable_key="variable", key="key"
|
||||
)
|
||||
entity_description.never_resets = True
|
||||
data.get_api_value.return_value = 3.0
|
||||
assert (
|
||||
data.get_data(entity_description=entity_description, device_type="total") == 3.0
|
||||
)
|
||||
data.get_api_value.return_value = 0
|
||||
assert (
|
||||
data.get_data(entity_description=entity_description, device_type="total") == 3.0
|
||||
)
|
||||
data.get_api_value.return_value = 2.8
|
||||
assert (
|
||||
data.get_data(entity_description=entity_description, device_type="total") == 2.8
|
||||
)
|
Loading…
x
Reference in New Issue
Block a user