Add squeezebox service sensors (#125349)

* Add server sensors

* Fix Platforms order

* Fix spelling

* Fix translations

* Add sensor test

* Case changes

* refactor to use native_value attr override

* Fix typing

* Fix cast to type

* add cast

* use update platform for LMS versions

* Fix translation

* remove update entity

* remove possible update entites

* Fix and clarify

* update to icon trans remove update plaform entitiy supporting items

* add UOM to sensors

* correct criptic prettier fail

* reword other players

* Apply suggestions from code review

---------

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Phill (pssc) 2024-09-10 16:17:26 +01:00 committed by GitHub
parent 300445948e
commit a16ef5b7ff
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 212 additions and 3 deletions

View file

@ -40,7 +40,11 @@ from .coordinator import LMSStatusDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.BINARY_SENSOR, Platform.MEDIA_PLAYER]
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.MEDIA_PLAYER,
Platform.SENSOR,
]
@dataclass

View file

@ -3,15 +3,18 @@
from asyncio import timeout
from datetime import timedelta
import logging
import re
from pysqueezebox import Server
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util import dt as dt_util
from .const import (
SENSOR_UPDATE_INTERVAL,
STATUS_API_TIMEOUT,
STATUS_SENSOR_LASTSCAN,
STATUS_SENSOR_NEEDSRESTART,
STATUS_SENSOR_RESCAN,
)
@ -32,6 +35,7 @@ class LMSStatusDataUpdateCoordinator(DataUpdateCoordinator):
always_update=False,
)
self.lms = lms
self.newversion_regex = re.compile("<.*$")
async def _async_update_data(self) -> dict:
"""Fetch data fromn LMS status call.
@ -50,10 +54,19 @@ class LMSStatusDataUpdateCoordinator(DataUpdateCoordinator):
def _prepare_status_data(self, data: dict) -> dict:
"""Sensors that need the data changing for HA presentation."""
# Binary sensors
# rescan bool are we rescanning alter poll not present if false
data[STATUS_SENSOR_RESCAN] = STATUS_SENSOR_RESCAN in data
# needsrestart bool pending lms plugin updates not present if false
data[STATUS_SENSOR_NEEDSRESTART] = STATUS_SENSOR_NEEDSRESTART in data
# Sensors that need special handling
# 'lastscan': '1718431678', epoc -> ISO 8601 not always present
data[STATUS_SENSOR_LASTSCAN] = (
dt_util.utc_from_timestamp(int(data[STATUS_SENSOR_LASTSCAN]))
if STATUS_SENSOR_LASTSCAN in data
else None
)
_LOGGER.debug("Processed serverstatus %s=%s", self.lms.name, data)
return data

View file

@ -21,7 +21,7 @@ class LMSStatusEntity(CoordinatorEntity[LMSStatusDataUpdateCoordinator]):
"""Initialize status sensor entity."""
super().__init__(coordinator)
self.entity_description = description
self._attr_translation_key = description.key
self._attr_translation_key = description.key.replace(" ", "_")
self._attr_unique_id = (
f"{coordinator.data[STATUS_QUERY_UUID]}_{description.key}"
)

View file

@ -1,4 +1,26 @@
{
"entity": {
"sensor": {
"info_total_albums": {
"default": "mdi:album"
},
"info_total_artists": {
"default": "mdi:account-music"
},
"info_total_genres": {
"default": "mdi:drama-masks"
},
"info_total_songs": {
"default": "mdi:file-music"
},
"player_count": {
"default": "mdi:folder-play"
},
"other_player_count": {
"default": "mdi:folder-play-outline"
}
}
},
"services": {
"call_method": {
"service": "mdi:console"

View file

@ -0,0 +1,98 @@
"""Platform for sensor integration for squeezebox."""
from __future__ import annotations
import logging
from typing import cast
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import UnitOfTime
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from . import SqueezeboxConfigEntry
from .const import (
STATUS_SENSOR_INFO_TOTAL_ALBUMS,
STATUS_SENSOR_INFO_TOTAL_ARTISTS,
STATUS_SENSOR_INFO_TOTAL_DURATION,
STATUS_SENSOR_INFO_TOTAL_GENRES,
STATUS_SENSOR_INFO_TOTAL_SONGS,
STATUS_SENSOR_LASTSCAN,
STATUS_SENSOR_OTHER_PLAYER_COUNT,
STATUS_SENSOR_PLAYER_COUNT,
)
from .entity import LMSStatusEntity
SENSORS: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key=STATUS_SENSOR_INFO_TOTAL_ALBUMS,
state_class=SensorStateClass.TOTAL,
native_unit_of_measurement="albums",
),
SensorEntityDescription(
key=STATUS_SENSOR_INFO_TOTAL_ARTISTS,
state_class=SensorStateClass.TOTAL,
native_unit_of_measurement="artists",
),
SensorEntityDescription(
key=STATUS_SENSOR_INFO_TOTAL_DURATION,
state_class=SensorStateClass.TOTAL,
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement=UnitOfTime.SECONDS,
),
SensorEntityDescription(
key=STATUS_SENSOR_INFO_TOTAL_GENRES,
state_class=SensorStateClass.TOTAL,
native_unit_of_measurement="genres",
),
SensorEntityDescription(
key=STATUS_SENSOR_INFO_TOTAL_SONGS,
state_class=SensorStateClass.TOTAL,
native_unit_of_measurement="songs",
),
SensorEntityDescription(
key=STATUS_SENSOR_LASTSCAN,
device_class=SensorDeviceClass.TIMESTAMP,
),
SensorEntityDescription(
key=STATUS_SENSOR_PLAYER_COUNT,
state_class=SensorStateClass.TOTAL,
native_unit_of_measurement="players",
),
SensorEntityDescription(
key=STATUS_SENSOR_OTHER_PLAYER_COUNT,
state_class=SensorStateClass.TOTAL,
entity_registry_visible_default=False,
native_unit_of_measurement="players",
),
)
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
entry: SqueezeboxConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Platform setup using common elements."""
async_add_entities(
ServerStatusSensor(entry.runtime_data.coordinator, description)
for description in SENSORS
)
class ServerStatusSensor(LMSStatusEntity, SensorEntity):
"""LMS Status based sensor from LMS via cooridnatior."""
@property
def native_value(self) -> StateType:
"""LMS Status directly from coordinator data."""
return cast(StateType, self.coordinator.data[self.entity_description.key])

