Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-24 21:06:19 +00:00
Code Activity

Add binary sensor support and refactor NS sensor integration (#154589)

Browse Source 
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: G Johansson <goran.johansson@shiftit.se>
Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
Heindrich Paul
2025-11-18 16:29:09 +01:00
committed by GitHub
parent c8c2413a09
commit 0de2a16d0f
8 changed files with 536 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/nederlandse_spoorwegen/__init__.py
View File

@@ -13,7 +13,7 @@ from .coordinator import NSConfigEntry, NSDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.SENSOR]
PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: NSConfigEntry) -> bool:

120
homeassistant/components/nederlandse_spoorwegen/binary_sensor.py Normal file
View File

@@ -0,0 +1,120 @@
"""Support for Nederlandse Spoorwegen public transport."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
import logging
from ns_api import Trip
from homeassistant.components.binary_sensor import (
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, INTEGRATION_TITLE, ROUTE_MODEL
from .coordinator import NSConfigEntry, NSDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 0 # since we use coordinator pattern
@dataclass(frozen=True, kw_only=True)
class NSBinarySensorEntityDescription(BinarySensorEntityDescription):
"""Describes Nederlandse Spoorwegen sensor entity."""
value_fn: Callable[[Trip], bool]
def get_delay(planned: datetime | None, actual: datetime | None) -> bool:
"""Return True if delay is present, False otherwise."""
return bool(planned and actual and planned != actual)
BINARY_SENSOR_DESCRIPTIONS = [
NSBinarySensorEntityDescription(
key="is_departure_delayed",
translation_key="is_departure_delayed",
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda trip: get_delay(
trip.departure_time_planned, trip.departure_time_actual
),
entity_registry_enabled_default=False,
),
NSBinarySensorEntityDescription(
key="is_arrival_delayed",
translation_key="is_arrival_delayed",
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda trip: get_delay(
trip.arrival_time_planned, trip.arrival_time_actual
),
entity_registry_enabled_default=False,
),
NSBinarySensorEntityDescription(
key="is_going",
translation_key="is_going",
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda trip: trip.going,
entity_registry_enabled_default=False,
),
]
async def async_setup_entry(
hass: HomeAssistant,
config_entry: NSConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the departure sensor from a config entry."""
coordinators = config_entry.runtime_data
for subentry_id, coordinator in coordinators.items():
async_add_entities(
(
NSBinarySensor(coordinator, subentry_id, description)
for description in BINARY_SENSOR_DESCRIPTIONS
),
config_subentry_id=subentry_id,
)
class NSBinarySensor(CoordinatorEntity[NSDataUpdateCoordinator], BinarySensorEntity):
"""Generic NS binary sensor based on entity description."""
_attr_has_entity_name = True
_attr_attribution = "Data provided by NS"
entity_description: NSBinarySensorEntityDescription
def __init__(
self,
coordinator: NSDataUpdateCoordinator,
subentry_id: str,
description: NSBinarySensorEntityDescription,
) -> None:
"""Initialize the binary sensor."""
super().__init__(coordinator)
self.entity_description = description
self._subentry_id = subentry_id
self._attr_unique_id = f"{subentry_id}-{description.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, subentry_id)},
name=coordinator.name,
manufacturer=INTEGRATION_TITLE,
model=ROUTE_MODEL,
)
@property
def is_on(self) -> bool | None:
"""Return true if the binary sensor is on."""
if not (trip := self.coordinator.data.first_trip):
return None
return self.entity_description.value_fn(trip)

15
homeassistant/components/nederlandse_spoorwegen/icons.json Normal file
View File

@@ -0,0 +1,15 @@
{
"entity": {
"binary_sensor": {
"is_arrival_delayed": {
"default": "mdi:bell-alert-outline"
},
"is_departure_delayed": {
"default": "mdi:bell-alert-outline"
},
"is_going": {
"default": "mdi:bell-cancel-outline"
}
}
}
}

