From 25cd0a2b8ba6b84d478b1663154c966798daef19 Mon Sep 17 00:00:00 2001 From: Tomasz Dylewski Date: Fri, 25 Sep 2026 15:47:06 +0200 Subject: [PATCH] Add voltage sensor support to PAJ GPS (#182992) Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../components/paj_gps/coordinator.py | 41 +++++- homeassistant/components/paj_gps/sensor.py | 58 ++++++-- tests/components/paj_gps/conftest.py | 2 + .../paj_gps/snapshots/test_sensor.ambr | 119 +++++++++++++++ tests/components/paj_gps/test_sensor.py | 136 +++++++++++++++++- 5 files changed, 346 insertions(+), 10 deletions(-) diff --git a/homeassistant/components/paj_gps/coordinator.py b/homeassistant/components/paj_gps/coordinator.py index 68adbca8e461..60cc965bd36d 100644 --- a/homeassistant/components/paj_gps/coordinator.py +++ b/homeassistant/components/paj_gps/coordinator.py @@ -1,5 +1,6 @@ """DataUpdateCoordinator for the PAJ GPS integration.""" +import asyncio from dataclasses import dataclass from datetime import timedelta import logging @@ -7,6 +8,7 @@ from typing import override from pajgps_api import PajGpsApi from pajgps_api.models.device import Device +from pajgps_api.models.sensordata import SensorData from pajgps_api.models.trackpoint import TrackPoint from pajgps_api.pajgps_api_error import ( AuthenticationError, @@ -34,6 +36,7 @@ class PajGpsData: devices: dict[int, Device] positions: dict[int, TrackPoint] + sensor_data: dict[int, SensorData] class PajGpsCoordinator(DataUpdateCoordinator[PajGpsData]): @@ -57,6 +60,7 @@ class PajGpsCoordinator(DataUpdateCoordinator[PajGpsData]): self._email: str = config_entry.data[CONF_EMAIL] self._user_id: int | None = None + self._voltage_data_failures: set[int] = set() self.api = PajGpsApi( email=self._email, password=config_entry.data[CONF_PASSWORD], @@ -107,4 +111,39 @@ class PajGpsCoordinator(DataUpdateCoordinator[PajGpsData]): tp.iddevice: tp for tp in track_points if tp.iddevice is not None } - return PajGpsData(devices=devices, positions=positions) + sensor_data: dict[int, SensorData] = {} + voltage_device_ids = [ + device_id + for device_id, device in devices.items() + if device.has_voltage_sensor + ] + + if voltage_device_ids: + results = await asyncio.gather( + *( + self.api.get_last_sensor_data(device_id) + for device_id in voltage_device_ids + ), + return_exceptions=True, + ) + + for device_id, result in zip(voltage_device_ids, results, strict=False): + if isinstance(result, PajGpsApiError): + if device_id not in self._voltage_data_failures: + _LOGGER.info( + "Failed to fetch voltage sensor data for device %s: %s", + device_id, + result, + ) + self._voltage_data_failures.add(device_id) + continue + if isinstance(result, BaseException): + raise result + if device_id in self._voltage_data_failures: + _LOGGER.info( + "Voltage sensor data recovered for device %s", device_id + ) + self._voltage_data_failures.remove(device_id) + sensor_data[device_id] = result + + return PajGpsData(devices=devices, positions=positions, sensor_data=sensor_data) diff --git a/homeassistant/components/paj_gps/sensor.py b/homeassistant/components/paj_gps/sensor.py index a578bbc064d6..6ef7f574f43b 100644 --- a/homeassistant/components/paj_gps/sensor.py +++ b/homeassistant/components/paj_gps/sensor.py @@ -4,6 +4,7 @@ from collections.abc import Callable from dataclasses import dataclass, field from typing import override +from pajgps_api.models.sensordata import SensorData from pajgps_api.models.trackpoint import TrackPoint from homeassistant.components.sensor import ( @@ -12,7 +13,12 @@ from homeassistant.components.sensor import ( SensorEntityDescription, SensorStateClass, ) -from homeassistant.const import PERCENTAGE, EntityCategory, UnitOfSpeed +from homeassistant.const import ( + PERCENTAGE, + EntityCategory, + UnitOfElectricPotential, + UnitOfSpeed, +) from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback @@ -27,7 +33,8 @@ PARALLEL_UPDATES = 0 class PajGpsSensorEntityDescription(SensorEntityDescription): """Describes a PAJ GPS sensor entity.""" - value_fn: Callable[[TrackPoint], int | None] + trackpoint_value_fn: Callable[[TrackPoint], int | None] | None = None + sensor_data_value_fn: Callable[[SensorData], int | None] | None = None supported_fn: Callable[[Device], bool] = field(default=lambda _: True) @@ -38,7 +45,7 @@ SENSOR_DESCRIPTIONS: tuple[PajGpsSensorEntityDescription, ...] = ( native_unit_of_measurement=UnitOfSpeed.KILOMETERS_PER_HOUR, state_class=SensorStateClass.MEASUREMENT, suggested_display_precision=0, - value_fn=lambda tp: tp.speed, + trackpoint_value_fn=lambda tp: tp.speed, ), PajGpsSensorEntityDescription( key="battery", @@ -47,9 +54,20 @@ SENSOR_DESCRIPTIONS: tuple[PajGpsSensorEntityDescription, ...] = ( state_class=SensorStateClass.MEASUREMENT, entity_category=EntityCategory.DIAGNOSTIC, suggested_display_precision=0, - value_fn=lambda tp: tp.battery_level, + trackpoint_value_fn=lambda tp: tp.battery_level, supported_fn=lambda device: device.has_battery, ), + PajGpsSensorEntityDescription( + key="voltage", + device_class=SensorDeviceClass.VOLTAGE, + native_unit_of_measurement=UnitOfElectricPotential.MILLIVOLT, + suggested_unit_of_measurement=UnitOfElectricPotential.VOLT, + state_class=SensorStateClass.MEASUREMENT, + entity_category=EntityCategory.DIAGNOSTIC, + suggested_display_precision=2, + sensor_data_value_fn=lambda data: data.volt, + supported_fn=lambda device: device.has_voltage_sensor, + ), ) @@ -102,8 +120,32 @@ class PajGpsSensor(PajGpsEntity, SensorEntity): @property @override def native_value(self) -> int | None: - """Return the sensor value from the latest trackpoint.""" - tp = self.coordinator.data.positions.get(self._device_id) - if tp is None: + """Return the sensor value from the latest coordinator snapshot.""" + if ( + trackpoint_value_fn := self.entity_description.trackpoint_value_fn + ) is not None: + tp = self.coordinator.data.positions.get(self._device_id) + if tp is None: + return None + return trackpoint_value_fn(tp) + + if ( + sensor_data_value_fn := self.entity_description.sensor_data_value_fn + ) is None: return None - return self.entity_description.value_fn(tp) + + sensor_data = self.coordinator.data.sensor_data.get(self._device_id) + if sensor_data is None: + return None + return sensor_data_value_fn(sensor_data) + + @property + @override + def available(self) -> bool: + """Return if the sensor is available.""" + if self.entity_description.key != "voltage": + return super().available + + return ( + super().available and self._device_id in self.coordinator.data.sensor_data + ) diff --git a/tests/components/paj_gps/conftest.py b/tests/components/paj_gps/conftest.py index 84bcf85b1ee4..fa9aef387dc8 100644 --- a/tests/components/paj_gps/conftest.py +++ b/tests/components/paj_gps/conftest.py @@ -7,6 +7,7 @@ from unittest.mock import AsyncMock, patch from pajgps_api.models.auth import AuthResponse from pajgps_api.models.device import Device +from pajgps_api.models.sensordata import SensorData from pajgps_api.models.trackpoint import TrackPoint import pytest @@ -63,4 +64,5 @@ def mock_paj_gps_api() -> Generator[AsyncMock]: api.get_all_last_positions.return_value = [ TrackPoint(**load_json_object_fixture("trackpoint.json", DOMAIN)) ] + api.get_last_sensor_data.return_value = SensorData(did=1, volt=12400) yield api diff --git a/tests/components/paj_gps/snapshots/test_sensor.ambr b/tests/components/paj_gps/snapshots/test_sensor.ambr index a1926b9e1873..a23750432dc5 100644 --- a/tests/components/paj_gps/snapshots/test_sensor.ambr +++ b/tests/components/paj_gps/snapshots/test_sensor.ambr @@ -115,3 +115,122 @@ 'state': '50', }) # --- +# name: test_voltage_sensor_created_when_has_voltage_support[sensor.device_1_speed-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.device_1_speed', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Speed', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 0, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Speed', + 'platform': 'paj_gps', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': None, + 'unique_id': '42_1_speed', + 'unit_of_measurement': , + }) +# --- +# name: test_voltage_sensor_created_when_has_voltage_support[sensor.device_1_speed-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'speed', + : 'Device 1 Speed', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.device_1_speed', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '50', + }) +# --- +# name: test_voltage_sensor_created_when_has_voltage_support[sensor.device_1_voltage-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': , + 'entity_id': 'sensor.device_1_voltage', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Voltage', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 2, + }), + 'sensor.private': dict({ + 'suggested_unit_of_measurement': , + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Voltage', + 'platform': 'paj_gps', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': None, + 'unique_id': '42_1_voltage', + 'unit_of_measurement': , + }) +# --- +# name: test_voltage_sensor_created_when_has_voltage_support[sensor.device_1_voltage-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'voltage', + : 'Device 1 Voltage', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.device_1_voltage', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '12.4', + }) +# --- diff --git a/tests/components/paj_gps/test_sensor.py b/tests/components/paj_gps/test_sensor.py index 031dcfdeae07..aec98fa5ce8a 100644 --- a/tests/components/paj_gps/test_sensor.py +++ b/tests/components/paj_gps/test_sensor.py @@ -6,12 +6,14 @@ from collections.abc import Generator from unittest.mock import AsyncMock, patch from pajgps_api.models.device import Device +from pajgps_api.models.sensordata import SensorData from pajgps_api.models.trackpoint import TrackPoint +from pajgps_api.pajgps_api_error import PajGpsApiError import pytest from syrupy.assertion import SnapshotAssertion from homeassistant.components.paj_gps.const import DOMAIN -from homeassistant.const import STATE_UNKNOWN, Platform +from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN, Platform from homeassistant.core import HomeAssistant from homeassistant.helpers import entity_registry as er @@ -61,6 +63,20 @@ def mock_paj_gps_api_with_battery(mock_paj_gps_api: AsyncMock) -> AsyncMock: return mock_paj_gps_api +@pytest.fixture +def mock_paj_gps_api_with_voltage(mock_paj_gps_api: AsyncMock) -> AsyncMock: + """Override get_devices to return a device with voltage sensor support.""" + mock_paj_gps_api.get_devices.return_value = [ + Device( + **{ + **load_json_object_fixture("device.json", DOMAIN), + "device_models": [{"alarm_volt": 1}], + } + ) + ] + return mock_paj_gps_api + + @pytest.mark.usefixtures("mock_paj_gps_api_with_battery") async def test_battery_sensor_created_when_has_battery( hass: HomeAssistant, @@ -100,3 +116,121 @@ async def test_battery_none_when_missing( state = hass.states.get("sensor.device_1_battery") assert state is not None assert state.state == STATE_UNKNOWN + + +@pytest.mark.usefixtures("mock_paj_gps_api_with_voltage") +async def test_voltage_sensor_created_when_has_voltage_support( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + entity_registry: er.EntityRegistry, + snapshot: SnapshotAssertion, +) -> None: + """Test that a voltage sensor is created when alarm_volt is supported.""" + await setup_integration(hass, mock_config_entry) + + await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id) + + +@pytest.mark.usefixtures("mock_paj_gps_api") +async def test_voltage_sensor_not_created_when_no_voltage_support( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, +) -> None: + """Test that no voltage sensor is created when alarm_volt is not supported.""" + await setup_integration(hass, mock_config_entry) + + assert hass.states.get("sensor.device_1_voltage") is None + + +async def test_voltage_none_when_missing( + hass: HomeAssistant, + mock_paj_gps_api_with_voltage: AsyncMock, + mock_config_entry: MockConfigEntry, +) -> None: + """Test that voltage state is unknown when SensorData has no volt value.""" + mock_paj_gps_api_with_voltage.get_last_sensor_data.return_value = SensorData( + did=1, + volt=None, + ) + + await setup_integration(hass, mock_config_entry) + + state = hass.states.get("sensor.device_1_voltage") + assert state is not None + assert state.state == STATE_UNKNOWN + + +async def test_voltage_partial_degrade_when_one_sensor_data_call_fails( + hass: HomeAssistant, + caplog: pytest.LogCaptureFixture, + mock_paj_gps_api: AsyncMock, + mock_config_entry: MockConfigEntry, +) -> None: + """Test one failed sensor-data request does not break all voltage sensors.""" + mock_paj_gps_api.get_devices.return_value = [ + Device( + id=1, + name="Device 1", + imei="IMEI1", + modellid=100, + device_models=[{"alarm_volt": 1}], + ), + Device( + id=2, + name="Device 2", + imei="IMEI2", + modellid=100, + device_models=[{"alarm_volt": 1}], + ), + ] + mock_paj_gps_api.get_all_last_positions.return_value = [ + TrackPoint(iddevice=1, speed=50), + TrackPoint(iddevice=2, speed=40), + ] + + device_sensor_data = { + 1: AsyncMock(return_value=SensorData(did=1, volt=12400)), + 2: AsyncMock(side_effect=PajGpsApiError("boom")), + } + + async def _get_last_sensor_data(device_id: int) -> SensorData: + return await device_sensor_data[device_id]() + + mock_paj_gps_api.get_last_sensor_data.side_effect = _get_last_sensor_data + + await setup_integration(hass, mock_config_entry) + + state_1 = hass.states.get("sensor.device_1_voltage") + assert state_1 is not None + assert state_1.state == "12.4" + + state_2 = hass.states.get("sensor.device_2_voltage") + assert state_2 is not None + assert state_2.state == STATE_UNAVAILABLE + assert ( + caplog.messages.count("Failed to fetch voltage sensor data for device 2: boom") + == 1 + ) + + await mock_config_entry.runtime_data.async_refresh() + await hass.async_block_till_done() + + assert ( + caplog.messages.count("Failed to fetch voltage sensor data for device 2: boom") + == 1 + ) + + device_sensor_data[2].side_effect = None + device_sensor_data[2].return_value = SensorData(did=2, volt=12500) + await mock_config_entry.runtime_data.async_refresh() + await hass.async_block_till_done() + + assert caplog.messages.count("Voltage sensor data recovered for device 2") == 1 + state_2 = hass.states.get("sensor.device_2_voltage") + assert state_2 is not None + assert state_2.state == "12.5" + + await mock_config_entry.runtime_data.async_refresh() + await hass.async_block_till_done() + + assert caplog.messages.count("Voltage sensor data recovered for device 2") == 1