jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Drop IoTaWatt Accumulated sensors (#86611)

Browse Source
This commit is contained in:
Stefan Agner 2023-01-25 11:08:43 +01:00 committed by GitHub
parent d0b67689e0
commit f3cf760772
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 3 additions and 278 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/iotawatt/const.py
View File

@ -9,6 +9,4 @@ DOMAIN = "iotawatt"
VOLT_AMPERE_REACTIVE = "VAR"
VOLT_AMPERE_REACTIVE_HOURS = "VARh"
ATTR_LAST_UPDATE = "last_update"
CONNECTION_ERRORS = (KeyError, json.JSONDecodeError, httpx.HTTPError)

86
homeassistant/components/iotawatt/sensor.py
View File

@ -27,17 +27,11 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity, entity_registry
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.util import dt
from .const import (
ATTR_LAST_UPDATE,
DOMAIN,
VOLT_AMPERE_REACTIVE,
VOLT_AMPERE_REACTIVE_HOURS,
)
from .const import DOMAIN, VOLT_AMPERE_REACTIVE, VOLT_AMPERE_REACTIVE_HOURS
from .coordinator import IotawattUpdater
_LOGGER = logging.getLogger(__name__)
@ -134,10 +128,6 @@ async def async_setup_entry(
description = ENTITY_DESCRIPTION_KEY_MAP.get(
data.getUnit(), IotaWattSensorEntityDescription("base_sensor")
)
if data.getUnit() == "WattHours" and not data.getFromStart():
return IotaWattAccumulatingSensor(
coordinator=coordinator, key=key, entity_description=description
)
return IotaWattSensor(
coordinator=coordinator,
@ -235,77 +225,3 @@ class IotaWattSensor(CoordinatorEntity[IotawattUpdater], SensorEntity):
return func(self._sensor_data.getValue())
return self._sensor_data.getValue()
class IotaWattAccumulatingSensor(IotaWattSensor, RestoreEntity):
"""Defines a IoTaWatt Accumulative Energy (High Accuracy) Sensor."""
def __init__(
self,
coordinator: IotawattUpdater,
key: str,
entity_description: IotaWattSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, key, entity_description)
if self._attr_unique_id is not None:
self._attr_unique_id += ".accumulated"
self._accumulated_value: float | None = None
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
assert (
self._accumulated_value is not None
), "async_added_to_hass must have been called first"
self._accumulated_value += float(self._sensor_data.getValue())
super()._handle_coordinator_update()
@property
def native_value(self) -> StateType:
"""Return the state of the sensor."""
if self._accumulated_value is None:
return None
return round(self._accumulated_value, 1)
async def async_added_to_hass(self) -> None:
"""Load the last known state value of the entity if the accumulated type."""
await super().async_added_to_hass()
state = await self.async_get_last_state()
self._accumulated_value = 0.0
if state:
try:
# Previous value could be `unknown` if the connection didn't originally
# complete.
self._accumulated_value = float(state.state)
except (ValueError) as err:
_LOGGER.warning("Could not restore last state: %s", err)
else:
if ATTR_LAST_UPDATE in state.attributes:
last_run = dt.parse_datetime(state.attributes[ATTR_LAST_UPDATE])
if last_run is not None:
self.coordinator.update_last_run(last_run)
# Force a second update from the iotawatt to ensure that sensors are up to date.
await self.coordinator.async_request_refresh()
@property
def name(self) -> str | None:
"""Return name of the entity."""
return f"{self._sensor_data.getSourceName()} Accumulated"
@property
def extra_state_attributes(self) -> dict[str, str]:
"""Return the extra state attributes of the entity."""
attrs = super().extra_state_attributes
assert (
self.coordinator.api is not None
and self.coordinator.api.getLastUpdateTime() is not None
)
attrs[ATTR_LAST_UPDATE] = self.coordinator.api.getLastUpdateTime().isoformat()
return attrs

24
tests/components/iotawatt/__init__.py
View File

@ -22,27 +22,3 @@ OUTPUT_SENSOR = Sensor(
mac_addr="mock-mac",
fromStart=True,
)
INPUT_ACCUMULATED_SENSOR = Sensor(
channel="N/A",
base_name="My WattHour Accumulated Input Sensor",
suffix=".wh",
io_type="Input",
unit="WattHours",
value=500,
begin="",
mac_addr="mock-mac",
fromStart=False,
)
OUTPUT_ACCUMULATED_SENSOR = Sensor(
channel="N/A",
base_name="My WattHour Accumulated Output Sensor",
suffix=".wh",
io_type="Output",
unit="WattHours",
value=200,
begin="",
mac_addr="mock-mac",
fromStart=False,
)

169
tests/components/iotawatt/test_sensor.py
View File

@ -1,7 +1,6 @@
"""Test setting up sensors."""
from datetime import timedelta
from homeassistant.components.iotawatt.const import ATTR_LAST_UPDATE
from homeassistant.components.sensor import (
ATTR_STATE_CLASS,
SensorDeviceClass,
@ -14,18 +13,12 @@ from homeassistant.const import (
UnitOfEnergy,
UnitOfPower,
)
from homeassistant.core import State
from homeassistant.setup import async_setup_component
from homeassistant.util.dt import utcnow
from . import (
INPUT_ACCUMULATED_SENSOR,
INPUT_SENSOR,
OUTPUT_ACCUMULATED_SENSOR,
OUTPUT_SENSOR,
)
from . import INPUT_SENSOR, OUTPUT_SENSOR
from tests.common import async_fire_time_changed, mock_restore_cache
from tests.common import async_fire_time_changed
async def test_sensor_type_input(hass, mock_iotawatt):
@ -83,161 +76,3 @@ async def test_sensor_type_output(hass, mock_iotawatt):
await hass.async_block_till_done()
assert hass.states.get("sensor.my_watthour_sensor") is None
async def test_sensor_type_accumulated_output(hass, mock_iotawatt):
"""Tests the sensor type of Accumulated Output and that it's properly restored from saved state."""
mock_iotawatt.getSensors.return_value["sensors"][
"my_watthour_accumulated_output_sensor_key"
] = OUTPUT_ACCUMULATED_SENSOR
DUMMY_DATE = "2021-09-01T14:00:00+10:00"
mock_restore_cache(
hass,
(
State(
"sensor.my_watthour_accumulated_output_sensor_wh_accumulated",
"100.0",
{
"device_class": SensorDeviceClass.ENERGY,
"unit_of_measurement": UnitOfEnergy.WATT_HOUR,
"last_update": DUMMY_DATE,
},
),
),
)
assert await async_setup_component(hass, "iotawatt", {})
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids()) == 1
state = hass.states.get(
"sensor.my_watthour_accumulated_output_sensor_wh_accumulated"
)
assert state is not None
assert state.state == "300.0" # 100 + 200
assert (
state.attributes[ATTR_FRIENDLY_NAME]
== "My WattHour Accumulated Output Sensor.wh Accumulated"
)
assert state.attributes[ATTR_STATE_CLASS] is SensorStateClass.TOTAL
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == UnitOfEnergy.WATT_HOUR
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.ENERGY
assert state.attributes["type"] == "Output"
assert state.attributes[ATTR_LAST_UPDATE] is not None
assert state.attributes[ATTR_LAST_UPDATE] != DUMMY_DATE
async def test_sensor_type_accumulated_output_error_restore(hass, mock_iotawatt):
"""Tests the sensor type of Accumulated Output and that it's properly restored from saved state."""
mock_iotawatt.getSensors.return_value["sensors"][
"my_watthour_accumulated_output_sensor_key"
] = OUTPUT_ACCUMULATED_SENSOR
DUMMY_DATE = "2021-09-01T14:00:00+10:00"
mock_restore_cache(
hass,
(
State(
"sensor.my_watthour_accumulated_output_sensor_wh_accumulated",
"unknown",
),
),
)
assert await async_setup_component(hass, "iotawatt", {})
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids()) == 1
state = hass.states.get(
"sensor.my_watthour_accumulated_output_sensor_wh_accumulated"
)
assert state is not None
assert state.state == "200.0" # Returns the new read as restore failed.
assert (
state.attributes[ATTR_FRIENDLY_NAME]
== "My WattHour Accumulated Output Sensor.wh Accumulated"
)
assert state.attributes[ATTR_STATE_CLASS] is SensorStateClass.TOTAL
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == UnitOfEnergy.WATT_HOUR
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.ENERGY
assert state.attributes["type"] == "Output"
assert state.attributes[ATTR_LAST_UPDATE] is not None
assert state.attributes[ATTR_LAST_UPDATE] != DUMMY_DATE
async def test_sensor_type_multiple_accumulated_output(hass, mock_iotawatt):
"""Tests the sensor type of Accumulated Output and that it's properly restored from saved state."""
mock_iotawatt.getSensors.return_value["sensors"][
"my_watthour_accumulated_output_sensor_key"
] = OUTPUT_ACCUMULATED_SENSOR
mock_iotawatt.getSensors.return_value["sensors"][
"my_watthour_accumulated_input_sensor_key"
] = INPUT_ACCUMULATED_SENSOR
DUMMY_DATE = "2021-09-01T14:00:00+10:00"
mock_restore_cache(
hass,
(
State(
"sensor.my_watthour_accumulated_output_sensor_wh_accumulated",
"100.0",
{
"device_class": SensorDeviceClass.ENERGY,
"unit_of_measurement": UnitOfEnergy.WATT_HOUR,
"last_update": DUMMY_DATE,
},
),
State(
"sensor.my_watthour_accumulated_input_sensor_wh_accumulated",
"50.0",
{
"device_class": SensorDeviceClass.ENERGY,
"unit_of_measurement": UnitOfEnergy.WATT_HOUR,
"last_update": DUMMY_DATE,
},
),
),
)
assert await async_setup_component(hass, "iotawatt", {})
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids()) == 2
state = hass.states.get(
"sensor.my_watthour_accumulated_output_sensor_wh_accumulated"
)
assert state is not None
assert state.state == "300.0" # 100 + 200
assert (
state.attributes[ATTR_FRIENDLY_NAME]
== "My WattHour Accumulated Output Sensor.wh Accumulated"
)
assert state.attributes[ATTR_STATE_CLASS] is SensorStateClass.TOTAL
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == UnitOfEnergy.WATT_HOUR
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.ENERGY
assert state.attributes["type"] == "Output"
assert state.attributes[ATTR_LAST_UPDATE] is not None
assert state.attributes[ATTR_LAST_UPDATE] != DUMMY_DATE
state = hass.states.get(
"sensor.my_watthour_accumulated_input_sensor_wh_accumulated"
)
assert state is not None
assert state.state == "550.0" # 50 + 500
assert (
state.attributes[ATTR_FRIENDLY_NAME]
== "My WattHour Accumulated Input Sensor.wh Accumulated"
)
assert state.attributes[ATTR_LAST_UPDATE] is not None
assert state.attributes[ATTR_LAST_UPDATE] != DUMMY_DATE