6
homeassistant/components/nederlandse_spoorwegen/sensor.py
View File

@@ -155,7 +155,7 @@ async def async_setup_entry(
class NSDepartureSensor(CoordinatorEntity[NSDataUpdateCoordinator], SensorEntity):
"""Implementation of a NS Departure Sensor."""
"""Implementation of a NS Departure Sensor (legacy)."""
_attr_device_class = SensorDeviceClass.TIMESTAMP
_attr_attribution = "Data provided by NS"
@@ -202,6 +202,8 @@ class NSDepartureSensor(CoordinatorEntity[NSDataUpdateCoordinator], SensorEntity
if not first_trip:
return None
status = first_trip.status
return {
"going": first_trip.going,
"departure_time_planned": _get_time_str(first_trip.departure_time_planned),
@@ -221,7 +223,7 @@ class NSDepartureSensor(CoordinatorEntity[NSDataUpdateCoordinator], SensorEntity
"arrival_platform_planned": first_trip.arrival_platform_planned,
"arrival_platform_actual": first_trip.arrival_platform_actual,
"next": _get_time_str(_get_departure_time(next_trip)),
"status": first_trip.status.lower() if first_trip.status else None,
"status": status.lower() if status else None,
"transfers": first_trip.nr_transfers,
"route": _get_route(first_trip),
"remarks": None,

13
homeassistant/components/nederlandse_spoorwegen/strings.json
View File

@@ -64,6 +64,19 @@
}
}
},
"entity": {
"binary_sensor": {
"is_arrival_delayed": {
"name": "Arrival delayed"
},
"is_departure_delayed": {
"name": "Departure delayed"
},
"is_going": {
"name": "Going"
}
}
},
"issues": {
"deprecated_yaml_import_issue_cannot_connect": {
"description": "Configuring Nederlandse Spoorwegen using YAML sensor platform is deprecated.\n\nWhile importing your configuration, Home Assistant could not connect to the NS API. Please check your internet connection and the status of the NS API, then restart Home Assistant to try again, or remove the existing YAML configuration and set the integration up via the UI.",

295
tests/components/nederlandse_spoorwegen/snapshots/test_binary_sensor.ambr Normal file
View File

@@ -0,0 +1,295 @@
# serializer version: 1
# name: test_binary_sensor[binary_sensor.to_home_arrival_delayed-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.to_home_arrival_delayed',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Arrival delayed',
'platform': 'nederlandse_spoorwegen',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'is_arrival_delayed',
'unique_id': '01K721DZPMEN39R5DK0ATBMSY9-is_arrival_delayed',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[binary_sensor.to_home_arrival_delayed-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by NS',
'friendly_name': 'To home Arrival delayed',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.to_home_arrival_delayed',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_binary_sensor[binary_sensor.to_home_departure_delayed-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.to_home_departure_delayed',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Departure delayed',
'platform': 'nederlandse_spoorwegen',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'is_departure_delayed',
'unique_id': '01K721DZPMEN39R5DK0ATBMSY9-is_departure_delayed',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[binary_sensor.to_home_departure_delayed-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by NS',
'friendly_name': 'To home Departure delayed',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.to_home_departure_delayed',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_binary_sensor[binary_sensor.to_home_going-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.to_home_going',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Going',
'platform': 'nederlandse_spoorwegen',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'is_going',
'unique_id': '01K721DZPMEN39R5DK0ATBMSY9-is_going',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[binary_sensor.to_home_going-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by NS',
'friendly_name': 'To home Going',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.to_home_going',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_binary_sensor[binary_sensor.to_work_arrival_delayed-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.to_work_arrival_delayed',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Arrival delayed',
'platform': 'nederlandse_spoorwegen',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'is_arrival_delayed',
'unique_id': '01K721DZPMEN39R5DK0ATBMSY8-is_arrival_delayed',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[binary_sensor.to_work_arrival_delayed-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by NS',
'friendly_name': 'To work Arrival delayed',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.to_work_arrival_delayed',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_binary_sensor[binary_sensor.to_work_departure_delayed-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.to_work_departure_delayed',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Departure delayed',
'platform': 'nederlandse_spoorwegen',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'is_departure_delayed',
'unique_id': '01K721DZPMEN39R5DK0ATBMSY8-is_departure_delayed',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[binary_sensor.to_work_departure_delayed-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by NS',
'friendly_name': 'To work Departure delayed',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.to_work_departure_delayed',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_binary_sensor[binary_sensor.to_work_going-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.to_work_going',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Going',
'platform': 'nederlandse_spoorwegen',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'is_going',
'unique_id': '01K721DZPMEN39R5DK0ATBMSY8-is_going',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[binary_sensor.to_work_going-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by NS',
'friendly_name': 'To work Going',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.to_work_going',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---

75
tests/components/nederlandse_spoorwegen/test_binary_sensor.py Normal file
View File

@@ -0,0 +1,75 @@
"""Test the Nederlandse Spoorwegen binary sensor."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
from requests.exceptions import ConnectionError as RequestsConnectionError
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant
import homeassistant.helpers.entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
@pytest.fixture(autouse=True)
def mock_binary_sensor_platform() -> Generator:
"""Override PLATFORMS for NS integration."""
with patch(
"homeassistant.components.nederlandse_spoorwegen.PLATFORMS",
[Platform.BINARY_SENSOR],
) as mock_platform:
yield mock_platform
@pytest.mark.freeze_time("2025-09-15 14:30:00+00:00")
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_binary_sensor(
hass: HomeAssistant,
mock_nsapi: AsyncMock,
mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
) -> None:
"""Test sensor initialization."""
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.freeze_time("2025-09-15 14:30:00+00:00")
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_no_upcoming_trips(
hass: HomeAssistant,
mock_nsapi: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test sensor initialization."""
mock_nsapi.get_trips.return_value = []
await setup_integration(hass, mock_config_entry)
assert (
hass.states.get("binary_sensor.to_work_departure_delayed").state
== STATE_UNKNOWN
)
async def test_sensor_with_api_connection_error(
hass: HomeAssistant,
mock_nsapi: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test sensor behavior when API connection fails."""
# Make API calls fail from the start
mock_nsapi.get_trips.side_effect = RequestsConnectionError("Connection failed")
await setup_integration(hass, mock_config_entry)
await hass.async_block_till_done()
# Sensors should not be created at all if initial API call fails
sensor_states = hass.states.async_all("binary_sensor")
assert len(sensor_states) == 0

15
tests/components/nederlandse_spoorwegen/test_sensor.py
View File

@@ -1,6 +1,7 @@
"""Test the Nederlandse Spoorwegen sensor."""
from unittest.mock import AsyncMock
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
from requests.exceptions import ConnectionError as RequestsConnectionError
@@ -18,7 +19,7 @@ from homeassistant.components.nederlandse_spoorwegen.const import (
)
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.config_entries import ConfigSubentryDataWithId
from homeassistant.const import CONF_API_KEY, CONF_NAME, CONF_PLATFORM
from homeassistant.const import CONF_API_KEY, CONF_NAME, CONF_PLATFORM, Platform
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
import homeassistant.helpers.entity_registry as er
import homeassistant.helpers.issue_registry as ir
@@ -30,6 +31,16 @@ from .const import API_KEY
from tests.common import MockConfigEntry, snapshot_platform
@pytest.fixture(autouse=True)
def mock_sensor_platform() -> Generator:
"""Override PLATFORMS for NS integration."""
with patch(
"homeassistant.components.nederlandse_spoorwegen.PLATFORMS",
[Platform.SENSOR],
) as mock_platform:
yield mock_platform
async def test_config_import(
hass: HomeAssistant,
mock_nsapi,