Move Ping binary sensor attributes to sensor entities (#112004)

* Move Ping binary sensor attributes to sensor entities

* Process code review

* Update snapshot
This commit is contained in:
Jan-Philipp Benecke 2024-03-03 11:08:28 +01:00 committed by GitHub
parent 6a243d6705
commit 25551fa938
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 366 additions and 22 deletions

View File

@ -19,7 +19,7 @@ from .helpers import PingDataICMPLib, PingDataSubProcess
_LOGGER = logging.getLogger(__name__)
CONFIG_SCHEMA = cv.platform_only_config_schema(DOMAIN)
PLATFORMS = [Platform.BINARY_SENSOR, Platform.DEVICE_TRACKER]
PLATFORMS = [Platform.BINARY_SENSOR, Platform.DEVICE_TRACKER, Platform.SENSOR]
@dataclass(slots=True)

View File

@ -18,11 +18,11 @@ import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import PingDomainData
from .const import CONF_IMPORTED_BY, CONF_PING_COUNT, DEFAULT_PING_COUNT, DOMAIN
from .coordinator import PingUpdateCoordinator
from .entity import PingEntity
_LOGGER = logging.getLogger(__name__)
@ -84,20 +84,18 @@ async def async_setup_entry(
async_add_entities([PingBinarySensor(entry, data.coordinators[entry.entry_id])])
class PingBinarySensor(CoordinatorEntity[PingUpdateCoordinator], BinarySensorEntity):
class PingBinarySensor(PingEntity, BinarySensorEntity):
"""Representation of a Ping Binary sensor."""
_attr_device_class = BinarySensorDeviceClass.CONNECTIVITY
_attr_available = False
_attr_name = None
def __init__(
self, config_entry: ConfigEntry, coordinator: PingUpdateCoordinator
) -> None:
"""Initialize the Ping Binary sensor."""
super().__init__(coordinator)
self._attr_name = config_entry.title
self._attr_unique_id = config_entry.entry_id
super().__init__(coordinator, config_entry.entry_id)
# if this was imported just enable it when it was enabled before
if CONF_IMPORTED_BY in config_entry.data:
@ -113,11 +111,9 @@ class PingBinarySensor(CoordinatorEntity[PingUpdateCoordinator], BinarySensorEnt
@property
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes of the ICMP checo request."""
if self.coordinator.data.data is None:
return None
return {
ATTR_ROUND_TRIP_TIME_AVG: self.coordinator.data.data["avg"],
ATTR_ROUND_TRIP_TIME_MAX: self.coordinator.data.data["max"],
ATTR_ROUND_TRIP_TIME_MDEV: self.coordinator.data.data["mdev"],
ATTR_ROUND_TRIP_TIME_MIN: self.coordinator.data.data["min"],
ATTR_ROUND_TRIP_TIME_AVG: self.coordinator.data.data.get("avg"),
ATTR_ROUND_TRIP_TIME_MAX: self.coordinator.data.data.get("max"),
ATTR_ROUND_TRIP_TIME_MDEV: self.coordinator.data.data.get("mdev"),
ATTR_ROUND_TRIP_TIME_MIN: self.coordinator.data.data.get("min"),
}

View File

@ -20,7 +20,7 @@ class PingResult:
ip_address: str
is_alive: bool
data: dict[str, Any] | None
data: dict[str, Any]
class PingUpdateCoordinator(DataUpdateCoordinator[PingResult]):
@ -49,5 +49,5 @@ class PingUpdateCoordinator(DataUpdateCoordinator[PingResult]):
return PingResult(
ip_address=self.ping.ip_address,
is_alive=self.ping.is_alive,
data=self.ping.data,
data=self.ping.data or {},
)

View File

@ -0,0 +1,26 @@
"""Base entity for the Ping component."""
from homeassistant.core import DOMAIN
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .coordinator import PingUpdateCoordinator
class PingEntity(CoordinatorEntity[PingUpdateCoordinator]):
"""Represents a Ping base entity."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: PingUpdateCoordinator,
unique_id: str,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self._attr_unique_id = unique_id
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, self.coordinator.data.ip_address)},
manufacturer="Ping",
)

View File

