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

Add preview to mold_indicator (#125530)

Browse Source
This commit is contained in:
G Johansson 2024-09-27 20:56:02 +02:00 committed by GitHub
parent 317b73ffaf
commit 57e041171b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 410 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
homeassistant/components/mold_indicator/config_flow.py
View File

@ -7,8 +7,11 @@ from typing import Any, cast
import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.components.sensor import SensorDeviceClass
from homeassistant.const import CONF_NAME, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.schema_config_entry_flow import (
SchemaCommonFlowHandler,
SchemaConfigFlowHandler,
@ -22,6 +25,7 @@ from homeassistant.helpers.selector import (
NumberSelectorMode,
TextSelector,
)
from homeassistant.util.unit_system import METRIC_SYSTEM
from .const import (
CONF_CALIBRATION_FACTOR,
@ -31,6 +35,7 @@ from .const import (
DEFAULT_NAME,
DOMAIN,
)
from .sensor import MoldIndicator
async def validate_duplicate(
@ -75,12 +80,14 @@ CONFIG_FLOW = {
"user": SchemaFlowFormStep(
schema=DATA_SCHEMA_CONFIG,
validate_user_input=validate_duplicate,
preview="mold_indicator",
),
}
OPTIONS_FLOW = {
"init": SchemaFlowFormStep(
DATA_SCHEMA_OPTIONS,
validate_user_input=validate_duplicate,
preview="mold_indicator",
)
}
@ -94,3 +101,71 @@ class MoldIndicatorConfigFlowHandler(SchemaConfigFlowHandler, domain=DOMAIN):
def async_config_entry_title(self, options: Mapping[str, Any]) -> str:
"""Return config entry title."""
return cast(str, options[CONF_NAME])
@staticmethod
async def async_setup_preview(hass: HomeAssistant) -> None:
"""Set up preview WS API."""
websocket_api.async_register_command(hass, ws_start_preview)
@websocket_api.websocket_command(
{
vol.Required("type"): "mold_indicator/start_preview",
vol.Required("flow_id"): str,
vol.Required("flow_type"): vol.Any("config_flow", "options_flow"),
vol.Required("user_input"): dict,
}
)
@callback
def ws_start_preview(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""Generate a preview."""
if msg["flow_type"] == "config_flow":
flow_status = hass.config_entries.flow.async_get(msg["flow_id"])
flow_sets = hass.config_entries.flow._handler_progress_index.get( # noqa: SLF001
flow_status["handler"]
)
assert flow_sets
config_entry = hass.config_entries.async_get_entry(flow_status["handler"])
indoor_temp = msg["user_input"].get(CONF_INDOOR_TEMP)
outdoor_temp = msg["user_input"].get(CONF_OUTDOOR_TEMP)
indoor_hum = msg["user_input"].get(CONF_INDOOR_HUMIDITY)
name = msg["user_input"].get(CONF_NAME)
else:
flow_status = hass.config_entries.options.async_get(msg["flow_id"])
config_entry = hass.config_entries.async_get_entry(flow_status["handler"])
if not config_entry:
raise HomeAssistantError("Config entry not found")
indoor_temp = config_entry.options[CONF_INDOOR_TEMP]
outdoor_temp = config_entry.options[CONF_OUTDOOR_TEMP]
indoor_hum = config_entry.options[CONF_INDOOR_HUMIDITY]
name = config_entry.options[CONF_NAME]
@callback
def async_preview_updated(state: str, attributes: Mapping[str, Any]) -> None:
"""Forward config entry state events to websocket."""
connection.send_message(
websocket_api.event_message(
msg["id"], {"attributes": attributes, "state": state}
)
)
preview_entity = MoldIndicator(
hass,
name,
hass.config.units is METRIC_SYSTEM,
indoor_temp,
outdoor_temp,
indoor_hum,
msg["user_input"].get(CONF_CALIBRATION_FACTOR),
)
preview_entity.hass = hass
connection.send_result(msg["id"])
connection.subscriptions[msg["id"]] = preview_entity.async_start_preview(
async_preview_updated
)

73
homeassistant/components/mold_indicator/sensor.py
View File

@ -2,6 +2,7 @@
from __future__ import annotations
from collections.abc import Callable, Mapping
import logging
import math
from typing import TYPE_CHECKING, Any
@ -19,12 +20,12 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_UNIT_OF_MEASUREMENT,
CONF_NAME,
EVENT_HOMEASSISTANT_START,
PERCENTAGE,
STATE_UNKNOWN,
UnitOfTemperature,
)
from homeassistant.core import (
CALLBACK_TYPE,
Event,
EventStateChangedData,
HomeAssistant,
@ -156,19 +157,48 @@ class MoldIndicator(SensorEntity):
indoor_humidity_sensor,
outdoor_temp_sensor,
}
self._dewpoint: float | None = None
self._indoor_temp: float | None = None
self._outdoor_temp: float | None = None
self._indoor_hum: float | None = None
self._crit_temp: float | None = None
self._attr_device_info = async_device_info_to_link_from_entity(
hass,
indoor_humidity_sensor,
)
if indoor_humidity_sensor:
self._attr_device_info = async_device_info_to_link_from_entity(
hass,
indoor_humidity_sensor,
)
self._preview_callback: Callable[[str, Mapping[str, Any]], None] | None = None
@callback
def async_start_preview(
self,
preview_callback: Callable[[str, Mapping[str, Any]], None],
) -> CALLBACK_TYPE:
"""Render a preview."""
# Abort early if there is no source entity_id's or calibration factor
if (
not self._outdoor_temp_sensor
or not self._indoor_temp_sensor
or not self._indoor_humidity_sensor
or not self._calib_factor
):
self._attr_available = False
calculated_state = self._async_calculate_state()
preview_callback(calculated_state.state, calculated_state.attributes)
return self._call_on_remove_callbacks
self._preview_callback = preview_callback
self._async_setup_sensor()
return self._call_on_remove_callbacks
async def async_added_to_hass(self) -> None:
"""Register callbacks."""
"""Run when entity about to be added to hass."""
self._async_setup_sensor()
@callback
def _async_setup_sensor(self) -> None:
"""Set up the sensor and start tracking state changes."""
@callback
def mold_indicator_sensors_state_listener(
@ -186,10 +216,17 @@ class MoldIndicator(SensorEntity):
)
if self._update_sensor(entity, old_state, new_state):
self.async_schedule_update_ha_state(True)
if self._preview_callback:
calculated_state = self._async_calculate_state()
self._preview_callback(
calculated_state.state, calculated_state.attributes
)
# only write state to the state machine if we are not in preview mode
else:
self.async_schedule_update_ha_state(True)
@callback
def mold_indicator_startup(event: Event) -> None:
def mold_indicator_startup() -> None:
"""Add listeners and get 1st state."""
_LOGGER.debug("Startup for %s", self.entity_id)
@ -222,12 +259,22 @@ class MoldIndicator(SensorEntity):
else schedule_update
)
if schedule_update:
if schedule_update and not self._preview_callback:
self.async_schedule_update_ha_state(True)
if self._preview_callback:
# re-calculate dewpoint and mold indicator
self._calc_dewpoint()
self._calc_moldindicator()
if self._state is None:
self._attr_available = False
else:
self._attr_available = True
calculated_state = self._async_calculate_state()
self._preview_callback(
calculated_state.state, calculated_state.attributes
)
self.hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_START, mold_indicator_startup
)
mold_indicator_startup()
def _update_sensor(
self, entity: str, old_state: State | None, new_state: State | None

49
tests/components/mold_indicator/snapshots/test_config_flow.ambr Normal file
View File

@ -0,0 +1,49 @@
# serializer version: 1
# name: test_config_flow_preview_success[missing_calibration_factor]
dict({
'attributes': dict({
'device_class': 'humidity',
'friendly_name': 'Mold Indicator',
'state_class': 'measurement',
'unit_of_measurement': '%',
}),
'state': 'unavailable',
})
# ---
# name: test_config_flow_preview_success[missing_humidity_entity]
dict({
'attributes': dict({
'device_class': 'humidity',
'friendly_name': 'Mold Indicator',
'state_class': 'measurement',
'unit_of_measurement': '%',
}),
'state': 'unavailable',
})
# ---
# name: test_config_flow_preview_success[success]
dict({
'attributes': dict({
'device_class': 'humidity',
'dewpoint': 12.01,
'estimated_critical_temp': 19.5,
'friendly_name': 'Mold Indicator',
'state_class': 'measurement',
'unit_of_measurement': '%',
}),
'state': '61',
})
# ---
# name: test_options_flow_preview
dict({
'attributes': dict({
'device_class': 'humidity',
'dewpoint': 12.01,
'estimated_critical_temp': 19.5,
'friendly_name': 'Mold Indicator',
'state_class': 'measurement',
'unit_of_measurement': '%',
}),
'state': '61',
})
# ---

227
tests/components/mold_indicator/test_config_flow.py
View File

@ -4,6 +4,10 @@ from __future__ import annotations
from unittest.mock import AsyncMock
import pytest
from syrupy import SnapshotAssertion
from homeassistant import config_entries
from homeassistant.components.mold_indicator.const import (
CONF_CALIBRATION_FACTOR,
CONF_INDOOR_HUMIDITY,
@ -13,11 +17,12 @@ from homeassistant.components.mold_indicator.const import (
DOMAIN,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_NAME
from homeassistant.const import CONF_NAME, CONF_UNIT_OF_MEASUREMENT, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
from tests.typing import WebSocketGenerator
async def test_form_sensor(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> None:
@ -113,3 +118,223 @@ async def test_entry_already_exist(
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
"user_input",
[
(
{
CONF_NAME: DEFAULT_NAME,
CONF_INDOOR_TEMP: "sensor.indoor_temp",
CONF_INDOOR_HUMIDITY: "sensor.indoor_humidity",
CONF_OUTDOOR_TEMP: "sensor.outdoor_temp",
CONF_CALIBRATION_FACTOR: 2.0,
}
),
(
{
CONF_NAME: DEFAULT_NAME,
CONF_INDOOR_TEMP: "sensor.indoor_temp",
CONF_INDOOR_HUMIDITY: "sensor.indoor_humidity",
CONF_OUTDOOR_TEMP: "sensor.outdoor_temp",
}
),
(
{
CONF_NAME: DEFAULT_NAME,
CONF_INDOOR_TEMP: "sensor.indoor_temp",
CONF_OUTDOOR_TEMP: "sensor.outdoor_temp",
CONF_CALIBRATION_FACTOR: 2.0,
}
),
],
ids=("success", "missing_calibration_factor", "missing_humidity_entity"),
)
async def test_config_flow_preview_success(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
user_input: str,
snapshot: SnapshotAssertion,
) -> None:
"""Test the config flow preview."""
client = await hass_ws_client(hass)
# add state for the tests
hass.states.async_set(
"sensor.indoor_temp",
23,
{CONF_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS},
)
hass.states.async_set(
"sensor.indoor_humidity",
50,
{CONF_UNIT_OF_MEASUREMENT: "%"},
)
hass.states.async_set(
"sensor.outdoor_temp",
16,
{CONF_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS},
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] is None
assert result["preview"] == "mold_indicator"
await client.send_json_auto_id(
{
"type": "mold_indicator/start_preview",
"flow_id": result["flow_id"],
"flow_type": "config_flow",
"user_input": user_input,
}
)
msg = await client.receive_json()
assert msg["success"]
assert msg["result"] is None
msg = await client.receive_json()
assert msg["event"] == snapshot
assert len(hass.states.async_all()) == 3
async def test_options_flow_preview(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,
) -> None:
"""Test the options flow preview."""
client = await hass_ws_client(hass)
# add state for the tests
hass.states.async_set(
"sensor.indoor_temp",
23,
{CONF_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS},
)
hass.states.async_set(
"sensor.indoor_humidity",
50,
{CONF_UNIT_OF_MEASUREMENT: "%"},
)
hass.states.async_set(
"sensor.outdoor_temp",
16,
{CONF_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS},
)
# Setup the config entry
config_entry = MockConfigEntry(
data={},
domain=DOMAIN,
options={
CONF_NAME: DEFAULT_NAME,
CONF_INDOOR_TEMP: "sensor.indoor_temp",
CONF_INDOOR_HUMIDITY: "sensor.indoor_humidity",
CONF_OUTDOOR_TEMP: "sensor.outdoor_temp",
CONF_CALIBRATION_FACTOR: 2.0,
},
title="Test Sensor",
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == FlowResultType.FORM
assert result["errors"] is None
assert result["preview"] == "mold_indicator"
await client.send_json_auto_id(
{
"type": "mold_indicator/start_preview",
"flow_id": result["flow_id"],
"flow_type": "options_flow",
"user_input": {
CONF_NAME: DEFAULT_NAME,
CONF_INDOOR_TEMP: "sensor.indoor_temp",
CONF_INDOOR_HUMIDITY: "sensor.indoor_humidity",
CONF_OUTDOOR_TEMP: "sensor.outdoor_temp",
CONF_CALIBRATION_FACTOR: 2.0,
},
}
)
msg = await client.receive_json()
assert msg["success"]
assert msg["result"] is None
msg = await client.receive_json()
assert msg["event"] == snapshot
assert len(hass.states.async_all()) == 4
async def test_options_flow_sensor_preview_config_entry_removed(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""Test the option flow preview where the config entry is removed."""
client = await hass_ws_client(hass)
hass.states.async_set(
"sensor.indoor_temp",
23,
{CONF_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS},
)
hass.states.async_set(
"sensor.indoor_humidity",
50,
{CONF_UNIT_OF_MEASUREMENT: "%"},
)
hass.states.async_set(
"sensor.outdoor_temp",
16,
{CONF_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS},
)
# Setup the config entry
config_entry = MockConfigEntry(
data={},
domain=DOMAIN,
options={
CONF_NAME: DEFAULT_NAME,
CONF_INDOOR_TEMP: "sensor.indoor_temp",
CONF_INDOOR_HUMIDITY: "sensor.indoor_humidity",
CONF_OUTDOOR_TEMP: "sensor.outdoor_temp",
CONF_CALIBRATION_FACTOR: 2.0,
},
title="Test Sensor",
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == FlowResultType.FORM
assert result["errors"] is None
assert result["preview"] == "mold_indicator"
await hass.config_entries.async_remove(config_entry.entry_id)
await client.send_json_auto_id(
{
"type": "mold_indicator/start_preview",
"flow_id": result["flow_id"],
"flow_type": "options_flow",
"user_input": {
CONF_NAME: DEFAULT_NAME,
CONF_INDOOR_TEMP: "sensor.indoor_temp",
CONF_INDOOR_HUMIDITY: "sensor.indoor_humidity",
CONF_OUTDOOR_TEMP: "sensor.outdoor_temp",
CONF_CALIBRATION_FACTOR: 2.0,
},
}
)
msg = await client.receive_json()
assert not msg["success"]
assert msg["error"] == {
"code": "home_assistant_error",
"message": "Config entry not found",
}