Refactor sensor platform of Steam integration (#174415)

This commit is contained in:
Manu
2026-06-22 22:59:42 +02:00
committed by GitHub
parent 982fe7a370
commit 249c3bb5dd
8 changed files with 257 additions and 30 deletions
@@ -10,7 +10,7 @@ from .coordinator import SteamDataUpdateCoordinator
class SteamEntity(CoordinatorEntity[SteamDataUpdateCoordinator]):
"""Representation of a Steam entity."""
_attr_attribution = "Data provided by Steam"
_attr_has_entity_name = True
def __init__(self, coordinator: SteamDataUpdateCoordinator) -> None:
"""Initialize a Steam entity."""
@@ -0,0 +1,9 @@
{
"entity": {
"sensor": {
"account": {
"default": "mdi:steam"
}
}
}
}
+74 -26
View File
@@ -1,14 +1,16 @@
"""Sensor for Steam account status."""
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
from time import localtime, mktime
from typing import cast, override
from enum import StrEnum
from typing import Any, cast, override
from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.util.dt import utc_from_timestamp
from homeassistant.util import dt as dt_util
from .const import (
CONF_ACCOUNTS,
@@ -24,47 +26,87 @@ from .entity import SteamEntity
PARALLEL_UPDATES = 1
class SteamSensor(StrEnum):
"""Steam sensors."""
ACCOUNT = "account"
@dataclass(kw_only=True, frozen=True)
class SteamSensorEntityDescription(SensorEntityDescription):
"""Steam sensor description."""
value_fn: Callable[[dict[str, Any]], StateType]
name_fn: Callable[[dict[str, Any]], str]
entity_picture_fn: Callable[[dict[str, Any]], str] | None = None
SENSOR_DESCRIPTIONS: tuple[SteamSensorEntityDescription, ...] = (
SteamSensorEntityDescription(
key=SteamSensor.ACCOUNT,
translation_key=SteamSensor.ACCOUNT,
value_fn=lambda x: STEAM_STATUSES[x["personastate"]],
name_fn=lambda x: x["personaname"],
entity_picture_fn=lambda x: x["avatarfull"],
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: SteamConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Steam platform."""
coordinator = entry.runtime_data
async_add_entities(
SteamSensor(entry.runtime_data, account)
for account in entry.options[CONF_ACCOUNTS]
SteamSensorEntity(coordinator, steamid, description)
for steamid in entry.options[CONF_ACCOUNTS]
for description in SENSOR_DESCRIPTIONS
if steamid in coordinator.data
)
class SteamSensor(SteamEntity, SensorEntity):
"""A class for the Steam account."""
class SteamSensorEntity(SteamEntity, SensorEntity):
"""Representation of a Steam sensor entity."""
def __init__(self, coordinator: SteamDataUpdateCoordinator, account: str) -> None:
entity_description: SteamSensorEntityDescription
def __init__(
self,
coordinator: SteamDataUpdateCoordinator,
steamid: str,
description: SteamSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self.entity_description = SensorEntityDescription(
key=account,
name=f"steam_{account}",
icon="mdi:steam",
)
self._attr_unique_id = f"sensor.steam_{account}"
self._steamid = steamid
self.entity_description = description
self._attr_unique_id = f"sensor.steam_{steamid}"
self._attr_name = self.entity_description.name_fn(coordinator.data[steamid])
@property
@override
def native_value(self) -> StateType:
"""Return the state of the sensor."""
if self.entity_description.key in self.coordinator.data:
player = self.coordinator.data[self.entity_description.key]
return STEAM_STATUSES[cast(int, player["personastate"])]
return None
return self.entity_description.value_fn(self.coordinator.data[self._steamid])
@property
@override
def extra_state_attributes(self) -> dict[str, str | int | datetime]:
def entity_picture(self) -> str | None:
"""Return the entity picture to use in the frontend, if any."""
return (
fn(self.coordinator.data[self._steamid])
if (fn := self.entity_description.entity_picture_fn) is not None
else super().entity_picture
)
@property
@override
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the sensor."""
if self.entity_description.key not in self.coordinator.data:
return {}
player = self.coordinator.data[self.entity_description.key]
player = self.coordinator.data[self._steamid]
attrs: dict[str, str | int | datetime] = {}
if game := player.get("gameextrainfo"):
@@ -76,11 +118,11 @@ class SteamSensor(SteamEntity, SensorEntity):
attrs["game_image_main"] = f"{game_url}{STEAM_MAIN_IMAGE_FILE}"
if info := self._get_game_icon(player):
attrs["game_icon"] = f"{STEAM_ICON_URL}{game_id}/{info}.jpg"
self._attr_name = str(player["personaname"]) or None
self._attr_entity_picture = str(player["avatarmedium"]) or None
if last_online := cast(int | None, player.get("lastlogoff")):
attrs["last_online"] = utc_from_timestamp(mktime(localtime(last_online)))
if level := self.coordinator.data[self.entity_description.key]["level"]:
attrs["last_online"] = dt_util.as_local(
dt_util.utc_from_timestamp(last_online)
)
if level := self.coordinator.data[self._steamid]["level"]:
attrs["level"] = level
return attrs
@@ -91,3 +133,9 @@ class SteamSensor(SteamEntity, SensorEntity):
# Reset game icons to have coordinator get id for new game
self.coordinator.game_icons = {}
return None
@property
@override
def available(self) -> bool:
"""Return True if entity is available."""
return super().available and self._steamid in self.coordinator.data
@@ -28,6 +28,30 @@
}
}
},
"entity": {
"sensor": {
"account": {
"state": {
"away": "Away",
"busy": "Busy",
"looking_to_play": "Looking to play",
"looking_to_trade": "Looking to trade",
"offline": "Offline",
"online": "Online",
"snooze": "Snooze"
},
"state_attributes": {
"game": { "name": "Game" },
"game_icon": { "name": "Game icon" },
"game_id": { "name": "Game ID" },
"game_image_header": { "name": "Game header image" },
"game_image_main": { "name": "Game image" },
"last_online": { "name": "Last online" },
"level": { "name": "Level" }
}
}
}
},
"options": {
"error": {
"unauthorized": "Friends list restricted: Please refer to the documentation on how to see all other friends"
+28 -3
View File
@@ -87,15 +87,34 @@ class MockedInterface(dict):
"player": [
{
"steamid": ACCOUNT_1,
"communityvisibilitystate": 1,
"profilestate": 1,
"personaname": ACCOUNT_NAME_1,
"profileurl": "https://steamcommunity.com/profiles/123456789/",
"avatar": "https://avatars.steamstatic.com/fef49e7fa7e1997310d705b2a6158ff8dc1cdfeb.jpg",
"avatarmedium": "https://avatars.steamstatic.com/fef49e7fa7e1997310d705b2a6158ff8dc1cdfeb_medium.jpg",
"avatarfull": "https://avatars.steamstatic.com/fef49e7fa7e1997310d705b2a6158ff8dc1cdfeb_full.jpg",
"avatarhash": "fef49e7fa7e1997310d705b2a6158ff8dc1cdfeb",
"lastlogoff": 1775409487,
"personastate": 1,
"avatarmedium": "",
"realname": "John Dough",
"personastateflags": 0,
"gameextrainfo": "The Witcher: Enhanced Edition",
"gameid": "20900",
},
{
"steamid": ACCOUNT_2,
"communityvisibilitystate": 1,
"profilestate": 1,
"personaname": ACCOUNT_NAME_2,
"profileurl": "https://steamcommunity.com/profiles/987654321/",
"avatar": "https://avatars.steamstatic.com/fef49e7fa7e1997310d705b2a6158ff8dc1cdfeb.jpg",
"avatarmedium": "https://avatars.steamstatic.com/fef49e7fa7e1997310d705b2a6158ff8dc1cdfeb_medium.jpg",
"avatarfull": "https://avatars.steamstatic.com/fef49e7fa7e1997310d705b2a6158ff8dc1cdfeb_full.jpg",
"avatarhash": "fef49e7fa7e1997310d705b2a6158ff8dc1cdfeb",
"lastlogoff": 1775409487,
"personastate": 2,
"avatarmedium": "",
"personastateflags": 0,
},
]
}
@@ -106,7 +125,13 @@ class MockedInterface(dict):
"""Get owned games."""
return {
"response": {"game_count": 1},
"games": [{"appid": 1, "img_icon_url": "1234567890"}],
"games": [
{"appid": 1, "img_icon_url": "1234567890"},
{
"appid": 20900,
"img_icon_url": "746d1cd48fb2e57d579b05b6e9eccba95859e549",
},
],
}
def GetSteamLevel(self, steamid: str) -> dict:
+20
View File
@@ -0,0 +1,20 @@
"""Common fixtures for Steam integration."""
import pytest
from homeassistant.components.steam_online.const import DOMAIN
from . import ACCOUNT_1, CONF_DATA, CONF_OPTIONS
from tests.common import MockConfigEntry
@pytest.fixture(name="config_entry")
def mock_config_entry() -> MockConfigEntry:
"""Mock Steam configuration entry."""
return MockConfigEntry(
domain=DOMAIN,
data=CONF_DATA,
options=CONF_OPTIONS,
unique_id=ACCOUNT_1,
)
@@ -0,0 +1,58 @@
# serializer version: 1
# name: test_sensors[sensor.steam_testaccount1-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.steam_testaccount1',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'testaccount1',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'testaccount1',
'platform': 'steam_online',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': <SteamSensor.ACCOUNT: 'account'>,
'unique_id': 'sensor.steam_12345678901234567',
'unit_of_measurement': None,
})
# ---
# name: test_sensors[sensor.steam_testaccount1-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'entity_picture': 'https://avatars.steamstatic.com/fef49e7fa7e1997310d705b2a6158ff8dc1cdfeb_full.jpg',
'friendly_name': 'Steam testaccount1',
'game': 'The Witcher: Enhanced Edition',
'game_id': '20900',
'game_image_header': 'https://steamcdn-a.akamaihd.net/steam/apps/20900/header.jpg',
'game_image_main': 'https://steamcdn-a.akamaihd.net/steam/apps/20900/capsule_616x353.jpg',
'last_online': datetime.datetime(2026, 4, 5, 10, 18, 7, tzinfo=zoneinfo.ZoneInfo(key='US/Pacific')),
'level': 10,
}),
'context': <ANY>,
'entity_id': 'sensor.steam_testaccount1',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'online',
})
# ---
@@ -0,0 +1,43 @@
"""Tests for Steam sensor platform."""
from collections.abc import Generator
from unittest.mock import patch
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import patch_interface
from tests.common import MockConfigEntry, snapshot_platform
@pytest.fixture(autouse=True)
def sensor_only() -> Generator[None]:
"""Enable only the sensor platform."""
with patch(
"homeassistant.components.steam_online.PLATFORMS",
[Platform.SENSOR],
):
yield
async def test_sensors(
hass: HomeAssistant,
config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
) -> None:
"""Test setup of the Steam sensor platform."""
with patch_interface():
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
await snapshot_platform(hass, entity_registry, snapshot, config_entry.entry_id)