@ -70,7 +70,6 @@ class PingDataICMPLib(PingData):
"min": data.min_rtt,
"max": data.max_rtt,
"avg": data.avg_rtt,
"mdev": "",
}
@ -135,7 +134,7 @@ class PingDataSubProcess(PingData):
if TYPE_CHECKING:
assert match is not None
rtt_min, rtt_avg, rtt_max = match.groups()
return {"min": rtt_min, "avg": rtt_avg, "max": rtt_max, "mdev": ""}
return {"min": rtt_min, "avg": rtt_avg, "max": rtt_max}
match = PING_MATCHER.search(str(out_data).rsplit("\n", maxsplit=1)[-1])
if TYPE_CHECKING:
assert match is not None

View File

@ -0,0 +1,116 @@
"""Sensor platform that for Ping integration."""
from collections.abc import Callable
from dataclasses import dataclass
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory, UnitOfTime
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import PingDomainData
from .const import DOMAIN
from .coordinator import PingResult, PingUpdateCoordinator
from .entity import PingEntity
@dataclass(frozen=True, kw_only=True)
class PingSensorEntityDescription(SensorEntityDescription):
"""Class to describe a Ping sensor entity."""
value_fn: Callable[[PingResult], float | None]
has_fn: Callable[[PingResult], bool]
SENSORS: tuple[PingSensorEntityDescription, ...] = (
PingSensorEntityDescription(
key="round_trip_time_avg",
translation_key="round_trip_time_avg",
native_unit_of_measurement=UnitOfTime.MILLISECONDS,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.DURATION,
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda result: result.data.get("avg"),
has_fn=lambda result: "avg" in result.data,
),
PingSensorEntityDescription(
key="round_trip_time_max",
translation_key="round_trip_time_max",
native_unit_of_measurement=UnitOfTime.MILLISECONDS,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.DURATION,
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda result: result.data.get("max"),
has_fn=lambda result: "max" in result.data,
),
PingSensorEntityDescription(
key="round_trip_time_mdev",
translation_key="round_trip_time_mdev",
native_unit_of_measurement=UnitOfTime.MILLISECONDS,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.DURATION,
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda result: result.data.get("mdev"),
has_fn=lambda result: "mdev" in result.data,
),
PingSensorEntityDescription(
key="round_trip_time_min",
translation_key="round_trip_time_min",
native_unit_of_measurement=UnitOfTime.MILLISECONDS,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.DURATION,
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda result: result.data.get("min"),
has_fn=lambda result: "min" in result.data,
),
)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up Ping sensors from config entry."""
data: PingDomainData = hass.data[DOMAIN]
coordinator = data.coordinators[entry.entry_id]
async_add_entities(
PingSensor(entry, description, coordinator)
for description in SENSORS
if description.has_fn(coordinator.data)
)
class PingSensor(PingEntity, SensorEntity):
"""Represents a Ping sensor."""
entity_description: PingSensorEntityDescription
def __init__(
self,
config_entry: ConfigEntry,
description: PingSensorEntityDescription,
coordinator: PingUpdateCoordinator,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, f"{config_entry.entry_id}-{description.key}")
self.entity_description = description
@property
def available(self) -> bool:
"""Return True if entity is available."""
return super().available and self.coordinator.data.is_alive
@property
def native_value(self) -> float | None:
"""Return the sensor state."""
return self.entity_description.value_fn(self.coordinator.data)

View File

@ -1,4 +1,20 @@
{
"entity": {
"sensor": {
"round_trip_time_avg": {
"name": "Round Trip Time Average"
},
"round_trip_time_max": {
"name": "Round Trip Time Maximum"
},
"round_trip_time_mdev": {
"name": "Round Trip Time Mean Deviation"
},
"round_trip_time_min": {
"name": "Round Trip Time Minimum"
}
}
},
"config": {
"step": {
"user": {

View File

@ -26,7 +26,7 @@ def patch_setup(*args, **kwargs):
@pytest.fixture(autouse=True)
async def patch_ping():
"""Patch icmplib async_ping."""
mock = Host("10.10.10.10", 5, [10, 1, 2])
mock = Host("10.10.10.10", 5, [10, 1, 2, 5, 6])
with patch(
"homeassistant.components.ping.helpers.async_ping", return_value=mock

View File

@ -72,7 +72,7 @@
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.10_10_10_10',
'has_entity_name': False,
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
@ -83,7 +83,7 @@
}),
'original_device_class': <BinarySensorDeviceClass.CONNECTIVITY: 'connectivity'>,
'original_icon': None,
'original_name': '10.10.10.10',
'original_name': None,
'platform': 'ping',
'previous_unique_id': None,
'supported_features': 0,
@ -96,9 +96,9 @@
'attributes': ReadOnlyDict({
'device_class': 'connectivity',
'friendly_name': '10.10.10.10',
'round_trip_time_avg': 4.333,
'round_trip_time_avg': 4.8,
'round_trip_time_max': 10,
'round_trip_time_mdev': '',
'round_trip_time_mdev': None,
'round_trip_time_min': 1,
}),
'context': <ANY>,
@ -113,6 +113,10 @@
'attributes': ReadOnlyDict({
'device_class': 'connectivity',
'friendly_name': '10.10.10.10',
'round_trip_time_avg': None,
'round_trip_time_max': None,
'round_trip_time_mdev': None,
'round_trip_time_min': None,
}),
'context': <ANY>,
'entity_id': 'binary_sensor.10_10_10_10',

View File

@ -0,0 +1,154 @@
# serializer version: 1
# name: test_setup_and_update[round_trip_time_average]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.10_10_10_10_round_trip_time_average',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.DURATION: 'duration'>,
'original_icon': None,
'original_name': 'Round Trip Time Average',
'platform': 'ping',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'round_trip_time_avg',
'unit_of_measurement': <UnitOfTime.MILLISECONDS: 'ms'>,
})
# ---
# name: test_setup_and_update[round_trip_time_average].1
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'duration',
'friendly_name': '10.10.10.10 Round Trip Time Average',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTime.MILLISECONDS: 'ms'>,
}),
'context': <ANY>,
'entity_id': 'sensor.10_10_10_10_round_trip_time_average',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '4.8',
})
# ---
# name: test_setup_and_update[round_trip_time_maximum]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.10_10_10_10_round_trip_time_maximum',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.DURATION: 'duration'>,
'original_icon': None,
'original_name': 'Round Trip Time Maximum',
'platform': 'ping',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'round_trip_time_max',
'unit_of_measurement': <UnitOfTime.MILLISECONDS: 'ms'>,
})
# ---
# name: test_setup_and_update[round_trip_time_maximum].1
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'duration',
'friendly_name': '10.10.10.10 Round Trip Time Maximum',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTime.MILLISECONDS: 'ms'>,
}),
'context': <ANY>,
'entity_id': 'sensor.10_10_10_10_round_trip_time_maximum',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '10',
})
# ---
# name: test_setup_and_update[round_trip_time_mean_deviation]
None
# ---
# name: test_setup_and_update[round_trip_time_mean_deviation].1
None
# ---
# name: test_setup_and_update[round_trip_time_minimum]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.10_10_10_10_round_trip_time_minimum',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.DURATION: 'duration'>,
'original_icon': None,
'original_name': 'Round Trip Time Minimum',
'platform': 'ping',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'round_trip_time_min',
'unit_of_measurement': <UnitOfTime.MILLISECONDS: 'ms'>,
})
# ---
# name: test_setup_and_update[round_trip_time_minimum].1
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'duration',
'friendly_name': '10.10.10.10 Round Trip Time Minimum',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTime.MILLISECONDS: 'ms'>,
}),
'context': <ANY>,
'entity_id': 'sensor.10_10_10_10_round_trip_time_minimum',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '1',
})
# ---

View File

@ -0,0 +1,33 @@
"""Test sensor platform of Ping."""
import pytest
from syrupy import SnapshotAssertion
from syrupy.filters import props
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "setup_integration")
@pytest.mark.parametrize(
"sensor_name",
[
"round_trip_time_average",
"round_trip_time_maximum",
"round_trip_time_mean_deviation", # should be None in the snapshot
"round_trip_time_minimum",
],
)
async def test_setup_and_update(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
sensor_name: str,
) -> None:
"""Test sensor setup and update."""
entry = entity_registry.async_get(f"sensor.10_10_10_10_{sensor_name}")
assert entry == snapshot(exclude=props("unique_id"))
state = hass.states.get(f"sensor.10_10_10_10_{sensor_name}")
assert state == snapshot