jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-21 12:17:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor Whirlpool sensor platform (#141958)

Browse Source 
* Refactor Whirlpool sensor platform

* Rename sensor classes

* Remove unused logging

* Split washer dryer translation keys to use icon translations

* Address review comments

* Remove entity name; fix sentence casing
This commit is contained in:
Abílio Costa 2025-04-01 19:02:24 +01:00 committed by GitHub
parent c28a6a867d
commit 704777444c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 187 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
homeassistant/components/whirlpool/entity.py Normal file
View File

@ -0,0 +1,38 @@
"""Base entity for the Whirlpool integration."""
from whirlpool.appliance import Appliance
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity
from .const import DOMAIN
class WhirlpoolEntity(Entity):
"""Base class for Whirlpool entities."""
_attr_has_entity_name = True
def __init__(self, appliance: Appliance, unique_id_suffix: str = "") -> None:
"""Initialize the entity."""
self._appliance = appliance
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, appliance.said)},
name=appliance.name.capitalize(),
manufacturer="Whirlpool",
)
self._attr_unique_id = f"{appliance.said}{unique_id_suffix}"
async def async_added_to_hass(self) -> None:
"""Register attribute updates callback."""
self._appliance.register_attr_callback(self.async_write_ha_state)
async def async_will_remove_from_hass(self) -> None:
"""Unregister attribute updates callback."""
self._appliance.unregister_attr_callback(self.async_write_ha_state)
@property
def available(self) -> bool:
"""Return True if entity is available."""
return self._appliance.get_online()

12
homeassistant/components/whirlpool/icons.json Normal file
View File

@ -0,0 +1,12 @@
{
"entity": {
"sensor": {
"washer_state": {
"default": "mdi:washing-machine"
},
"dryer_state": {
"default": "mdi:tumble-dryer"
}
}
}
}

179
homeassistant/components/whirlpool/sensor.py
View File

