Add extra sensors to Swiss Public Transport (#114636)

* convert extra_state_attributes to sensors

* add deprecation notice for extra state attributes

* cleanup after comments

* remove exists_fn as it does not add value

* move function outside the class
This commit is contained in:
Cyrill Raccaud 2024-04-02 21:35:11 +02:00 committed by GitHub
parent 17f0002549
commit bf9627ad07
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 80 additions and 28 deletions

View File

@ -25,16 +25,24 @@ class DataConnection(TypedDict):
departure: datetime | None
next_departure: datetime | None
next_on_departure: datetime | None
duration: str
duration: int | None
platform: str
remaining_time: str
start: str
destination: str
train_number: str
transfers: str
transfers: int
delay: int
def calculate_duration_in_seconds(duration_text: str) -> int | None:
"""Transform and calculate the duration into seconds."""
# Transform 01d03:21:23 into 01 days 03:21:23
duration_text_pg_format = duration_text.replace("d", " days ")
duration = dt_util.parse_duration(duration_text_pg_format)
return duration.seconds if duration else None
class SwissPublicTransportDataUpdateCoordinator(
DataUpdateCoordinator[list[DataConnection]]
):
@ -77,7 +85,6 @@ class SwissPublicTransportDataUpdateCoordinator(
raise UpdateFailed from e
connections = self._opendata.connections
return [
DataConnection(
departure=self.nth_departure_time(i),
@ -86,7 +93,7 @@ class SwissPublicTransportDataUpdateCoordinator(
train_number=connections[i]["number"],
platform=connections[i]["platform"],
transfers=connections[i]["transfers"],
duration=connections[i]["duration"],
duration=calculate_duration_in_seconds(connections[i]["duration"]),
start=self._opendata.from_name,
destination=self._opendata.to_name,
remaining_time=str(self.remaining_time(connections[i]["departure"])),

View File

@ -1,8 +1,26 @@
{
"entity": {
"sensor": {
"departure": {
"default": "mdi:bus"
"departure0": {
"default": "mdi:bus-clock"
},
"departure1": {
"default": "mdi:bus-clock"
},
"departure2": {
"default": "mdi:bus-clock"
},
"duration": {
"default": "mdi:timeline-clock"
},
"transfers": {
"default": "mdi:transit-transfer"
},
"platform": {
"default": "mdi:bus-stop-uncovered"
},
"delay": {
"default": "mdi:clock-plus"
}
}
}

View File

@ -18,14 +18,14 @@ from homeassistant.components.sensor import (
SensorEntityDescription,
)
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import CONF_NAME
from homeassistant.const import CONF_NAME, UnitOfTime
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResultType
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
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.typing import ConfigType, DiscoveryInfoType, StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import (
@ -55,11 +55,10 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
class SwissPublicTransportSensorEntityDescription(SensorEntityDescription):
"""Describes swiss public transport sensor entity."""
exists_fn: Callable[[DataConnection], bool]
value_fn: Callable[[DataConnection], datetime | None]
value_fn: Callable[[DataConnection], StateType | datetime]
index: int
has_legacy_attributes: bool
index: int = 0
has_legacy_attributes: bool = False
SENSORS: tuple[SwissPublicTransportSensorEntityDescription, ...] = (
@ -70,11 +69,33 @@ SENSORS: tuple[SwissPublicTransportSensorEntityDescription, ...] = (
device_class=SensorDeviceClass.TIMESTAMP,
has_legacy_attributes=i == 0,
value_fn=lambda data_connection: data_connection["departure"],
exists_fn=lambda data_connection: data_connection is not None,
index=i,
)
for i in range(SENSOR_CONNECTIONS_COUNT)
],
SwissPublicTransportSensorEntityDescription(
key="duration",
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement=UnitOfTime.SECONDS,
value_fn=lambda data_connection: data_connection["duration"],
),
SwissPublicTransportSensorEntityDescription(
key="transfers",
translation_key="transfers",
value_fn=lambda data_connection: data_connection["transfers"],
),
SwissPublicTransportSensorEntityDescription(
key="platform",
translation_key="platform",
value_fn=lambda data_connection: data_connection["platform"],
),
SwissPublicTransportSensorEntityDescription(
key="delay",
translation_key="delay",
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement=UnitOfTime.MINUTES,
value_fn=lambda data_connection: data_connection["delay"],
),
)
@ -167,14 +188,7 @@ class SwissPublicTransportSensor(
)
@property
def enabled(self) -> bool:
"""Enable the sensor if data is available."""
return self.entity_description.exists_fn(
self.coordinator.data[self.entity_description.index]
)
@property
def native_value(self) -> datetime | None:
def native_value(self) -> StateType | datetime:
"""Return the state of the sensor."""
return self.entity_description.value_fn(
self.coordinator.data[self.entity_description.index]
@ -196,10 +210,11 @@ class SwissPublicTransportSensor(
@callback
def _async_update_attrs(self) -> None:
"""Update the extra state attributes based on the coordinator data."""
self._attr_extra_state_attributes = {
key: value
for key, value in self.coordinator.data[
self.entity_description.index
].items()
if key not in {"departure"}
}
if self.entity_description.has_legacy_attributes:
self._attr_extra_state_attributes = {
key: value
for key, value in self.coordinator.data[
self.entity_description.index
].items()
if key not in {"departure"}
}

View File

@ -32,6 +32,18 @@
},
"departure2": {
"name": "Departure +2"
},
"duration": {
"name": "Duration"
},
"transfers": {
"name": "Transfers"
},
"platform": {
"name": "Platform"
},
"delay": {
"name": "Delay"
}
}
},