feat(foreground): foreground process sensors + translation key migration

Adds Home Assistant entities for the foreground-process feature shipped
in the media server, plus migrates existing display entities to use HA
translation keys (strings.json / translations/*) so per-language UI text
flows through the standard locale mechanism.

Foreground entities (all share one HA "Foreground" device linked to the
hub via via_device):
- sensor.foreground_process — process name as state + full payload
  (pid, exec path, window title, fullscreen flag, monitor, geometry,
  is_browser, browser_page_title, browser_url, error) as attributes
- sensor.window_title, sensor.pid, sensor.foreground_monitor,
  sensor.process_started (TIMESTAMP device class)
- binary_sensor.fullscreen, binary_sensor.minimized

Data flow:
- ForegroundCoordinator polls GET /api/foreground every 5s (HTTP fallback)
- media_player's WebSocket receiver forwards `foreground` /
  `foreground_update` push frames into the coordinator via
  apply_websocket_snapshot, so sensors update in near-real-time when WS
  is connected and fall back to polling otherwise

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-05-18 03:13:23 +03:00
parent ab0585278c
commit 9d277276b8
14 changed files with 463 additions and 32 deletions
@@ -25,6 +25,7 @@ from .const import (
SERVICE_PLAY_MEDIA_FILE,
)
from .display_coordinator import DisplayCoordinator
from .foreground_coordinator import ForegroundCoordinator
_LOGGER = logging.getLogger(__name__)
@@ -88,11 +89,20 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
except Exception as err: # noqa: BLE001 - first refresh wraps its own errors
_LOGGER.warning("Initial display monitor fetch failed, will retry: %s", err)
# Foreground coordinator — shared by sensor + binary_sensor platforms and
# nudged by the media-player WebSocket receiver when it gets a push.
foreground_coordinator = ForegroundCoordinator(hass, client)
try:
await foreground_coordinator.async_config_entry_first_refresh()
except Exception as err: # noqa: BLE001
_LOGGER.warning("Initial foreground fetch failed, will retry: %s", err)
# Store client in hass.data
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = {
"client": client,
"display_coordinator": display_coordinator,
"foreground_coordinator": foreground_coordinator,
}
# Register services if not already registered
@@ -31,6 +31,7 @@ from .const import (
API_BROWSER_BROWSE,
API_BROWSER_PLAY,
API_DISPLAY_MONITORS,
API_FOREGROUND,
API_DISPLAY_BRIGHTNESS,
API_DISPLAY_POWER,
API_DISPLAY_CONTRAST,
@@ -420,6 +421,15 @@ class MediaServerClient:
"POST", f"{API_DISPLAY_PICTURE_MODE}/{monitor_id}", {"code": code}
)
async def get_foreground(self) -> dict[str, Any]:
"""Get the foreground window/process snapshot.
Returns the structured payload described in the media server's
``ForegroundInfo`` dataclass: process name, window title, fullscreen
flag, owning monitor, geometry, and process start time.
"""
return await self._request("GET", API_FOREGROUND)
class MediaServerWebSocket:
"""WebSocket client for real-time media status updates."""
@@ -432,6 +442,7 @@ class MediaServerWebSocket:
on_status_update: Callable[[dict[str, Any]], None],
on_disconnect: Callable[[], None] | None = None,
on_scripts_changed: Callable[[], None] | None = None,
on_foreground_update: Callable[[dict[str, Any]], None] | None = None,
) -> None:
"""Initialize the WebSocket client.
@@ -442,6 +453,7 @@ class MediaServerWebSocket:
on_status_update: Callback when status update received
on_disconnect: Callback when connection lost
on_scripts_changed: Callback when scripts have changed
on_foreground_update: Callback when foreground process changes
"""
self._host = host
self._port = int(port)
@@ -449,6 +461,7 @@ class MediaServerWebSocket:
self._on_status_update = on_status_update
self._on_disconnect = on_disconnect
self._on_scripts_changed = on_scripts_changed
self._on_foreground_update = on_foreground_update
# The server's WS endpoint accepts an unauthenticated connection when
# api_tokens is empty (see media.py:websocket_endpoint), so we only
# append ?token=... when one was configured.
@@ -537,6 +550,9 @@ class MediaServerWebSocket:
_LOGGER.info("Scripts changed notification received")
if self._on_scripts_changed:
self._on_scripts_changed()
elif msg_type in ("foreground", "foreground_update"):
if self._on_foreground_update:
self._on_foreground_update(data.get("data", {}))
elif msg_type == "pong":
_LOGGER.debug("Received pong")
@@ -15,6 +15,8 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .display_coordinator import DisplayCoordinator
from .display_device import display_device_info
from .foreground import FOREGROUND_BINARY_SENSORS
from .foreground_coordinator import ForegroundCoordinator
_LOGGER = logging.getLogger(__name__)
@@ -24,22 +26,33 @@ async def async_setup_entry(
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up per-display binary sensor entities."""
coordinator: DisplayCoordinator = hass.data[DOMAIN][entry.entry_id][
"display_coordinator"
]
if not coordinator.data:
return
"""Set up display + foreground binary sensor entities."""
store = hass.data[DOMAIN][entry.entry_id]
display_coordinator: DisplayCoordinator = store["display_coordinator"]
foreground_coordinator: ForegroundCoordinator | None = store.get(
"foreground_coordinator"
)
entities: list[Any] = []
for monitor in coordinator.data.values():
entities.append(DisplayPrimaryBinarySensor(coordinator, entry, monitor))
entities.append(DisplayPowerControlBinarySensor(coordinator, entry, monitor))
if display_coordinator.data:
for monitor in display_coordinator.data.values():
entities.append(
DisplayPrimaryBinarySensor(display_coordinator, entry, monitor)
)
entities.append(
DisplayPowerControlBinarySensor(display_coordinator, entry, monitor)
)
if foreground_coordinator is not None:
entities.extend(
cls(foreground_coordinator, entry) for cls in FOREGROUND_BINARY_SENSORS
)
if entities:
async_add_entities(entities)
_LOGGER.info("Added %d display binary sensor entities", len(entities))
_LOGGER.info(
"Added %d binary sensor entities (display + foreground)", len(entities)
)
class _DisplayBinarySensorBase(
@@ -70,7 +83,7 @@ class _DisplayBinarySensorBase(
class DisplayPrimaryBinarySensor(_DisplayBinarySensorBase):
"""Indicates whether the display is the OS primary monitor."""
_attr_name = "Primary display"
_attr_translation_key = "primary_display"
_attr_icon = "mdi:monitor-star"
def __init__(
@@ -90,7 +103,7 @@ class DisplayPrimaryBinarySensor(_DisplayBinarySensorBase):
class DisplayPowerControlBinarySensor(_DisplayBinarySensorBase):
"""Indicates whether DDC/CI power control is available for this display."""
_attr_name = "Power control supported"
_attr_translation_key = "power_control_supported"
_attr_icon = "mdi:power-plug"
def __init__(
@@ -41,6 +41,7 @@ API_WEBSOCKET = "/api/media/ws"
API_BROWSER_FOLDERS = "/api/browser/folders"
API_BROWSER_BROWSE = "/api/browser/browse"
API_BROWSER_PLAY = "/api/browser/play"
API_FOREGROUND = "/api/foreground"
API_DISPLAY_MONITORS = "/api/display/monitors"
API_DISPLAY_BRIGHTNESS = "/api/display/brightness"
API_DISPLAY_POWER = "/api/display/power"
@@ -0,0 +1,225 @@
"""Foreground process sensor and binary-sensor entities."""
from __future__ import annotations
import logging
from datetime import datetime, timezone
from typing import Any
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .foreground_coordinator import ForegroundCoordinator
_LOGGER = logging.getLogger(__name__)
def _foreground_device_info(entry: ConfigEntry) -> DeviceInfo:
"""All foreground entities share one HA device, linked to the hub."""
return DeviceInfo(
identifiers={(DOMAIN, f"{entry.entry_id}_foreground")},
via_device=(DOMAIN, entry.entry_id),
name="Foreground",
manufacturer="Remote Media Player",
model="Foreground Process",
)
class _ForegroundEntityBase(CoordinatorEntity[ForegroundCoordinator]):
"""Boilerplate shared by every foreground entity."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: ForegroundCoordinator,
entry: ConfigEntry,
) -> None:
super().__init__(coordinator)
self._entry = entry
self._attr_device_info = _foreground_device_info(entry)
@property
def _data(self) -> dict[str, Any]:
return self.coordinator.data or {}
@property
def available(self) -> bool:
# Coordinator availability covers HTTP failures; the per-platform
# ``available`` flag in the payload reports e.g. "Wayland session".
if not super().available:
return False
return bool(self._data.get("available", True))
class ForegroundProcessSensor(_ForegroundEntityBase, SensorEntity):
"""Primary sensor: the process name plus full payload as attributes."""
_attr_icon = "mdi:application"
_attr_translation_key = "foreground_process"
def __init__(
self,
coordinator: ForegroundCoordinator,
entry: ConfigEntry,
) -> None:
super().__init__(coordinator, entry)
self._attr_unique_id = f"{entry.entry_id}_foreground_process"
@property
def native_value(self) -> str | None:
return self._data.get("process_name")
@property
def extra_state_attributes(self) -> dict[str, Any]:
d = self._data
return {
"pid": d.get("pid"),
"executable_path": d.get("executable_path"),
"window_title": d.get("window_title"),
"window_handle": d.get("window_handle"),
"is_fullscreen": d.get("is_fullscreen"),
"is_minimized": d.get("is_minimized"),
"monitor_id": d.get("monitor_id"),
"monitor_geometry": d.get("monitor_geometry"),
"window_geometry": d.get("window_geometry"),
"started_at": d.get("started_at"),
"platform": d.get("platform"),
"is_browser": d.get("is_browser"),
"browser_page_title": d.get("browser_page_title"),
"browser_url": d.get("browser_url"),
"available": d.get("available"),
"error": d.get("error"),
}
class ForegroundWindowTitleSensor(_ForegroundEntityBase, SensorEntity):
_attr_icon = "mdi:window-restore"
_attr_translation_key = "window_title"
def __init__(
self,
coordinator: ForegroundCoordinator,
entry: ConfigEntry,
) -> None:
super().__init__(coordinator, entry)
self._attr_unique_id = f"{entry.entry_id}_foreground_window_title"
@property
def native_value(self) -> str | None:
return self._data.get("window_title")
class ForegroundPidSensor(_ForegroundEntityBase, SensorEntity):
_attr_icon = "mdi:identifier"
_attr_translation_key = "pid"
_attr_entity_registry_enabled_default = False # diagnostic-leaning
def __init__(
self,
coordinator: ForegroundCoordinator,
entry: ConfigEntry,
) -> None:
super().__init__(coordinator, entry)
self._attr_unique_id = f"{entry.entry_id}_foreground_pid"
@property
def native_value(self) -> int | None:
return self._data.get("pid")
class ForegroundMonitorSensor(_ForegroundEntityBase, SensorEntity):
_attr_icon = "mdi:monitor"
_attr_translation_key = "foreground_monitor"
def __init__(
self,
coordinator: ForegroundCoordinator,
entry: ConfigEntry,
) -> None:
super().__init__(coordinator, entry)
self._attr_unique_id = f"{entry.entry_id}_foreground_monitor"
@property
def native_value(self) -> int | None:
return self._data.get("monitor_id")
class ForegroundStartedAtSensor(_ForegroundEntityBase, SensorEntity):
"""Process start time as a timezone-aware datetime."""
_attr_icon = "mdi:clock-start"
_attr_translation_key = "process_started"
_attr_device_class = SensorDeviceClass.TIMESTAMP
_attr_entity_registry_enabled_default = False
def __init__(
self,
coordinator: ForegroundCoordinator,
entry: ConfigEntry,
) -> None:
super().__init__(coordinator, entry)
self._attr_unique_id = f"{entry.entry_id}_foreground_started_at"
@property
def native_value(self) -> datetime | None:
ts = self._data.get("started_at")
if ts is None:
return None
try:
return datetime.fromtimestamp(float(ts), tz=timezone.utc)
except (TypeError, ValueError, OSError):
return None
class ForegroundFullscreenBinarySensor(_ForegroundEntityBase, BinarySensorEntity):
_attr_icon = "mdi:fullscreen"
_attr_translation_key = "fullscreen"
def __init__(
self,
coordinator: ForegroundCoordinator,
entry: ConfigEntry,
) -> None:
super().__init__(coordinator, entry)
self._attr_unique_id = f"{entry.entry_id}_foreground_fullscreen"
@property
def is_on(self) -> bool:
return bool(self._data.get("is_fullscreen"))
class ForegroundMinimizedBinarySensor(_ForegroundEntityBase, BinarySensorEntity):
_attr_icon = "mdi:window-minimize"
_attr_translation_key = "minimized"
_attr_entity_registry_enabled_default = False
def __init__(
self,
coordinator: ForegroundCoordinator,
entry: ConfigEntry,
) -> None:
super().__init__(coordinator, entry)
self._attr_unique_id = f"{entry.entry_id}_foreground_minimized"
@property
def is_on(self) -> bool:
return bool(self._data.get("is_minimized"))
FOREGROUND_SENSORS: tuple[type[_ForegroundEntityBase], ...] = (
ForegroundProcessSensor,
ForegroundWindowTitleSensor,
ForegroundPidSensor,
ForegroundMonitorSensor,
ForegroundStartedAtSensor,
)
FOREGROUND_BINARY_SENSORS: tuple[type[_ForegroundEntityBase], ...] = (
ForegroundFullscreenBinarySensor,
ForegroundMinimizedBinarySensor,
)
@@ -0,0 +1,59 @@
"""Shared coordinator for the foreground (topmost) process snapshot.
The media server already broadcasts the foreground process over the media
WebSocket, but the WS client lives inside the media-player entity. Sensors
need their own polling fallback so they keep working when the user disables
the WebSocket feature in options, or while the WS is reconnecting.
"""
from __future__ import annotations
import logging
from datetime import timedelta
from typing import Any
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .api_client import MediaServerClient, MediaServerError
_LOGGER = logging.getLogger(__name__)
# Foreground polls fairly often — the user-facing value (process name)
# changes whenever the user alt-tabs, so a coarse poll would feel laggy.
# The server side is cached at ~500ms so even a 5s poll stays cheap.
DEFAULT_FOREGROUND_POLL_INTERVAL = 5
class ForegroundCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""Polls ``/api/foreground`` and fans out to sensor entities."""
def __init__(
self,
hass: HomeAssistant,
client: MediaServerClient,
poll_interval: int = DEFAULT_FOREGROUND_POLL_INTERVAL,
) -> None:
super().__init__(
hass,
_LOGGER,
name="Remote Media Player Foreground",
update_interval=timedelta(seconds=poll_interval),
)
self.client = client
async def _async_update_data(self) -> dict[str, Any]:
try:
return await self.client.get_foreground()
except MediaServerError as err:
raise UpdateFailed(f"Failed to fetch foreground info: {err}") from err
def apply_websocket_snapshot(self, data: dict[str, Any]) -> None:
"""Update from a push event (WebSocket) without an HTTP roundtrip.
Called by the media-player WS receiver when a ``foreground``/
``foreground_update`` frame arrives. Updates ``self.data`` directly
so all listening sensors refresh immediately, and avoids the next
scheduled poll spending bandwidth on the same value.
"""
self.async_set_updated_data(data)
@@ -172,6 +172,7 @@ class MediaPlayerCoordinator(DataUpdateCoordinator[dict[str, Any]]):
on_status_update=self._handle_ws_status_update,
on_disconnect=self._handle_ws_disconnect,
on_scripts_changed=self._handle_ws_scripts_changed,
on_foreground_update=self._handle_ws_foreground_update,
)
if await self._ws_client.connect():
@@ -206,6 +207,19 @@ class MediaPlayerCoordinator(DataUpdateCoordinator[dict[str, Any]]):
# Schedule reconnect attempt
self._schedule_reconnect()
@callback
def _handle_ws_foreground_update(self, data: dict[str, Any]) -> None:
"""Forward a foreground WS push into the shared foreground coordinator."""
if not self._entry:
return
try:
store = self.hass.data[DOMAIN][self._entry.entry_id]
except KeyError:
return
coordinator = store.get("foreground_coordinator")
if coordinator is not None:
coordinator.apply_websocket_snapshot(data)
@callback
def _handle_ws_scripts_changed(self) -> None:
"""Handle scripts changed notification from WebSocket."""
@@ -76,7 +76,7 @@ class _DisplayNumberBase(CoordinatorEntity[DisplayCoordinator], NumberEntity):
class DisplayBrightnessNumber(_DisplayNumberBase):
"""Number entity for controlling display brightness."""
_attr_name = "Brightness"
_attr_translation_key = "brightness"
_attr_icon = "mdi:brightness-6"
def __init__(
@@ -106,7 +106,7 @@ class DisplayBrightnessNumber(_DisplayNumberBase):
class DisplayContrastNumber(_DisplayNumberBase):
"""Number entity for controlling DDC/CI display contrast."""
_attr_name = "Contrast"
_attr_translation_key = "contrast"
_attr_icon = "mdi:contrast-circle"
def __init__(
@@ -73,7 +73,7 @@ class _DisplaySelectBase(CoordinatorEntity[DisplayCoordinator], SelectEntity):
class DisplayInputSourceSelect(_DisplaySelectBase):
"""Switch the monitor's active input (HDMI1, DP1, ...)."""
_attr_name = "Input source"
_attr_translation_key = "input_source"
_attr_icon = "mdi:video-input-hdmi"
def __init__(
@@ -113,7 +113,7 @@ class DisplayInputSourceSelect(_DisplaySelectBase):
class DisplayColorPresetSelect(_DisplaySelectBase):
"""Switch the monitor's color temperature preset (sRGB / 6500K / ...)."""
_attr_name = "Color preset"
_attr_translation_key = "color_preset"
_attr_icon = "mdi:palette"
def __init__(
@@ -155,7 +155,7 @@ class DisplayPictureModeSelect(_DisplaySelectBase):
are exposed as user-facing options and a label→code map drives writes.
"""
_attr_name = "Picture mode"
_attr_translation_key = "picture_mode"
_attr_icon = "mdi:image-multiple"
def __init__(
+22 -13
View File
@@ -15,6 +15,8 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .display_coordinator import DisplayCoordinator
from .display_device import display_device_info
from .foreground import FOREGROUND_SENSORS
from .foreground_coordinator import ForegroundCoordinator
_LOGGER = logging.getLogger(__name__)
@@ -24,30 +26,37 @@ async def async_setup_entry(
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up per-display sensor entities."""
coordinator: DisplayCoordinator = hass.data[DOMAIN][entry.entry_id][
"display_coordinator"
]
"""Set up display + foreground sensor entities."""
store = hass.data[DOMAIN][entry.entry_id]
display_coordinator: DisplayCoordinator = store["display_coordinator"]
foreground_coordinator: ForegroundCoordinator | None = store.get(
"foreground_coordinator"
)
if not coordinator.data:
return
entities: list[Any] = []
entities = [
DisplayResolutionSensor(coordinator, entry, monitor)
for monitor in coordinator.data.values()
if monitor.get("resolution")
]
if display_coordinator.data:
entities.extend(
DisplayResolutionSensor(display_coordinator, entry, monitor)
for monitor in display_coordinator.data.values()
if monitor.get("resolution")
)
if foreground_coordinator is not None:
entities.extend(
cls(foreground_coordinator, entry) for cls in FOREGROUND_SENSORS
)
if entities:
async_add_entities(entities)
_LOGGER.info("Added %d display sensor entities", len(entities))
_LOGGER.info("Added %d sensor entities (display + foreground)", len(entities))
class DisplayResolutionSensor(CoordinatorEntity[DisplayCoordinator], SensorEntity):
"""Diagnostic sensor reporting the EDID-derived display resolution."""
_attr_has_entity_name = True
_attr_name = "Resolution"
_attr_translation_key = "resolution"
_attr_entity_category = EntityCategory.DIAGNOSTIC
_attr_icon = "mdi:monitor-screenshot"
@@ -42,6 +42,34 @@
}
}
},
"entity": {
"binary_sensor": {
"primary_display": { "name": "Primary display" },
"power_control_supported": { "name": "Power control supported" },
"fullscreen": { "name": "Fullscreen" },
"minimized": { "name": "Minimized" }
},
"sensor": {
"resolution": { "name": "Resolution" },
"foreground_process": { "name": "Foreground process" },
"window_title": { "name": "Window title" },
"pid": { "name": "PID" },
"foreground_monitor": { "name": "Monitor" },
"process_started": { "name": "Process started" }
},
"number": {
"brightness": { "name": "Brightness" },
"contrast": { "name": "Contrast" }
},
"switch": {
"power": { "name": "Power" }
},
"select": {
"input_source": { "name": "Input source" },
"color_preset": { "name": "Color preset" },
"picture_mode": { "name": "Picture mode" }
}
},
"services": {
"execute_script": {
"name": "Execute Script",
@@ -48,7 +48,7 @@ class DisplayPowerSwitch(CoordinatorEntity[DisplayCoordinator], SwitchEntity):
_attr_has_entity_name = True
_attr_device_class = SwitchDeviceClass.SWITCH
_attr_name = "Power"
_attr_translation_key = "power"
def __init__(
self,
@@ -42,6 +42,34 @@
}
}
},
"entity": {
"binary_sensor": {
"primary_display": { "name": "Primary display" },
"power_control_supported": { "name": "Power control supported" },
"fullscreen": { "name": "Fullscreen" },
"minimized": { "name": "Minimized" }
},
"sensor": {
"resolution": { "name": "Resolution" },
"foreground_process": { "name": "Foreground process" },
"window_title": { "name": "Window title" },
"pid": { "name": "PID" },
"foreground_monitor": { "name": "Monitor" },
"process_started": { "name": "Process started" }
},
"number": {
"brightness": { "name": "Brightness" },
"contrast": { "name": "Contrast" }
},
"switch": {
"power": { "name": "Power" }
},
"select": {
"input_source": { "name": "Input source" },
"color_preset": { "name": "Color preset" },
"picture_mode": { "name": "Picture mode" }
}
},
"services": {
"execute_script": {
"name": "Execute Script",
@@ -42,6 +42,34 @@
}
}
},
"entity": {
"binary_sensor": {
"primary_display": { "name": "Основной дисплей" },
"power_control_supported": { "name": "Поддержка управления питанием" },
"fullscreen": { "name": "Полноэкранный режим" },
"minimized": { "name": "Свёрнуто" }
},
"sensor": {
"resolution": { "name": "Разрешение" },
"foreground_process": { "name": "Активный процесс" },
"window_title": { "name": "Заголовок окна" },
"pid": { "name": "PID" },
"foreground_monitor": { "name": "Монитор" },
"process_started": { "name": "Запуск процесса" }
},
"number": {
"brightness": { "name": "Яркость" },
"contrast": { "name": "Контрастность" }
},
"switch": {
"power": { "name": "Питание" }
},
"select": {
"input_source": { "name": "Источник сигнала" },
"color_preset": { "name": "Цветовая температура" },
"picture_mode": { "name": "Режим изображения" }
}
},
"services": {
"execute_script": {
"name": "Выполнить скрипт",