@ -3,8 +3,9 @@
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime, timedelta from datetime import datetime, timedelta
import logging from typing import override
from whirlpool.appliance import Appliance
from whirlpool.washerdryer import MachineState, WasherDryer from whirlpool.washerdryer import MachineState, WasherDryer
from homeassistant.components.sensor import ( from homeassistant.components.sensor import (
@ -13,16 +14,17 @@ from homeassistant.components.sensor import (
SensorEntity, SensorEntity,
SensorEntityDescription, SensorEntityDescription,
) )
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType from homeassistant.helpers.typing import StateType
from homeassistant.util.dt import utcnow from homeassistant.util.dt import utcnow
from . import WhirlpoolConfigEntry from . import WhirlpoolConfigEntry
from .const import DOMAIN from .entity import WhirlpoolEntity
TANK_FILL = { SCAN_INTERVAL = timedelta(minutes=5)
WASHER_TANK_FILL = {
"0": "unknown", "0": "unknown",
"1": "empty", "1": "empty",
"2": "25", "2": "25",
@ -31,7 +33,7 @@ TANK_FILL = {
"5": "active", "5": "active",
} }
MACHINE_STATE = { WASHER_DRYER_MACHINE_STATE = {
MachineState.Standby: "standby", MachineState.Standby: "standby",
MachineState.Setting: "setting", MachineState.Setting: "setting",
MachineState.DelayCountdownMode: "delay_countdown", MachineState.DelayCountdownMode: "delay_countdown",
@ -53,7 +55,7 @@ MACHINE_STATE = {
MachineState.SystemInit: "system_initialize", MachineState.SystemInit: "system_initialize",
} }
CYCLE_FUNC = [ WASHER_DRYER_CYCLE_FUNC = [
(WasherDryer.get_cycle_status_filling, "cycle_filling"), (WasherDryer.get_cycle_status_filling, "cycle_filling"),
(WasherDryer.get_cycle_status_rinsing, "cycle_rinsing"), (WasherDryer.get_cycle_status_rinsing, "cycle_rinsing"),
(WasherDryer.get_cycle_status_sensing, "cycle_sensing"), (WasherDryer.get_cycle_status_sensing, "cycle_sensing"),
@ -62,60 +64,69 @@ CYCLE_FUNC = [
(WasherDryer.get_cycle_status_washing, "cycle_washing"), (WasherDryer.get_cycle_status_washing, "cycle_washing"),
] ]
DOOR_OPEN = "door_open" STATE_DOOR_OPEN = "door_open"
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=5)
def washer_state(washer: WasherDryer) -> str | None: def washer_dryer_state(washer_dryer: WasherDryer) -> str | None:
"""Determine correct states for a washer.""" """Determine correct states for a washer/dryer."""
if washer.get_attribute("Cavity_OpStatusDoorOpen") == "1": if washer_dryer.get_attribute("Cavity_OpStatusDoorOpen") == "1":
return DOOR_OPEN return STATE_DOOR_OPEN
machine_state = washer.get_machine_state() machine_state = washer_dryer.get_machine_state()
if machine_state == MachineState.RunningMainCycle: if machine_state == MachineState.RunningMainCycle:
for func, cycle_name in CYCLE_FUNC: for func, cycle_name in WASHER_DRYER_CYCLE_FUNC:
if func(washer): if func(washer_dryer):
return cycle_name return cycle_name
return MACHINE_STATE.get(machine_state) return WASHER_DRYER_MACHINE_STATE.get(machine_state)
@dataclass(frozen=True, kw_only=True) @dataclass(frozen=True, kw_only=True)
class WhirlpoolSensorEntityDescription(SensorEntityDescription): class WhirlpoolSensorEntityDescription(SensorEntityDescription):
"""Describes Whirlpool Washer sensor entity.""" """Describes a Whirlpool sensor entity."""
value_fn: Callable value_fn: Callable[[Appliance], str | None]
SENSORS: tuple[WhirlpoolSensorEntityDescription, ...] = ( WASHER_DRYER_STATE_OPTIONS = (
list(WASHER_DRYER_MACHINE_STATE.values())
+ [value for _, value in WASHER_DRYER_CYCLE_FUNC]
+ [STATE_DOOR_OPEN]
)
WASHER_SENSORS: tuple[WhirlpoolSensorEntityDescription, ...] = (
WhirlpoolSensorEntityDescription( WhirlpoolSensorEntityDescription(
key="state", key="state",
translation_key="whirlpool_machine", translation_key="washer_state",
device_class=SensorDeviceClass.ENUM, device_class=SensorDeviceClass.ENUM,
options=( options=WASHER_DRYER_STATE_OPTIONS,
list(MACHINE_STATE.values()) value_fn=washer_dryer_state,
+ [value for _, value in CYCLE_FUNC]
+ [DOOR_OPEN]
),
value_fn=washer_state,
), ),
WhirlpoolSensorEntityDescription( WhirlpoolSensorEntityDescription(
key="DispenseLevel", key="DispenseLevel",
translation_key="whirlpool_tank", translation_key="whirlpool_tank",
entity_registry_enabled_default=False, entity_registry_enabled_default=False,
device_class=SensorDeviceClass.ENUM, device_class=SensorDeviceClass.ENUM,
options=list(TANK_FILL.values()), options=list(WASHER_TANK_FILL.values()),
value_fn=lambda WasherDryer: TANK_FILL.get( value_fn=lambda washer: WASHER_TANK_FILL.get(
WasherDryer.get_attribute("WashCavity_OpStatusBulkDispense1Level") washer.get_attribute("WashCavity_OpStatusBulkDispense1Level")
), ),
), ),
) )
SENSOR_TIMER: tuple[SensorEntityDescription] = ( DRYER_SENSORS: tuple[WhirlpoolSensorEntityDescription, ...] = (
WhirlpoolSensorEntityDescription(
key="state",
translation_key="dryer_state",
device_class=SensorDeviceClass.ENUM,
options=WASHER_DRYER_STATE_OPTIONS,
value_fn=washer_dryer_state,
),
)
WASHER_DRYER_TIME_SENSORS: tuple[SensorEntityDescription] = (
SensorEntityDescription( SensorEntityDescription(
key="timeremaining", key="timeremaining",
translation_key="end_time", translation_key="end_time",
@ -134,106 +145,71 @@ async def async_setup_entry(
entities: list = [] entities: list = []
appliances_manager = config_entry.runtime_data appliances_manager = config_entry.runtime_data
for washer_dryer in appliances_manager.washer_dryers: for washer_dryer in appliances_manager.washer_dryers:
sensor_descriptions = (
DRYER_SENSORS
if "dryer" in washer_dryer.appliance_info.data_model.lower()
else WASHER_SENSORS
)
entities.extend( entities.extend(
[WasherDryerClass(washer_dryer, description) for description in SENSORS] WhirlpoolSensor(washer_dryer, description)
for description in sensor_descriptions
) )
entities.extend( entities.extend(
[ WasherDryerTimeSensor(washer_dryer, description)
WasherDryerTimeClass(washer_dryer, description) for description in WASHER_DRYER_TIME_SENSORS
for description in SENSOR_TIMER
]
) )
async_add_entities(entities) async_add_entities(entities)
class WasherDryerClass(SensorEntity): class WhirlpoolSensor(WhirlpoolEntity, SensorEntity):
"""A class for the whirlpool/maytag washer account.""" """A class for the Whirlpool sensors."""
_attr_should_poll = False _attr_should_poll = False
_attr_has_entity_name = True
def __init__( def __init__(
self, washer_dryer: WasherDryer, description: WhirlpoolSensorEntityDescription self, appliance: Appliance, description: WhirlpoolSensorEntityDescription
) -> None: ) -> None:
"""Initialize the washer sensor.""" """Initialize the washer sensor."""
self._wd: WasherDryer = washer_dryer super().__init__(appliance, unique_id_suffix=f"-{description.key}")
self._attr_icon = (
"mdi:tumble-dryer"
if "dryer" in washer_dryer.appliance_info.data_model.lower()
else "mdi:washing-machine"
)
self.entity_description: WhirlpoolSensorEntityDescription = description self.entity_description: WhirlpoolSensorEntityDescription = description
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, washer_dryer.said)},
name=washer_dryer.name.capitalize(),
manufacturer="Whirlpool",
)
self._attr_unique_id = f"{washer_dryer.said}-{description.key}"
async def async_added_to_hass(self) -> None:
"""Register updates callback."""
self._wd.register_attr_callback(self.async_write_ha_state)
async def async_will_remove_from_hass(self) -> None:
"""Unregister updates callback."""
self._wd.unregister_attr_callback(self.async_write_ha_state)
@property
def available(self) -> bool:
"""Return True if entity is available."""
return self._wd.get_online()
@property @property
def native_value(self) -> StateType | str: def native_value(self) -> StateType | str:
"""Return native value of sensor.""" """Return native value of sensor."""
return self.entity_description.value_fn(self._wd) return self.entity_description.value_fn(self._appliance)
class WasherDryerTimeClass(RestoreSensor): class WasherDryerTimeSensor(WhirlpoolEntity, RestoreSensor):
"""A timestamp class for the whirlpool/maytag washer account.""" """A timestamp class for the Whirlpool washer/dryer."""
_attr_should_poll = True _attr_should_poll = True
_attr_has_entity_name = True
def __init__( def __init__(
self, washer_dryer: WasherDryer, description: SensorEntityDescription self, washer_dryer: WasherDryer, description: SensorEntityDescription
) -> None: ) -> None:
"""Initialize the washer sensor.""" """Initialize the washer sensor."""
self._wd: WasherDryer = washer_dryer super().__init__(washer_dryer, unique_id_suffix=f"-{description.key}")
self.entity_description = description
self.entity_description: SensorEntityDescription = description self._wd = washer_dryer
self._running: bool | None = None self._running: bool | None = None
self._attr_device_info = DeviceInfo( self._value: datetime | None = None
identifiers={(DOMAIN, washer_dryer.said)},
name=washer_dryer.name.capitalize(),
manufacturer="Whirlpool",
)
self._attr_unique_id = f"{washer_dryer.said}-{description.key}"
async def async_added_to_hass(self) -> None: async def async_added_to_hass(self) -> None:
"""Connect washer/dryer to the cloud.""" """Register attribute updates callback."""
if restored_data := await self.async_get_last_sensor_data(): if restored_data := await self.async_get_last_sensor_data():
self._attr_native_value = restored_data.native_value if isinstance(restored_data.native_value, datetime):
self._value = restored_data.native_value
await super().async_added_to_hass() await super().async_added_to_hass()
self._wd.register_attr_callback(self.update_from_latest_data)
async def async_will_remove_from_hass(self) -> None:
"""Close Whrilpool Appliance sockets before removing."""
self._wd.unregister_attr_callback(self.update_from_latest_data)
@property
def available(self) -> bool:
"""Return True if entity is available."""
return self._wd.get_online()
async def async_update(self) -> None: async def async_update(self) -> None:
"""Update status of Whirlpool.""" """Update status of Whirlpool."""
await self._wd.fetch_data() await self._wd.fetch_data()
@callback @override
def update_from_latest_data(self) -> None: @property
def native_value(self) -> datetime | None:
"""Calculate the time stamp for completion.""" """Calculate the time stamp for completion."""
machine_state = self._wd.get_machine_state() machine_state = self._wd.get_machine_state()
now = utcnow() now = utcnow()
@ -243,8 +219,7 @@ class WasherDryerTimeClass(RestoreSensor):
and self._running and self._running
): ):
self._running = False self._running = False
self._attr_native_value = now self._value = now
self._async_write_ha_state()
if machine_state is MachineState.RunningMainCycle: if machine_state is MachineState.RunningMainCycle:
self._running = True self._running = True
@ -253,9 +228,9 @@ class WasherDryerTimeClass(RestoreSensor):
seconds=int(self._wd.get_attribute("Cavity_TimeStatusEstTimeRemaining")) seconds=int(self._wd.get_attribute("Cavity_TimeStatusEstTimeRemaining"))
) )
if self._attr_native_value is None or ( if self._value is None or (
isinstance(self._attr_native_value, datetime) isinstance(self._value, datetime)
and abs(new_timestamp - self._attr_native_value) > timedelta(seconds=60) and abs(new_timestamp - self._value) > timedelta(seconds=60)
): ):
self._attr_native_value = new_timestamp self._value = new_timestamp
self._async_write_ha_state() return self._value

75
homeassistant/components/whirlpool/strings.json
View File

@ -43,35 +43,64 @@
}, },
"entity": { "entity": {
"sensor": { "sensor": {
"whirlpool_machine": { "washer_state": {
"name": "State",
"state": { "state": {
"standby": "[%key:common::state::standby%]", "standby": "[%key:common::state::standby%]",
"setting": "Setting", "setting": "Setting",
"delay_countdown": "Delay Countdown", "delay_countdown": "Delay countdown",
"delay_paused": "Delay Paused", "delay_paused": "Delay paused",
"smart_delay": "Smart Delay", "smart_delay": "Smart delay",
"smart_grid_pause": "[%key:component::whirlpool::entity::sensor::whirlpool_machine::state::smart_delay%]", "smart_grid_pause": "[%key:component::whirlpool::entity::sensor::washer_state::state::smart_delay%]",
"pause": "[%key:common::state::paused%]", "pause": "[%key:common::state::paused%]",
"running_maincycle": "Running Maincycle", "running_maincycle": "Running maincycle",
"running_postcycle": "Running Postcycle", "running_postcycle": "Running postcycle",
"exception": "Exception", "exception": "Exception",
"complete": "Complete", "complete": "Complete",
"power_failure": "Power Failure", "power_failure": "Power failure",
"service_diagnostic_mode": "Service Diagnostic Mode", "service_diagnostic_mode": "Service diagnostic mode",
"factory_diagnostic_mode": "Factory Diagnostic Mode", "factory_diagnostic_mode": "Factory diagnostic mode",
"life_test": "Life Test", "life_test": "Life test",
"customer_focus_mode": "Customer Focus Mode", "customer_focus_mode": "Customer focus mode",
"demo_mode": "Demo Mode", "demo_mode": "Demo mode",
"hard_stop_or_error": "Hard Stop or Error", "hard_stop_or_error": "Hard stop or error",
"system_initialize": "System Initialize", "system_initialize": "System initialize",
"cycle_filling": "Cycle Filling", "cycle_filling": "Cycle filling",
"cycle_rinsing": "Cycle Rinsing", "cycle_rinsing": "Cycle rinsing",
"cycle_sensing": "Cycle Sensing", "cycle_sensing": "Cycle sensing",
"cycle_soaking": "Cycle Soaking", "cycle_soaking": "Cycle soaking",
"cycle_spinning": "Cycle Spinning", "cycle_spinning": "Cycle spinning",
"cycle_washing": "Cycle Washing", "cycle_washing": "Cycle washing",
"door_open": "Door Open" "door_open": "Door open"
}
},
"dryer_state": {
"state": {
"standby": "[%key:common::state::standby%]",
"setting": "[%key:component::whirlpool::entity::sensor::washer_state::state::setting%]",
"delay_countdown": "[%key:component::whirlpool::entity::sensor::washer_state::state::delay_countdown%]",
"delay_paused": "[%key:component::whirlpool::entity::sensor::washer_state::state::delay_paused%]",
"smart_delay": "[%key:component::whirlpool::entity::sensor::washer_state::state::smart_delay%]",
"smart_grid_pause": "[%key:component::whirlpool::entity::sensor::washer_state::state::smart_delay%]",
"pause": "[%key:common::state::paused%]",
"running_maincycle": "[%key:component::whirlpool::entity::sensor::washer_state::state::running_maincycle%]",
"running_postcycle": "[%key:component::whirlpool::entity::sensor::washer_state::state::running_postcycle%]",
"exception": "[%key:component::whirlpool::entity::sensor::washer_state::state::exception%]",
"complete": "[%key:component::whirlpool::entity::sensor::washer_state::state::complete%]",
"power_failure": "[%key:component::whirlpool::entity::sensor::washer_state::state::power_failure%]",
"service_diagnostic_mode": "[%key:component::whirlpool::entity::sensor::washer_state::state::service_diagnostic_mode%]",
"factory_diagnostic_mode": "[%key:component::whirlpool::entity::sensor::washer_state::state::factory_diagnostic_mode%]",
"life_test": "[%key:component::whirlpool::entity::sensor::washer_state::state::life_test%]",
"customer_focus_mode": "[%key:component::whirlpool::entity::sensor::washer_state::state::customer_focus_mode%]",
"demo_mode": "[%key:component::whirlpool::entity::sensor::washer_state::state::demo_mode%]",
"hard_stop_or_error": "[%key:component::whirlpool::entity::sensor::washer_state::state::hard_stop_or_error%]",
"system_initialize": "[%key:component::whirlpool::entity::sensor::washer_state::state::system_initialize%]",
"cycle_filling": "[%key:component::whirlpool::entity::sensor::washer_state::state::cycle_filling%]",
"cycle_rinsing": "[%key:component::whirlpool::entity::sensor::washer_state::state::cycle_rinsing%]",
"cycle_sensing": "[%key:component::whirlpool::entity::sensor::washer_state::state::cycle_sensing%]",
"cycle_soaking": "[%key:component::whirlpool::entity::sensor::washer_state::state::cycle_soaking%]",
"cycle_spinning": "[%key:component::whirlpool::entity::sensor::washer_state::state::cycle_spinning%]",
"cycle_washing": "[%key:component::whirlpool::entity::sensor::washer_state::state::cycle_washing%]",
"door_open": "[%key:component::whirlpool::entity::sensor::washer_state::state::door_open%]"
} }
}, },
"whirlpool_tank": { "whirlpool_tank": {

8
tests/components/whirlpool/conftest.py
View File

@ -153,12 +153,12 @@ def side_effect_function(*args, **kwargs):
return None return None
def get_sensor_mock(said): def get_sensor_mock(said: str, data_model: str):
"""Get a mock of a sensor.""" """Get a mock of a sensor."""
mock_sensor = mock.Mock(said=said) mock_sensor = mock.Mock(said=said)
mock_sensor.name = f"WasherDryer {said}" mock_sensor.name = f"WasherDryer {said}"
mock_sensor.register_attr_callback = MagicMock() mock_sensor.register_attr_callback = MagicMock()
mock_sensor.appliance_info.data_model = "washer_dryer_model" mock_sensor.appliance_info.data_model = data_model
mock_sensor.appliance_info.category = "washer_dryer" mock_sensor.appliance_info.category = "washer_dryer"
mock_sensor.appliance_info.model_number = "12345" mock_sensor.appliance_info.model_number = "12345"
mock_sensor.get_online.return_value = True mock_sensor.get_online.return_value = True
@ -179,13 +179,13 @@ def get_sensor_mock(said):
@pytest.fixture(name="mock_sensor1_api", autouse=False) @pytest.fixture(name="mock_sensor1_api", autouse=False)
def fixture_mock_sensor1_api(): def fixture_mock_sensor1_api():
"""Set up sensor API fixture.""" """Set up sensor API fixture."""
return get_sensor_mock(MOCK_SAID3) return get_sensor_mock(MOCK_SAID3, "washer")
@pytest.fixture(name="mock_sensor2_api", autouse=False) @pytest.fixture(name="mock_sensor2_api", autouse=False)
def fixture_mock_sensor2_api(): def fixture_mock_sensor2_api():
"""Set up sensor API fixture.""" """Set up sensor API fixture."""
return get_sensor_mock(MOCK_SAID4) return get_sensor_mock(MOCK_SAID4, "dryer")
@pytest.fixture(name="mock_sensor_api_instances", autouse=False) @pytest.fixture(name="mock_sensor_api_instances", autouse=False)

4
tests/components/whirlpool/snapshots/test_diagnostics.ambr
View File

@ -19,12 +19,12 @@
'washer_dryers': dict({ 'washer_dryers': dict({
'WasherDryer said3': dict({ 'WasherDryer said3': dict({
'category': 'washer_dryer', 'category': 'washer_dryer',
'data_model': 'washer_dryer_model', 'data_model': 'washer',
'model_number': '12345', 'model_number': '12345',
}), }),
'WasherDryer said4': dict({ 'WasherDryer said4': dict({
'category': 'washer_dryer', 'category': 'washer_dryer',
'data_model': 'washer_dryer_model', 'data_model': 'dryer',
'model_number': '12345', 'model_number': '12345',
}), }),
}), }),

4
tests/components/whirlpool/test_sensor.py
View File

@ -66,7 +66,7 @@ async def test_dryer_sensor_values(
await init_integration(hass) await init_integration(hass)
entity_id = f"sensor.washerdryer_{MOCK_SAID4}_state" entity_id = f"sensor.washerdryer_{MOCK_SAID4}_none"
mock_instance = mock_sensor2_api mock_instance = mock_sensor2_api
entry = entity_registry.async_get(entity_id) entry = entity_registry.async_get(entity_id)
assert entry assert entry
@ -130,7 +130,7 @@ async def test_washer_sensor_values(
) )
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = f"sensor.washerdryer_{MOCK_SAID3}_state" entity_id = f"sensor.washerdryer_{MOCK_SAID3}_none"
mock_instance = mock_sensor1_api mock_instance = mock_sensor1_api
entry = entity_registry.async_get(entity_id) entry = entity_registry.async_get(entity_id)
assert entry assert entry