starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-10-06 22:38:02 -04:00
Code Issues Packages Projects Releases Wiki Activity

indi_allsky: Add dynamic hardware sensors and host CPU temperature (#182989)

Browse Source
This commit is contained in:
Hamish
2026-09-30 06:44:12 +02:00
committed by GitHub
parent 0dcada8811
commit 2fe09bdf75
7 changed files with 1353 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/indi_allsky/coordinator.py
+90 -6
View File
@@ -4,7 +4,7 @@ from dataclasses import dataclass
import logging
from typing import override
from aioindiallsky import ExposureData, IndiAllSkyClient, IndiAllSkyError
from aioindiallsky import ExposureData, IndiAllSkyClient, IndiAllSkyError, SensorData
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_SSL, CONF_VERIFY_SSL
@@ -25,6 +25,7 @@ class IndiAllSkyData:
"""Data model for INDI Allsky coordinator data."""
exposure: ExposureData | None = None
sensor: SensorData | None = None
class IndiAllSkyDataUpdateCoordinator(DataUpdateCoordinator[IndiAllSkyData]):
@@ -42,11 +43,16 @@ class IndiAllSkyDataUpdateCoordinator(DataUpdateCoordinator[IndiAllSkyData]):
session=async_get_clientsession(hass),
)
self.latest_exposure: ExposureData | None = None
self.latest_sensor: SensorData | None = None
unsub = self.client.register_callback(
"exposure_complete", self._handle_exposure_complete
entry.async_on_unload(
self.client.register_callback(
"exposure_complete", self._handle_exposure_complete
)
)
entry.async_on_unload(
self.client.register_callback("sensor_update", self._handle_sensor_update)
)
entry.async_on_unload(unsub)
entry.async_on_unload(self.client.disconnect)
super().__init__(
@@ -60,7 +66,80 @@ class IndiAllSkyDataUpdateCoordinator(DataUpdateCoordinator[IndiAllSkyData]):
def _handle_exposure_complete(self, exposure: ExposureData) -> None:
"""Handle new exposure_complete event from WebSocket stream."""
self.latest_exposure = exposure
self.async_set_updated_data(IndiAllSkyData(exposure=exposure))
self.async_set_updated_data(
IndiAllSkyData(
exposure=exposure,
sensor=self.latest_sensor,
)
)
def _handle_sensor_update(self, sensor: SensorData) -> None:
"""Handle new sensor_update event from WebSocket stream."""
if self.latest_sensor is not None:
merged_sensors = dict(self.latest_sensor.sensors)
merged_sensors.update(sensor.sensors)
merged_temp = (
list(sensor.raw_temp)
if sensor.raw_temp
else list(self.latest_sensor.raw_temp)
)
merged_user = (
list(sensor.raw_user)
if sensor.raw_user
else list(self.latest_sensor.raw_user)
)
merged_data = dict(self.latest_sensor.raw_data or {})
if sensor.raw_data:
merged_data.update(sensor.raw_data)
self.latest_sensor = SensorData(
last_update=sensor.last_update or self.latest_sensor.last_update,
sensors=merged_sensors,
raw_temp=merged_temp,
raw_user=merged_user,
raw_data=merged_data,
dew_heater=sensor.dew_heater
if sensor.dew_heater is not None
else self.latest_sensor.dew_heater,
dew_point=sensor.dew_point
if sensor.dew_point is not None
else self.latest_sensor.dew_point,
frost_point=sensor.frost_point
if sensor.frost_point is not None
else self.latest_sensor.frost_point,
fan_duty_cycle=sensor.fan_duty_cycle
if sensor.fan_duty_cycle is not None
else self.latest_sensor.fan_duty_cycle,
heat_index=sensor.heat_index
if sensor.heat_index is not None
else self.latest_sensor.heat_index,
wind_direction=sensor.wind_direction
if sensor.wind_direction is not None
else self.latest_sensor.wind_direction,
device_sqm=sensor.device_sqm
if sensor.device_sqm is not None
else self.latest_sensor.device_sqm,
camera_sqm=sensor.camera_sqm
if sensor.camera_sqm is not None
else self.latest_sensor.camera_sqm,
camera_sqm_adu=sensor.camera_sqm_adu
if sensor.camera_sqm_adu is not None
else self.latest_sensor.camera_sqm_adu,
cpu_temperature=sensor.cpu_temperature
if sensor.cpu_temperature is not None
else self.latest_sensor.cpu_temperature,
)
else:
self.latest_sensor = sensor
self.async_set_updated_data(
IndiAllSkyData(
exposure=self.latest_exposure,
sensor=self.latest_sensor,
)
)
@override
async def _async_update_data(self) -> IndiAllSkyData:
@@ -69,10 +148,15 @@ class IndiAllSkyDataUpdateCoordinator(DataUpdateCoordinator[IndiAllSkyData]):
await self.client.fetch_image("latestimage")
if not self.client.is_connected:
await self.client.connect()
if self.latest_sensor is None:
await self.client.fetch_sensors()
except IndiAllSkyError as err:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="update_failed",
) from err
return IndiAllSkyData(exposure=self.latest_exposure)
return IndiAllSkyData(
exposure=self.latest_exposure,
sensor=self.latest_sensor,
)
homeassistant/components/indi_allsky/icons.json
+21
View File
@@ -0,0 +1,21 @@
{
"entity": {
"sensor": {
"camera_sqm": {
"default": "mdi:weather-night"
},
"camera_sqm_adu": {
"default": "mdi:counter"
},
"device_sqm": {
"default": "mdi:weather-night"
},
"dew_heater": {
"default": "mdi:heating-coil"
},
"fan_duty_cycle": {
"default": "mdi:fan"
}
}
}
}
homeassistant/components/indi_allsky/sensor.py
+301 -8
View File
@@ -1,18 +1,27 @@
"""Support for INDI Allsky sensors."""
from collections.abc import Callable
from contextlib import suppress
from dataclasses import dataclass
from datetime import datetime
from typing import override
import re
from typing import Any, override
from homeassistant.components.sensor import (
DEVICE_CLASS_UNITS,
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import EntityCategory, UnitOfTemperature, UnitOfTime
from homeassistant.core import HomeAssistant
from homeassistant.const import (
DEGREE,
PERCENTAGE,
EntityCategory,
UnitOfTemperature,
UnitOfTime,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.util import dt as dt_util
@@ -38,12 +47,12 @@ def _parse_timestamp(data: IndiAllSkyData) -> datetime | None:
@dataclass(frozen=True, kw_only=True)
class IndiAllSkySensorEntityDescription(SensorEntityDescription):
"""Class describing INDI Allsky sensor entities."""
"""Class describing INDI Allsky hardware sensor entities."""
value_fn: Callable[[IndiAllSkyData], StateType | datetime]
SENSOR_DESCRIPTIONS: tuple[IndiAllSkySensorEntityDescription, ...] = (
PREDEFINED_SENSOR_DESCRIPTIONS: tuple[IndiAllSkySensorEntityDescription, ...] = (
IndiAllSkySensorEntityDescription(
key="binmode",
translation_key="binmode",
@@ -108,9 +117,189 @@ SENSOR_DESCRIPTIONS: tuple[IndiAllSkySensorEntityDescription, ...] = (
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.exposure.temp if data.exposure else None,
),
IndiAllSkySensorEntityDescription(
key="dew_heater",
translation_key="dew_heater",
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.sensor.dew_heater if data.sensor else None,
),
IndiAllSkySensorEntityDescription(
key="dewpoint",
translation_key="dewpoint",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.sensor.dew_point if data.sensor else None,
),
IndiAllSkySensorEntityDescription(
key="frost_point",
translation_key="frost_point",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.sensor.frost_point if data.sensor else None,
),
IndiAllSkySensorEntityDescription(
key="fan_duty_cycle",
translation_key="fan_duty_cycle",
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.sensor.fan_duty_cycle if data.sensor else None,
),
IndiAllSkySensorEntityDescription(
key="heat_index",
translation_key="heat_index",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.sensor.heat_index if data.sensor else None,
),
IndiAllSkySensorEntityDescription(
key="wind_direction",
translation_key="wind_direction",
device_class=SensorDeviceClass.WIND_DIRECTION,
native_unit_of_measurement=DEGREE,
state_class=SensorStateClass.MEASUREMENT_ANGLE,
value_fn=lambda data: data.sensor.wind_direction if data.sensor else None,
),
IndiAllSkySensorEntityDescription(
key="device_sqm",
translation_key="device_sqm",
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.sensor.device_sqm if data.sensor else None,
),
IndiAllSkySensorEntityDescription(
key="camera_sqm",
translation_key="camera_sqm",
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.sensor.camera_sqm if data.sensor else None,
),
IndiAllSkySensorEntityDescription(
key="camera_sqm_adu",
translation_key="camera_sqm_adu",
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.sensor.camera_sqm_adu if data.sensor else None,
),
IndiAllSkySensorEntityDescription(
key="cpu_temperature",
translation_key="cpu_temperature",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: data.sensor.cpu_temperature if data.sensor else None,
),
)
IGNORED_DYNAMIC_KEYS: set[str] = {
"adu",
"analog_to_digital_unit",
"binmode",
"binning_mode",
"camera_sensor_temp",
"camera_sqm",
"camera_sqm_adu",
"camera_sqm_magnitude",
"camera_temp",
"cpu_temp",
"cpu_temperature",
"device_sqm",
"device_sqm_magnitude",
"dew_heater",
"dew_heater_duty_cycle",
"dew_heater_level",
"dew_heater_output",
"dew_point",
"dewpoint",
"exposure",
"exposure_time",
"fan_duty_cycle",
"fan_level",
"fan_speed",
"fan_speed_level",
"fan_state",
"filename",
"frost_point",
"gain",
"heat_index",
"night",
"sky_quality",
"sqm",
"sqm_adu",
"stars",
"temp",
"wind_dir",
"wind_direction",
}
DEVICE_CLASS_KEYWORDS: tuple[tuple[tuple[str, ...], SensorDeviceClass], ...] = (
(
("temperature", "temp", "dewpoint", "dew point", "heat index", "frost point"),
SensorDeviceClass.TEMPERATURE,
),
(("humidity",), SensorDeviceClass.HUMIDITY),
(("pressure",), SensorDeviceClass.PRESSURE),
(("wind speed", "wind gust"), SensorDeviceClass.WIND_SPEED),
(("wind direction", "wind dir"), SensorDeviceClass.WIND_DIRECTION),
(("voltage", "volt"), SensorDeviceClass.VOLTAGE),
(("current", "amp"), SensorDeviceClass.CURRENT),
(("power", "watt"), SensorDeviceClass.POWER),
(("lux", "illuminance"), SensorDeviceClass.ILLUMINANCE),
(("duration",), SensorDeviceClass.DURATION),
)
KNOWN_TRANSLATION_KEYS: set[str] = {
"ambient_temperature",
"humidity",
"pressure",
}
def _infer_sensor_metadata(
key: str,
raw_name: str | None,
raw_device_class: str | None,
raw_unit: str | None,
) -> tuple[str, SensorDeviceClass | None, str | None, SensorStateClass | None]:
"""Infer entity name, device class, native unit, and state class for dynamic hardware sensors."""
device_class: SensorDeviceClass | None = None
if raw_device_class:
with suppress(ValueError):
device_class = SensorDeviceClass(raw_device_class.lower())
name_str = raw_name or key.replace("_", " ").capitalize()
tokens = re.findall(r"[a-z0-9]+", f"{key} {name_str}".lower())
normalized_phrase = f" {' '.join(tokens)} "
if device_class is None:
for keywords, candidate_class in DEVICE_CLASS_KEYWORDS:
if any(f" {kw} " in normalized_phrase for kw in keywords):
device_class = candidate_class
break
unit = raw_unit
if (
device_class is not None
and (valid_units := DEVICE_CLASS_UNITS.get(device_class)) is not None
and unit not in valid_units
):
device_class = None
if device_class == SensorDeviceClass.WIND_DIRECTION:
state_class = SensorStateClass.MEASUREMENT_ANGLE
elif device_class is not None or unit is not None:
state_class = SensorStateClass.MEASUREMENT
else:
state_class = None
return name_str, device_class, unit, state_class
async def async_setup_entry(
hass: HomeAssistant,
entry: IndiAllSkyConfigEntry,
@@ -118,14 +307,46 @@ async def async_setup_entry(
) -> None:
"""Set up INDI Allsky sensors based on a config entry."""
coordinator = entry.runtime_data
async_add_entities(
entities: list[SensorEntity] = [
IndiAllSkySensor(coordinator, entry, description)
for description in SENSOR_DESCRIPTIONS
for description in PREDEFINED_SENSOR_DESCRIPTIONS
]
registered_keys: set[str] = {
desc.key for desc in PREDEFINED_SENSOR_DESCRIPTIONS
} | IGNORED_DYNAMIC_KEYS
@callback
def _async_update_hardware_sensors() -> None:
"""Register newly discovered hardware sensors from coordinator data."""
if not coordinator.data.sensor:
return
new_entities: list[SensorEntity] = []
sensors_dict = coordinator.data.sensor.sensors or {}
for key in sensors_dict:
if key in registered_keys:
continue
registered_keys.add(key)
new_entities.append(
IndiAllSkyDynamicHardwareSensor(coordinator, entry, key)
)
if new_entities:
async_add_entities(new_entities)
_async_update_hardware_sensors()
entry.async_on_unload(
coordinator.async_add_listener(_async_update_hardware_sensors)
)
async_add_entities(entities)
class IndiAllSkySensor(IndiAllSkyEntity, SensorEntity):
"""Representation of an INDI Allsky sensor."""
"""Representation of an INDI Allsky predefined sensor."""
entity_description: IndiAllSkySensorEntityDescription
@@ -145,3 +366,75 @@ class IndiAllSkySensor(IndiAllSkyEntity, SensorEntity):
def native_value(self) -> StateType | datetime:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self.coordinator.data)
class IndiAllSkyDynamicHardwareSensor(IndiAllSkyEntity, SensorEntity):
"""Representation of a dynamic hardware sensor exposed by INDI Allsky."""
def __init__(
self,
coordinator: IndiAllSkyDataUpdateCoordinator,
entry: IndiAllSkyConfigEntry,
sensor_key: str,
) -> None:
"""Initialize the dynamic hardware sensor."""
super().__init__(coordinator, entry)
self._sensor_key = sensor_key
self._attr_unique_id = f"{entry.entry_id}_{sensor_key}"
self._update_attributes()
def _get_sensor_item(self) -> Any:
if self.coordinator.data.sensor and isinstance(
self.coordinator.data.sensor.sensors, dict
):
return self.coordinator.data.sensor.sensors.get(self._sensor_key)
return None
def _update_attributes(self) -> None:
item = self._get_sensor_item()
raw_name: str | None = None
raw_device_class: str | None = None
raw_unit: str | None = None
if isinstance(item, dict):
raw_name = item.get("name") or item.get("label")
raw_device_class = item.get("device_class")
raw_unit = item.get("unit")
name_str, device_class, unit, state_class = _infer_sensor_metadata(
self._sensor_key, raw_name, raw_device_class, raw_unit
)
if self._sensor_key in KNOWN_TRANSLATION_KEYS:
self._attr_translation_key = self._sensor_key
elif "translation_key" not in self.__dict__ and not hasattr(
self, "_attr_translation_key"
):
self._attr_name = name_str
self._attr_device_class = device_class
self._attr_native_unit_of_measurement = unit
self._attr_state_class = state_class
@property
@override
def native_value(self) -> StateType:
"""Return the state of the sensor."""
if not self.coordinator.data.sensor:
return None
item = self._get_sensor_item()
if item is not None:
if isinstance(item, dict):
val = item.get("value")
else:
val = item
if val is not None:
with suppress(ValueError, TypeError):
return float(val)
if self.state_class is None:
return str(val)
return None
return None
homeassistant/components/indi_allsky/strings.json
+39
View File
@@ -33,6 +33,9 @@
}
},
"sensor": {
"ambient_temperature": {
"name": "Ambient temperature"
},
"binmode": {
"name": "Binning mode"
},
@@ -42,23 +45,59 @@
"camera_sensor_temp": {
"name": "Camera sensor temperature"
},
"camera_sqm": {
"name": "Camera SQM"
},
"camera_sqm_adu": {
"name": "Camera SQM ADU"
},
"cpu_temperature": {
"name": "CPU temperature"
},
"create_date": {
"name": "Exposure creation time"
},
"device_sqm": {
"name": "Device SQM"
},
"dew_heater": {
"name": "Dew heater duty cycle"
},
"dewpoint": {
"name": "Dew point"
},
"exposure": {
"name": "Exposure time"
},
"fan_duty_cycle": {
"name": "Fan duty cycle"
},
"filename": {
"name": "Filename"
},
"frost_point": {
"name": "Frost point"
},
"gain": {
"name": "Gain"
},
"heat_index": {
"name": "Heat index"
},
"humidity": {
"name": "Humidity"
},
"pressure": {
"name": "Pressure"
},
"sqm": {
"name": "Sky quality"
},
"stars": {
"name": "Stars"
},
"wind_direction": {
"name": "Wind direction"
}
}
},
tests/components/indi_allsky/conftest.py
+29 -1
View File
@@ -4,7 +4,7 @@ from collections.abc import Callable, Generator
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
from aioindiallsky import ExposureData
from aioindiallsky import ExposureData, SensorData
import pytest
from homeassistant.components.indi_allsky.const import DOMAIN
@@ -70,6 +70,34 @@ def mock_exposure_data() -> ExposureData:
return ExposureData.from_dict(raw_data)
@pytest.fixture
def mock_sensor_data() -> SensorData:
"""Fixture to provide sample SensorData."""
return SensorData.from_dict(
{
"last_update": "2026-08-13 22:53:41",
"sensors": {
"ambient_temperature": 21.5,
"humidity": 65.0,
"pressure": 1013.25,
},
"sensor_user": [
0,
50.0,
14.8,
10.2,
75.0,
22.1,
180.0,
21.4,
20.8,
15000.0,
],
"sensor_temp": [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 45.2],
}
)
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Fixture to cleanly create an INDI Allsky configuration entry."""
tests/components/indi_allsky/snapshots/test_sensor.ambr
+554
View File
@@ -157,6 +157,335 @@
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_camera_sqm-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.indi_allsky_camera_sqm',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Camera SQM',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Camera SQM',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'camera_sqm',
'unique_id': '1234567890abcdef1234567890abcdef_camera_sqm',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_camera_sqm-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Camera SQM',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_camera_sqm',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_camera_sqm_adu-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.indi_allsky_camera_sqm_adu',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Camera SQM ADU',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Camera SQM ADU',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'camera_sqm_adu',
'unique_id': '1234567890abcdef1234567890abcdef_camera_sqm_adu',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_camera_sqm_adu-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Camera SQM ADU',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_camera_sqm_adu',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_cpu_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.indi_allsky_cpu_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'CPU temperature',
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'CPU temperature',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'cpu_temperature',
'unique_id': '1234567890abcdef1234567890abcdef_cpu_temperature',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_cpu_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'temperature',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky CPU temperature',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_cpu_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_device_sqm-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.indi_allsky_device_sqm',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Device SQM',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Device SQM',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'device_sqm',
'unique_id': '1234567890abcdef1234567890abcdef_device_sqm',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_device_sqm-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Device SQM',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_device_sqm',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_dew_heater_duty_cycle-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.indi_allsky_dew_heater_duty_cycle',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Dew heater duty cycle',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Dew heater duty cycle',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'dew_heater',
'unique_id': '1234567890abcdef1234567890abcdef_dew_heater',
'unit_of_measurement': '%',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_dew_heater_duty_cycle-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Dew heater duty cycle',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: '%',
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_dew_heater_duty_cycle',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_dew_point-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.indi_allsky_dew_point',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Dew point',
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Dew point',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'dewpoint',
'unique_id': '1234567890abcdef1234567890abcdef_dewpoint',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_dew_point-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'temperature',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Dew point',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_dew_point',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_exposure_creation_time-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -266,6 +595,60 @@
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_fan_duty_cycle-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.indi_allsky_fan_duty_cycle',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Fan duty cycle',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Fan duty cycle',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'fan_duty_cycle',
'unique_id': '1234567890abcdef1234567890abcdef_fan_duty_cycle',
'unit_of_measurement': '%',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_fan_duty_cycle-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Fan duty cycle',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: '%',
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_fan_duty_cycle',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_filename-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -316,6 +699,64 @@
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_frost_point-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.indi_allsky_frost_point',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Frost point',
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Frost point',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'frost_point',
'unique_id': '1234567890abcdef1234567890abcdef_frost_point',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_frost_point-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'temperature',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Frost point',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_frost_point',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_gain-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -366,6 +807,64 @@
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_heat_index-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.indi_allsky_heat_index',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Heat index',
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Heat index',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'heat_index',
'unique_id': '1234567890abcdef1234567890abcdef_heat_index',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_heat_index-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'temperature',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Heat index',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_heat_index',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_sky_quality-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -472,3 +971,58 @@
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_wind_direction-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT_ANGLE: 'measurement_angle'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.indi_allsky_wind_direction',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Wind direction',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.WIND_DIRECTION: 'wind_direction'>,
'original_icon': None,
'original_name': 'Wind direction',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'wind_direction',
'unique_id': '1234567890abcdef1234567890abcdef_wind_direction',
'unit_of_measurement': '°',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_wind_direction-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'wind_direction',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Wind direction',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT_ANGLE: 'measurement_angle'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: '°',
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_wind_direction',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
tests/components/indi_allsky/test_sensor.py
+319 -4
View File
@@ -3,7 +3,7 @@
from dataclasses import replace
from unittest.mock import AsyncMock, patch
from aioindiallsky import ExposureData
from aioindiallsky import ExposureData, SensorData
import pytest
from syrupy.assertion import SnapshotAssertion
@@ -46,6 +46,7 @@ async def test_disabled_sensors(
for entity_id in (
"sensor.indi_allsky_binning_mode",
"sensor.indi_allsky_camera_id",
"sensor.indi_allsky_cpu_temperature",
"sensor.indi_allsky_exposure_creation_time",
"sensor.indi_allsky_filename",
"sensor.indi_allsky_gain",
@@ -55,8 +56,9 @@ async def test_disabled_sensors(
assert entry.disabled_by is er.RegistryEntryDisabler.INTEGRATION
for entity_id in (
"sensor.indi_allsky_exposure_time",
"sensor.indi_allsky_camera_sensor_temperature",
"sensor.indi_allsky_dew_heater_duty_cycle",
"sensor.indi_allsky_exposure_time",
"sensor.indi_allsky_sky_quality",
"sensor.indi_allsky_stars",
):
@@ -65,7 +67,7 @@ async def test_disabled_sensors(
assert entry.disabled_by is None
async def test_sensor_updates(
async def test_exposure_sensor_updates(
hass: HomeAssistant,
mock_indi_allsky_client: AsyncMock,
mock_config_entry: MockConfigEntry,
@@ -73,7 +75,6 @@ async def test_sensor_updates(
entity_registry: er.EntityRegistry,
) -> None:
"""Test sensor state values update on exposure_complete event."""
# Enable disabled sensors for testing
entity_registry.async_get_or_create(
domain="sensor",
platform="indi_allsky",
@@ -160,3 +161,317 @@ async def test_sensor_updates(
state = hass.states.get("sensor.indi_allsky_camera_sensor_temperature")
assert state is not None
assert state.state == "12.5"
async def test_hardware_sensor_updates(
hass: HomeAssistant,
mock_indi_allsky_client: AsyncMock,
mock_config_entry: MockConfigEntry,
mock_sensor_data: SensorData,
entity_registry: er.EntityRegistry,
) -> None:
"""Test hardware sensor state values update on sensor_update event."""
entity_registry.async_get_or_create(
domain="sensor",
platform="indi_allsky",
unique_id=f"{mock_config_entry.entry_id}_cpu_temperature",
suggested_object_id="indi_allsky_cpu_temperature",
disabled_by=None,
)
with patch("homeassistant.components.indi_allsky._PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_config_entry)
for callback in mock_indi_allsky_client.callbacks.get("sensor_update", []):
callback(mock_sensor_data)
await hass.async_block_till_done()
state = hass.states.get("sensor.indi_allsky_ambient_temperature")
assert state is not None
assert state.state == "21.5"
state = hass.states.get("sensor.indi_allsky_humidity")
assert state is not None
assert state.state == "65.0"
state = hass.states.get("sensor.indi_allsky_pressure")
assert state is not None
assert state.state == "1013.25"
state = hass.states.get("sensor.indi_allsky_dew_heater_duty_cycle")
assert state is not None
assert state.state == "50.0"
state = hass.states.get("sensor.indi_allsky_dew_point")
assert state is not None
assert state.state == "14.8"
state = hass.states.get("sensor.indi_allsky_frost_point")
assert state is not None
assert state.state == "10.2"
state = hass.states.get("sensor.indi_allsky_fan_duty_cycle")
assert state is not None
assert state.state == "75.0"
state = hass.states.get("sensor.indi_allsky_heat_index")
assert state is not None
assert state.state == "22.1"
state = hass.states.get("sensor.indi_allsky_wind_direction")
assert state is not None
assert state.state == "180.0"
state = hass.states.get("sensor.indi_allsky_device_sqm")
assert state is not None
assert state.state == "21.4"
state = hass.states.get("sensor.indi_allsky_camera_sqm")
assert state is not None
assert state.state == "20.8"
state = hass.states.get("sensor.indi_allsky_camera_sqm_adu")
assert state is not None
assert state.state == "15000.0"
state = hass.states.get("sensor.indi_allsky_cpu_temperature")
assert state is not None
assert state.state == "45.2"
async def test_hardware_sensor_fallback_updates(
hass: HomeAssistant,
mock_indi_allsky_client: AsyncMock,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test hardware sensor fallback resolution from raw_user and raw_temp slots."""
entity_registry.async_get_or_create(
domain="sensor",
platform="indi_allsky",
unique_id=f"{mock_config_entry.entry_id}_cpu_temperature",
suggested_object_id="indi_allsky_cpu_temperature",
disabled_by=None,
)
with patch("homeassistant.components.indi_allsky._PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_config_entry)
fallback_data = SensorData.from_dict(
{
"sensors": {},
"sensor_temp": [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 57.5],
"sensor_user": [0, 0, 0, 0, 0, 0, 0, 0, 0, 12500.5],
}
)
for callback in mock_indi_allsky_client.callbacks.get("sensor_update", []):
callback(fallback_data)
await hass.async_block_till_done()
state = hass.states.get("sensor.indi_allsky_cpu_temperature")
assert state is not None
assert state.state == "57.5"
state = hass.states.get("sensor.indi_allsky_camera_sqm_adu")
assert state is not None
assert state.state == "12500.5"
async def test_consecutive_sensor_updates_merging(
hass: HomeAssistant,
mock_indi_allsky_client: AsyncMock,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test consecutive sensor updates merge retaining previous values."""
entity_registry.async_get_or_create(
domain="sensor",
platform="indi_allsky",
unique_id=f"{mock_config_entry.entry_id}_cpu_temperature",
suggested_object_id="indi_allsky_cpu_temperature",
disabled_by=None,
)
with patch("homeassistant.components.indi_allsky._PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_config_entry)
first_update = SensorData.from_dict(
{
"sensors": {
"humidity": 65.0,
},
"sensor_user": [0, 0, 14.8],
"sensor_temp": [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 42.0],
}
)
for callback in mock_indi_allsky_client.callbacks.get("sensor_update", []):
callback(first_update)
await hass.async_block_till_done()
state = hass.states.get("sensor.indi_allsky_humidity")
assert state is not None
assert state.state == "65.0"
state = hass.states.get("sensor.indi_allsky_dew_point")
assert state is not None
assert state.state == "14.8"
state = hass.states.get("sensor.indi_allsky_cpu_temperature")
assert state is not None
assert state.state == "42.0"
second_update = SensorData.from_dict(
{
"sensors": {
"humidity": 70.0,
"pressure": 1015.0,
},
}
)
for callback in mock_indi_allsky_client.callbacks.get("sensor_update", []):
callback(second_update)
await hass.async_block_till_done()
state = hass.states.get("sensor.indi_allsky_humidity")
assert state is not None
assert state.state == "70.0"
state = hass.states.get("sensor.indi_allsky_pressure")
assert state is not None
assert state.state == "1015.0"
state = hass.states.get("sensor.indi_allsky_dew_point")
assert state is not None
assert state.state == "14.8"
state = hass.states.get("sensor.indi_allsky_cpu_temperature")
assert state is not None
assert state.state == "42.0"
async def test_dynamic_hardware_sensor_discovery(
hass: HomeAssistant,
mock_indi_allsky_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test dynamic discovery and metadata inference of custom hardware sensors."""
with patch("homeassistant.components.indi_allsky._PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_config_entry)
dynamic_data = SensorData.from_dict(
{
"sensors": {
"sensor_a_temperature": {
"name": "SHT4x (i2c) - SHT40 - Temperature",
"value": 30.83,
"unit": "°C",
"device_class": "temperature",
},
"sensor_b_pressure": {
"name": "Ecowitt API - Sensor B - Pressure",
"value": 1017.27,
"unit": "hPa",
"device_class": "pressure",
},
"sensor_c_humidity": {
"name": "DHT22 - Sensor C - Humidity",
"value": 65.5,
"unit": "%",
},
"sensor_d_lux": {
"name": "TSL2561 Lux",
"value": 150.0,
"unit": "lx",
},
"sensor_e_status": {
"name": "Hardware Status",
"value": "Operational",
},
"sensor_f_no_unit_temp": {
"name": "Generic Temperature Sensor",
"value": 24.5,
},
"lamp_status": {
"name": "Dome Lamp Status",
"value": "Off",
},
}
}
)
for callback in mock_indi_allsky_client.callbacks.get("sensor_update", []):
callback(dynamic_data)
await hass.async_block_till_done()
state = hass.states.get("sensor.indi_allsky_sht4x_i2c_sht40_temperature")
assert state is not None
assert state.state == "30.83"
assert state.attributes.get("unit_of_measurement") == "°C"
assert state.attributes.get("device_class") == "temperature"
state = hass.states.get("sensor.indi_allsky_ecowitt_api_sensor_b_pressure")
assert state is not None
assert state.state == "1017.27"
assert state.attributes.get("unit_of_measurement") == "hPa"
assert state.attributes.get("device_class") == "pressure"
state = hass.states.get("sensor.indi_allsky_dht22_sensor_c_humidity")
assert state is not None
assert state.state == "65.5"
assert state.attributes.get("unit_of_measurement") == "%"
state = hass.states.get("sensor.indi_allsky_tsl2561_lux")
assert state is not None
assert state.state == "150.0"
assert state.attributes.get("unit_of_measurement") == "lx"
assert state.attributes.get("device_class") == "illuminance"
state = hass.states.get("sensor.indi_allsky_hardware_status")
assert state is not None
assert state.state == "Operational"
assert state.attributes.get("state_class") is None
# Verify sensor with no unit does not infer default unit or device class
state = hass.states.get("sensor.indi_allsky_generic_temperature_sensor")
assert state is not None
assert state.state == "24.5"
assert state.attributes.get("device_class") is None
assert state.attributes.get("unit_of_measurement") is None
assert state.attributes.get("state_class") is None
# Verify lamp_status does not incorrectly match 'amp' substring
state = hass.states.get("sensor.indi_allsky_dome_lamp_status")
assert state is not None
assert state.state == "Off"
assert state.attributes.get("device_class") is None
assert state.attributes.get("state_class") is None
assert state.attributes.get("unit_of_measurement") is None
async def test_initial_sensor_fetch_preserved(
hass: HomeAssistant,
mock_indi_allsky_client: AsyncMock,
mock_config_entry: MockConfigEntry,
mock_sensor_data: SensorData,
) -> None:
"""Test initial sensor fetch result is preserved during setup before websocket stream."""
def _fetch_sensors_side_effect() -> bool:
for cb in mock_indi_allsky_client.callbacks.get("sensor_update", []):
cb(mock_sensor_data)
return True
mock_indi_allsky_client.fetch_sensors.side_effect = _fetch_sensors_side_effect
with patch("homeassistant.components.indi_allsky._PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_config_entry)
# Ambient temperature and dew point should be immediately present
state = hass.states.get("sensor.indi_allsky_ambient_temperature")
assert state is not None
assert state.state == "21.5"
state = hass.states.get("sensor.indi_allsky_dew_point")
assert state is not None
assert state.state == "14.8"