View file

@ -84,6 +84,32 @@
"needsrestart": {
"name": "Needs restart"
}
},
"sensor": {
"lastscan": {
"name": "Last scan"
},
"info_total_albums": {
"name": "Total albums"
},
"info_total_artists": {
"name": "Total artists"
},
"info_total_duration": {
"name": "Total duration"
},
"info_total_genres": {
"name": "Total genres"
},
"info_total_songs": {
"name": "Total songs"
},
"player_count": {
"name": "Player count"
},
"other_player_count": {
"name": "Player count off service"
}
}
}
}

View file

@ -6,6 +6,14 @@ from homeassistant.components.squeezebox.const import (
STATUS_QUERY_MAC,
STATUS_QUERY_UUID,
STATUS_QUERY_VERSION,
STATUS_SENSOR_INFO_TOTAL_ALBUMS,
STATUS_SENSOR_INFO_TOTAL_ARTISTS,
STATUS_SENSOR_INFO_TOTAL_DURATION,
STATUS_SENSOR_INFO_TOTAL_GENRES,
STATUS_SENSOR_INFO_TOTAL_SONGS,
STATUS_SENSOR_LASTSCAN,
STATUS_SENSOR_OTHER_PLAYER_COUNT,
STATUS_SENSOR_PLAYER_COUNT,
STATUS_SENSOR_RESCAN,
)
from homeassistant.const import CONF_HOST, CONF_PORT
@ -25,7 +33,15 @@ FAKE_QUERY_RESPONSE = {
STATUS_QUERY_MAC: FAKE_MAC,
STATUS_QUERY_VERSION: FAKE_VERSION,
STATUS_SENSOR_RESCAN: 1,
STATUS_SENSOR_LASTSCAN: 0,
STATUS_QUERY_LIBRARYNAME: "FakeLib",
STATUS_SENSOR_INFO_TOTAL_ALBUMS: 4,
STATUS_SENSOR_INFO_TOTAL_ARTISTS: 2,
STATUS_SENSOR_INFO_TOTAL_DURATION: 500,
STATUS_SENSOR_INFO_TOTAL_GENRES: 1,
STATUS_SENSOR_INFO_TOTAL_SONGS: 42,
STATUS_SENSOR_PLAYER_COUNT: 10,
STATUS_SENSOR_OTHER_PLAYER_COUNT: 0,
"players_loop": [
{
"isplaying": 0,

View file

@ -1,5 +1,6 @@
"""Test squeezebox binary sensors."""
import copy
from unittest.mock import patch
from homeassistant.const import Platform
@ -23,7 +24,7 @@ async def test_binary_sensor(
),
patch(
"homeassistant.components.squeezebox.Server.async_query",
return_value=FAKE_QUERY_RESPONSE,
return_value=copy.deepcopy(FAKE_QUERY_RESPONSE),
),
):
await setup_mocked_integration(hass)

View file

@ -0,0 +1,29 @@
"""Test squeezebox sensors."""
from unittest.mock import patch
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from . import FAKE_QUERY_RESPONSE, setup_mocked_integration
async def test_sensor(hass: HomeAssistant) -> None:
"""Test binary sensor states and attributes."""
# Setup component
with (
patch(
"homeassistant.components.squeezebox.PLATFORMS",
[Platform.SENSOR],
),
patch(
"homeassistant.components.squeezebox.Server.async_query",
return_value=FAKE_QUERY_RESPONSE,
),
):
await setup_mocked_integration(hass)
state = hass.states.get("sensor.fakelib_player_count")
assert state is not None
assert state.state == "10"