Media Player

media_player

media_player domain implementation.

This module also contains the FavoriteItem helper returned by MediaPlayer.favorites, which recursively traverses the media_player/browse_media tree and flattens it into a list of directly playable items.

SPEC `module-attribute`

SPEC: DomainSpec[MediaPlayer] = register_domain(DomainSpec(name='media_player', entity_cls=MediaPlayer))

The DomainSpec registered with the shared DomainRegistry.

NowPlaying `dataclass`

Structured snapshot of the media currently playing on a media player.

Groups identity-related media attributes into a single object. Position/progress fields are intentionally excluded — they change continuously during playback. Instances are frozen so two snapshots can be compared with == to detect whether the playing media changed.

Attributes:

Name	Type	Description
`source`	`str or None`	Active input source.
`title`	`str or None`	Media title.
`artist`	`str or None`	Media artist.
`album`	`str or None`	Album name.
`channel`	`str or None`	Channel name.
`content_type`	`str or None`	Media content type identifier.
`content_id`	`str or None`	Media content id.
`duration`	`int or None`	Media duration in seconds.
`entity_picture`	`str or None`	Absolute URL of the entity picture / album art.
`queue_position`	`int or None`	Current position in the play queue.
`queue_size`	`int or None`	Total items in the play queue.
`playlist`	`str or None`	Playlist name.
`repeat`	`str or None`	Repeat mode.
`next`	`bool`	Whether skip-next is supported.
`previous`	`bool`	Whether skip-previous is supported.

Source code in src/haclient/domains/media_player.py

@dataclass(frozen=True)
class NowPlaying:
    """Structured snapshot of the media currently playing on a media player.

    Groups identity-related media attributes into a single object.
    Position/progress fields are intentionally excluded — they change
    continuously during playback. Instances are frozen so two snapshots
    can be compared with ``==`` to detect whether the playing media
    changed.

    Attributes
    ----------
    source : str or None
        Active input source.
    title : str or None
        Media title.
    artist : str or None
        Media artist.
    album : str or None
        Album name.
    channel : str or None
        Channel name.
    content_type : str or None
        Media content type identifier.
    content_id : str or None
        Media content id.
    duration : int or None
        Media duration in seconds.
    entity_picture : str or None
        Absolute URL of the entity picture / album art.
    queue_position : int or None
        Current position in the play queue.
    queue_size : int or None
        Total items in the play queue.
    playlist : str or None
        Playlist name.
    repeat : str or None
        Repeat mode.
    next : bool
        Whether skip-next is supported.
    previous : bool
        Whether skip-previous is supported.
    """

    source: str | None = None
    title: str | None = None
    artist: str | None = None
    album: str | None = None
    channel: str | None = None
    content_type: str | None = None
    content_id: str | None = None
    duration: int | None = None
    entity_picture: str | None = None
    queue_position: int | None = None
    queue_size: int | None = None
    playlist: str | None = None
    repeat: str | None = None
    next: bool = False
    previous: bool = False

FavoriteItem

A flattened, directly-playable entry discovered via browse_media.

The item remembers which MediaPlayer it belongs to, along with the media_content_id / media_content_type pair needed to play it.

Attributes:

Name	Type	Description
`title`	`str`	Display title.
`media_content_id`	`str`	Content identifier for playback.
`media_content_type`	`str`	Content type for playback.
`thumbnail`	`str or None`	Image URL.
`category`	`str or None`	Human-readable category label.
`media_class`	`str or None`	Raw media class from Home Assistant.

Source code in src/haclient/domains/media_player.py

class FavoriteItem:
    """A flattened, directly-playable entry discovered via ``browse_media``.

    The item remembers which `MediaPlayer` it belongs to, along with the
    ``media_content_id`` / ``media_content_type`` pair needed to play it.

    Attributes
    ----------
    title : str
        Display title.
    media_content_id : str
        Content identifier for playback.
    media_content_type : str
        Content type for playback.
    thumbnail : str or None
        Image URL.
    category : str or None
        Human-readable category label.
    media_class : str or None
        Raw media class from Home Assistant.
    """

    __slots__ = (
        "title",
        "media_content_id",
        "media_content_type",
        "thumbnail",
        "category",
        "media_class",
        "_player",
    )

    def __init__(
        self,
        *,
        title: str,
        media_content_id: str,
        media_content_type: str,
        player: MediaPlayer,
        thumbnail: str | None = None,
        category: str | None = None,
        media_class: str | None = None,
    ) -> None:
        self.title = title
        self.media_content_id = media_content_id
        self.media_content_type = media_content_type
        self.thumbnail = thumbnail
        self.category = category
        self.media_class = media_class
        self._player = player

    async def play(self) -> None:
        """Play this favorite on its `MediaPlayer`.

        Delegates to `MediaPlayer.play_media` with the captured
        ``media_content_type`` and ``media_content_id``, which invokes
        the ``media_player.play_media`` Home Assistant service.

        Raises
        ------
        CommandError
            If Home Assistant rejects the service call.
        HTTPError
            If the REST call returns a non-2xx response.
        TimeoutError
            If the call exceeds the configured request timeout.
        ConnectionClosedError
            If the WebSocket disconnects mid-call.
        """
        await self._player.play_media(self.media_content_type, self.media_content_id)

    def __repr__(self) -> str:
        return (
            f"FavoriteItem(title={self.title!r}, "
            f"category={self.category!r}, "
            f"media_class={self.media_class!r}, "
            f"media_content_type={self.media_content_type!r}, "
            f"media_content_id={self.media_content_id!r}, "
            f"thumbnail={self.thumbnail!r})"
        )

play `async`

play() -> None

Play this favorite on its MediaPlayer.

Delegates to MediaPlayer.play_media with the captured media_content_type and media_content_id, which invokes the media_player.play_media Home Assistant service.

Raises:

Type	Description
`CommandError`	If Home Assistant rejects the service call.
`HTTPError`	If the REST call returns a non-2xx response.
`TimeoutError`	If the call exceeds the configured request timeout.
`ConnectionClosedError`	If the WebSocket disconnects mid-call.

Source code in src/haclient/domains/media_player.py

async def play(self) -> None:
    """Play this favorite on its `MediaPlayer`.

    Delegates to `MediaPlayer.play_media` with the captured
    ``media_content_type`` and ``media_content_id``, which invokes
    the ``media_player.play_media`` Home Assistant service.

    Raises
    ------
    CommandError
        If Home Assistant rejects the service call.
    HTTPError
        If the REST call returns a non-2xx response.
    TimeoutError
        If the call exceeds the configured request timeout.
    ConnectionClosedError
        If the WebSocket disconnects mid-call.
    """
    await self._player.play_media(self.media_content_type, self.media_content_id)

MediaPlayer

Bases: Entity

A Home Assistant media player entity.

Provides intent-specific methods for playback control, volume management, power, source selection, and media browsing/favorites.

Source code in src/haclient/domains/media_player.py

class MediaPlayer(Entity):
    """A Home Assistant media player entity.

    Provides intent-specific methods for playback control, volume
    management, power, source selection, and media browsing/favorites.
    """

    domain = "media_player"

    def __init__(
        self,
        entity_id: str,
        services: ServiceCaller,
        store: StateStore,
        clock: Clock,
    ) -> None:
        """Initialise the media player and its media-change listener list.

        Parameters
        ----------
        entity_id : str
            Fully-qualified entity id (e.g. ``"media_player.kitchen"``).
        services : ServiceCaller
            Service-call port used to invoke HA services.
        store : StateStore
            State store the entity registers itself with.
        clock : Clock
            Scheduler used to dispatch async listeners.
        """
        super().__init__(entity_id, services, store, clock)
        self._media_change_listeners: list[ValueChangeHandler] = []

    # -- Listener decorators ------------------------------------------

    def on_volume_change(self, func: ValueChangeHandler) -> ValueChangeHandler:
        """Register a listener for volume level changes.

        Parameters
        ----------
        func : callable
            Callable invoked with ``(old_value, new_value)`` whenever
            the ``volume_level`` attribute (``0.0``-``1.0``) changes.

        Returns
        -------
        callable
            The same *func*, returned for decorator use.
        """
        return self._register_attr_listener("volume_level", func)

    def on_mute_change(self, func: ValueChangeHandler) -> ValueChangeHandler:
        """Register a listener for mute state changes.

        Parameters
        ----------
        func : callable
            Callable invoked with ``(old_value, new_value)`` whenever
            the ``is_volume_muted`` attribute changes.

        Returns
        -------
        callable
            The same *func*, returned for decorator use.
        """
        return self._register_attr_listener("is_volume_muted", func)

    def on_media_change(self, func: ValueChangeHandler) -> ValueChangeHandler:
        """Register a listener for when the playing media changes.

        Parameters
        ----------
        func : callable
            Callable invoked with ``(old, new)`` where both arguments
            are `NowPlaying` snapshots whenever they differ.

        Returns
        -------
        callable
            The same *func*, returned for decorator use.
        """
        self._media_change_listeners.append(func)
        return func

    def on_play(self, func: ValueChangeHandler) -> ValueChangeHandler:
        """Register a listener for when playback starts.

        Parameters
        ----------
        func : callable
            Sync or async callable invoked with ``(old_state, new_state)``
            on every transition into the ``playing`` state.

        Returns
        -------
        callable
            The same *func*, returned for decorator use.
        """
        return self._register_state_transition_listener("playing", func)

    def on_pause(self, func: ValueChangeHandler) -> ValueChangeHandler:
        """Register a listener for when playback pauses.

        Parameters
        ----------
        func : callable
            Sync or async callable invoked with ``(old_state, new_state)``
            on every transition into the ``paused`` state.

        Returns
        -------
        callable
            The same *func*, returned for decorator use.
        """
        return self._register_state_transition_listener("paused", func)

    def on_stop(self, func: ValueChangeHandler) -> ValueChangeHandler:
        """Register a listener for when playback stops.

        Parameters
        ----------
        func : callable
            Sync or async callable invoked with ``(old_state, new_state)``
            on every transition into the ``idle`` state.

        Returns
        -------
        callable
            The same *func*, returned for decorator use.
        """
        return self._register_state_transition_listener("idle", func)

    def _dispatch_granular_events(
        self,
        old_state: dict[str, Any] | None,
        new_state: dict[str, Any] | None,
    ) -> None:
        """Dispatch base events plus :meth:`on_media_change`."""
        super()._dispatch_granular_events(old_state, new_state)
        old_attrs = (old_state or {}).get("attributes") or {}
        new_attrs = (new_state or {}).get("attributes") or {}
        base = self._services.rest.base_url
        old_np = _now_playing_from_attrs(old_attrs, base)
        new_np = _now_playing_from_attrs(new_attrs, base)
        if old_np != new_np:
            for listener in list(self._media_change_listeners):
                self._schedule_value(listener, old_np, new_np)

    def remove_granular_listener(self, func: ValueChangeHandler) -> None:
        """Remove a granular listener, including media-change listeners.

        Parameters
        ----------
        func : ValueChangeHandler
            The exact handler previously registered via one of the
            ``on_*`` listener methods. Unknown handlers are silently
            ignored.
        """
        with contextlib.suppress(ValueError):
            self._media_change_listeners.remove(func)
            return
        super().remove_granular_listener(func)

    # -- State properties ---------------------------------------------

    @property
    def is_playing(self) -> bool:
        """Whether the media player is currently playing."""
        return self.state == "playing"

    @property
    def is_paused(self) -> bool:
        """Whether the media player is currently paused."""
        return self.state == "paused"

    @property
    def is_muted(self) -> bool:
        """Whether the media player is currently muted."""
        return bool(self.attributes.get("is_volume_muted"))

    @property
    def volume_level(self) -> float | None:
        """Current volume level (``0.0``–``1.0``) or ``None``."""
        value = self.attributes.get("volume_level")
        return float(value) if isinstance(value, (int, float)) else None

    @property
    def now_playing(self) -> NowPlaying:
        """Structured snapshot of the currently playing media.

        Returns
        -------
        NowPlaying
            A fresh, frozen snapshot built from the entity's current
            attributes. The ``entity_picture`` field is resolved against
            the REST `base_url` so consumers receive an absolute URL.
            Two snapshots can be compared with ``==`` to detect whether
            the playing media actually changed.

        Notes
        -----
        Every access constructs a new dataclass; no caching or I/O is
        performed. The result is independent of any subsequent state
        update, so it is safe to retain across event-loop turns.
        """
        return _now_playing_from_attrs(self.attributes, self._services.rest.base_url)

    # -- Actions ------------------------------------------------------

    async def play(self) -> None:
        """Resume / start playback.

        Invokes the ``media_player.media_play`` Home Assistant service.

        Raises
        ------
        CommandError
            If Home Assistant rejects the service call (for example, no
            media is loaded).
        HTTPError
            If the REST call returns a non-2xx response.
        TimeoutError
            If the call exceeds the configured request timeout.
        ConnectionClosedError
            If the WebSocket disconnects mid-call.
        """
        await self._call_service("media_play")

    async def pause(self) -> None:
        """Pause playback.

        Invokes the ``media_player.media_pause`` Home Assistant service.

        Raises
        ------
        CommandError
            If Home Assistant rejects the service call.
        HTTPError
            If the REST call returns a non-2xx response.
        TimeoutError
            If the call exceeds the configured request timeout.
        ConnectionClosedError
            If the WebSocket disconnects mid-call.
        """
        await self._call_service("media_pause")

    async def play_pause(self) -> None:
        """Toggle play/pause.

        Invokes the ``media_player.media_play_pause`` Home Assistant
        service.

        Raises
        ------
        CommandError
            If Home Assistant rejects the service call.
        HTTPError
            If the REST call returns a non-2xx response.
        TimeoutError
            If the call exceeds the configured request timeout.
        ConnectionClosedError
            If the WebSocket disconnects mid-call.
        """
        await self._call_service("media_play_pause")

    async def stop(self) -> None:
        """Stop playback.

        Invokes the ``media_player.media_stop`` Home Assistant service.

        Raises
        ------
        CommandError
            If Home Assistant rejects the service call.
        HTTPError
            If the REST call returns a non-2xx response.
        TimeoutError
            If the call exceeds the configured request timeout.
        ConnectionClosedError
            If the WebSocket disconnects mid-call.
        """
        await self._call_service("media_stop")

    async def next(self) -> None:
        """Skip to the next track.

        Invokes the ``media_player.media_next_track`` Home Assistant
        service. This method does **not** consult
        `NowPlaying.next`; calls to players that do not advertise
        skip-next support will surface as `CommandError`. Pre-check
        ``self.now_playing.next`` to avoid that.

        Raises
        ------
        CommandError
            If Home Assistant rejects the service call (e.g. the player
            does not support skip-next).
        HTTPError
            If the REST call returns a non-2xx response.
        TimeoutError
            If the call exceeds the configured request timeout.
        ConnectionClosedError
            If the WebSocket disconnects mid-call.
        """
        await self._call_service("media_next_track")

    async def previous(self) -> None:
        """Skip to the previous track.

        Invokes the ``media_player.media_previous_track`` Home Assistant
        service. This method does **not** consult
        `NowPlaying.previous`; calls to players that do not advertise
        skip-previous support will surface as `CommandError`. Pre-check
        ``self.now_playing.previous`` to avoid that.

        Raises
        ------
        CommandError
            If Home Assistant rejects the service call.
        HTTPError
            If the REST call returns a non-2xx response.
        TimeoutError
            If the call exceeds the configured request timeout.
        ConnectionClosedError
            If the WebSocket disconnects mid-call.
        """
        await self._call_service("media_previous_track")

    async def set_volume(self, level: float) -> None:
        """Set the volume level.

        Parameters
        ----------
        level : float
            Volume level between ``0.0`` and ``1.0``.

        Raises
        ------
        ValueError
            If *level* is outside the ``[0.0, 1.0]`` range.
        """
        if not 0.0 <= level <= 1.0:
            raise ValueError("Volume level must be between 0.0 and 1.0")
        await self._call_service("volume_set", {"volume_level": float(level)})

    async def mute(self, muted: bool = True) -> None:
        """Mute or unmute the media player.

        Parameters
        ----------
        muted : bool, optional
            ``True`` to mute (default), ``False`` to unmute.
        """
        await self._call_service("volume_mute", {"is_volume_muted": bool(muted)})

    async def power_on(self) -> None:
        """Power the media player on.

        Invokes the ``media_player.turn_on`` Home Assistant service.

        Raises
        ------
        CommandError
            If Home Assistant rejects the service call.
        HTTPError
            If the REST call returns a non-2xx response.
        TimeoutError
            If the call exceeds the configured request timeout.
        ConnectionClosedError
            If the WebSocket disconnects mid-call.
        """
        await self._call_service("turn_on")

    async def power_off(self) -> None:
        """Power the media player off.

        Invokes the ``media_player.turn_off`` Home Assistant service.

        Raises
        ------
        CommandError
            If Home Assistant rejects the service call.
        HTTPError
            If the REST call returns a non-2xx response.
        TimeoutError
            If the call exceeds the configured request timeout.
        ConnectionClosedError
            If the WebSocket disconnects mid-call.
        """
        await self._call_service("turn_off")

    async def select_source(self, source: str) -> None:
        """Select an input source.

        Parameters
        ----------
        source : str
            Name of the input source. Must be one of the values reported
            in the entity's ``source_list`` attribute.
        """
        await self._call_service("select_source", {"source": source})

    async def play_media(
        self,
        media_content_type: str,
        media_content_id: str,
        **extra: Any,
    ) -> None:
        """Play a specific media item.

        Parameters
        ----------
        media_content_type : str
            Media content type identifier (e.g. ``"music"``).
        media_content_id : str
            Media content id understood by the underlying integration.
        **extra : Any
            Additional fields forwarded verbatim to Home Assistant
            (e.g. ``enqueue``, ``announce``).
        """
        data: dict[str, Any] = {
            "media_content_type": media_content_type,
            "media_content_id": media_content_id,
            **extra,
        }
        await self._call_service("play_media", data)

    async def browse_media(
        self,
        media_content_type: str | None = None,
        media_content_id: str | None = None,
    ) -> dict[str, Any]:
        """Issue a single ``media_player/browse_media`` WebSocket command.

        Parameters
        ----------
        media_content_type : str or None, optional
            Restrict browsing to this content type. ``None`` returns the
            root browse node.
        media_content_id : str or None, optional
            Restrict browsing to this content id. ``None`` returns the
            root browse node.

        Returns
        -------
        dict
            The raw result dictionary from Home Assistant.

        Raises
        ------
        HAClientError
            If the command fails or returns an unexpected response.
        """
        payload: dict[str, Any] = {
            "type": "media_player/browse_media",
            "entity_id": self.entity_id,
        }
        if media_content_type is not None:
            payload["media_content_type"] = media_content_type
        if media_content_id is not None:
            payload["media_content_id"] = media_content_id
        result = await self._services.ws.send_command(payload)
        if not isinstance(result, dict):
            raise HAClientError("Unexpected browse_media response")
        return result

    async def favorites(
        self,
        *,
        max_depth: int = _MAX_BROWSE_DEPTH,
        max_nodes: int = _MAX_BROWSE_NODES,
    ) -> list[FavoriteItem]:
        """Return a flattened list of playable items in the media tree.

        Parameters
        ----------
        max_depth : int, optional
            Maximum recursion depth.
        max_nodes : int, optional
            Hard upper bound on total nodes visited.

        Returns
        -------
        list of FavoriteItem
            The discovered playable items.
        """
        try:
            root = await self.browse_media()
        except CommandError as err:
            _LOGGER.debug("browse_media unsupported for %s: %s", self.entity_id, err)
            return []
        except HAClientError as err:
            _LOGGER.debug("browse_media failed for %s: %s", self.entity_id, err)
            return []

        collected: list[FavoriteItem] = []
        seen: set[tuple[str, str]] = set()
        node_count = 0

        async def walk(node: dict[str, Any], depth: int, category: str | None) -> None:
            """Recurse through *node*'s children, collecting playable items.

            Updates *collected*, *seen*, and *node_count* from the
            enclosing scope. Honours the *max_depth* and *max_nodes*
            bounds and silently skips children that are missing
            playback identifiers.
            """
            nonlocal node_count
            if node_count >= max_nodes:
                return
            node_count += 1

            children = node.get("children")
            if not isinstance(children, list):
                return

            for child in children:
                if not isinstance(child, dict):
                    continue
                content_id = child.get("media_content_id")
                content_type = child.get("media_content_type")
                title = child.get("title") or child.get("name") or ""
                can_play = bool(child.get("can_play"))
                can_expand = bool(child.get("can_expand"))
                thumbnail = child.get("thumbnail")
                media_class = child.get("media_class")

                if can_play and isinstance(content_id, str) and isinstance(content_type, str):
                    key = (content_type, content_id)
                    if key not in seen:
                        seen.add(key)
                        collected.append(
                            FavoriteItem(
                                title=str(title),
                                media_content_id=content_id,
                                media_content_type=content_type,
                                player=self,
                                thumbnail=(thumbnail if isinstance(thumbnail, str) else None),
                                category=category
                                or (media_class if isinstance(media_class, str) else None),
                                media_class=(media_class if isinstance(media_class, str) else None),
                            )
                        )

                if (
                    can_expand
                    and depth + 1 < max_depth
                    and isinstance(content_id, str)
                    and isinstance(content_type, str)
                ):
                    try:
                        sub = await self.browse_media(content_type, content_id)
                    except (CommandError, HAClientError) as err:
                        _LOGGER.debug(
                            "browse_media sublevel failed (%s/%s): %s",
                            content_type,
                            content_id,
                            err,
                        )
                        continue
                    except asyncio.CancelledError:
                        raise
                    child_category = str(title) if title else category
                    await walk(sub, depth + 1, child_category)

        await walk(root, 0, None)
        return collected

is_playing `property`

is_playing: bool

Whether the media player is currently playing.

is_paused `property`

is_paused: bool

Whether the media player is currently paused.

is_muted `property`

is_muted: bool

Whether the media player is currently muted.

volume_level `property`

volume_level: float | None

Current volume level (0.0–1.0) or None.

now_playing `property`

now_playing: NowPlaying

Structured snapshot of the currently playing media.

Returns:

Type	Description
`NowPlaying`	A fresh, frozen snapshot built from the entity's current attributes. The `entity_picture` field is resolved against the REST `base_url` so consumers receive an absolute URL. Two snapshots can be compared with `==` to detect whether the playing media actually changed.

Notes

Every access constructs a new dataclass; no caching or I/O is performed. The result is independent of any subsequent state update, so it is safe to retain across event-loop turns.

init

__init__(entity_id: str, services: ServiceCaller, store: StateStore, clock: Clock) -> None

Initialise the media player and its media-change listener list.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	Fully-qualified entity id (e.g. `"media_player.kitchen"`).	required
`services`	`ServiceCaller`	Service-call port used to invoke HA services.	required
`store`	`StateStore`	State store the entity registers itself with.	required
`clock`	`Clock`	Scheduler used to dispatch async listeners.	required

Source code in src/haclient/domains/media_player.py

def __init__(
    self,
    entity_id: str,
    services: ServiceCaller,
    store: StateStore,
    clock: Clock,
) -> None:
    """Initialise the media player and its media-change listener list.

    Parameters
    ----------
    entity_id : str
        Fully-qualified entity id (e.g. ``"media_player.kitchen"``).
    services : ServiceCaller
        Service-call port used to invoke HA services.
    store : StateStore
        State store the entity registers itself with.
    clock : Clock
        Scheduler used to dispatch async listeners.
    """
    super().__init__(entity_id, services, store, clock)
    self._media_change_listeners: list[ValueChangeHandler] = []

on_volume_change

on_volume_change(func: ValueChangeHandler) -> ValueChangeHandler

Register a listener for volume level changes.

Parameters:

Name	Type	Description	Default
`func`	`callable`	Callable invoked with `(old_value, new_value)` whenever the `volume_level` attribute (`0.0`-`1.0`) changes.	required

Returns:

Type	Description
`callable`	The same func, returned for decorator use.

Source code in src/haclient/domains/media_player.py

def on_volume_change(self, func: ValueChangeHandler) -> ValueChangeHandler:
    """Register a listener for volume level changes.

    Parameters
    ----------
    func : callable
        Callable invoked with ``(old_value, new_value)`` whenever
        the ``volume_level`` attribute (``0.0``-``1.0``) changes.

    Returns
    -------
    callable
        The same *func*, returned for decorator use.
    """
    return self._register_attr_listener("volume_level", func)

on_mute_change

on_mute_change(func: ValueChangeHandler) -> ValueChangeHandler

Register a listener for mute state changes.

Parameters:

Name	Type	Description	Default
`func`	`callable`	Callable invoked with `(old_value, new_value)` whenever the `is_volume_muted` attribute changes.	required

Returns:

Type	Description
`callable`	The same func, returned for decorator use.

Source code in src/haclient/domains/media_player.py

def on_mute_change(self, func: ValueChangeHandler) -> ValueChangeHandler:
    """Register a listener for mute state changes.

    Parameters
    ----------
    func : callable
        Callable invoked with ``(old_value, new_value)`` whenever
        the ``is_volume_muted`` attribute changes.

    Returns
    -------
    callable
        The same *func*, returned for decorator use.
    """
    return self._register_attr_listener("is_volume_muted", func)

on_media_change

on_media_change(func: ValueChangeHandler) -> ValueChangeHandler

Register a listener for when the playing media changes.

Parameters:

Name	Type	Description	Default
`func`	`callable`	Callable invoked with `(old, new)` where both arguments are `NowPlaying` snapshots whenever they differ.	required

Returns:

Type	Description
`callable`	The same func, returned for decorator use.

Source code in src/haclient/domains/media_player.py

def on_media_change(self, func: ValueChangeHandler) -> ValueChangeHandler:
    """Register a listener for when the playing media changes.

    Parameters
    ----------
    func : callable
        Callable invoked with ``(old, new)`` where both arguments
        are `NowPlaying` snapshots whenever they differ.

    Returns
    -------
    callable
        The same *func*, returned for decorator use.
    """
    self._media_change_listeners.append(func)
    return func

on_play

on_play(func: ValueChangeHandler) -> ValueChangeHandler

Register a listener for when playback starts.

Parameters:

Name	Type	Description	Default
`func`	`callable`	Sync or async callable invoked with `(old_state, new_state)` on every transition into the `playing` state.	required

Returns:

Type	Description
`callable`	The same func, returned for decorator use.

Source code in src/haclient/domains/media_player.py

def on_play(self, func: ValueChangeHandler) -> ValueChangeHandler:
    """Register a listener for when playback starts.

    Parameters
    ----------
    func : callable
        Sync or async callable invoked with ``(old_state, new_state)``
        on every transition into the ``playing`` state.

    Returns
    -------
    callable
        The same *func*, returned for decorator use.
    """
    return self._register_state_transition_listener("playing", func)

on_pause

on_pause(func: ValueChangeHandler) -> ValueChangeHandler

Register a listener for when playback pauses.

Parameters:

Name	Type	Description	Default
`func`	`callable`	Sync or async callable invoked with `(old_state, new_state)` on every transition into the `paused` state.	required

Returns:

Type	Description
`callable`	The same func, returned for decorator use.

Source code in src/haclient/domains/media_player.py

def on_pause(self, func: ValueChangeHandler) -> ValueChangeHandler:
    """Register a listener for when playback pauses.

    Parameters
    ----------
    func : callable
        Sync or async callable invoked with ``(old_state, new_state)``
        on every transition into the ``paused`` state.

    Returns
    -------
    callable
        The same *func*, returned for decorator use.
    """
    return self._register_state_transition_listener("paused", func)

on_stop

on_stop(func: ValueChangeHandler) -> ValueChangeHandler

Register a listener for when playback stops.

Parameters:

Name	Type	Description	Default
`func`	`callable`	Sync or async callable invoked with `(old_state, new_state)` on every transition into the `idle` state.	required

Returns:

Type	Description
`callable`	The same func, returned for decorator use.

Source code in src/haclient/domains/media_player.py

def on_stop(self, func: ValueChangeHandler) -> ValueChangeHandler:
    """Register a listener for when playback stops.

    Parameters
    ----------
    func : callable
        Sync or async callable invoked with ``(old_state, new_state)``
        on every transition into the ``idle`` state.

    Returns
    -------
    callable
        The same *func*, returned for decorator use.
    """
    return self._register_state_transition_listener("idle", func)

remove_granular_listener

remove_granular_listener(func: ValueChangeHandler) -> None

Remove a granular listener, including media-change listeners.

Parameters:

Name	Type	Description	Default
`func`	`ValueChangeHandler`	The exact handler previously registered via one of the `on_*` listener methods. Unknown handlers are silently ignored.	required

Source code in src/haclient/domains/media_player.py

def remove_granular_listener(self, func: ValueChangeHandler) -> None:
    """Remove a granular listener, including media-change listeners.

    Parameters
    ----------
    func : ValueChangeHandler
        The exact handler previously registered via one of the
        ``on_*`` listener methods. Unknown handlers are silently
        ignored.
    """
    with contextlib.suppress(ValueError):
        self._media_change_listeners.remove(func)
        return
    super().remove_granular_listener(func)

play `async`

play() -> None

Resume / start playback.

Invokes the media_player.media_play Home Assistant service.

Raises:

Type	Description
`CommandError`	If Home Assistant rejects the service call (for example, no media is loaded).
`HTTPError`	If the REST call returns a non-2xx response.
`TimeoutError`	If the call exceeds the configured request timeout.
`ConnectionClosedError`	If the WebSocket disconnects mid-call.

Source code in src/haclient/domains/media_player.py

async def play(self) -> None:
    """Resume / start playback.

    Invokes the ``media_player.media_play`` Home Assistant service.

    Raises
    ------
    CommandError
        If Home Assistant rejects the service call (for example, no
        media is loaded).
    HTTPError
        If the REST call returns a non-2xx response.
    TimeoutError
        If the call exceeds the configured request timeout.
    ConnectionClosedError
        If the WebSocket disconnects mid-call.
    """
    await self._call_service("media_play")

pause `async`

pause() -> None

Pause playback.

Invokes the media_player.media_pause Home Assistant service.

Raises:

Type	Description
`CommandError`	If Home Assistant rejects the service call.
`HTTPError`	If the REST call returns a non-2xx response.
`TimeoutError`	If the call exceeds the configured request timeout.
`ConnectionClosedError`	If the WebSocket disconnects mid-call.

Source code in src/haclient/domains/media_player.py

async def pause(self) -> None:
    """Pause playback.

    Invokes the ``media_player.media_pause`` Home Assistant service.

    Raises
    ------
    CommandError
        If Home Assistant rejects the service call.
    HTTPError
        If the REST call returns a non-2xx response.
    TimeoutError
        If the call exceeds the configured request timeout.
    ConnectionClosedError
        If the WebSocket disconnects mid-call.
    """
    await self._call_service("media_pause")

play_pause `async`

play_pause() -> None

Toggle play/pause.

Invokes the media_player.media_play_pause Home Assistant service.

Raises:

Type	Description
`CommandError`	If Home Assistant rejects the service call.
`HTTPError`	If the REST call returns a non-2xx response.
`TimeoutError`	If the call exceeds the configured request timeout.
`ConnectionClosedError`	If the WebSocket disconnects mid-call.

Source code in src/haclient/domains/media_player.py

async def play_pause(self) -> None:
    """Toggle play/pause.

    Invokes the ``media_player.media_play_pause`` Home Assistant
    service.

    Raises
    ------
    CommandError
        If Home Assistant rejects the service call.
    HTTPError
        If the REST call returns a non-2xx response.
    TimeoutError
        If the call exceeds the configured request timeout.
    ConnectionClosedError
        If the WebSocket disconnects mid-call.
    """
    await self._call_service("media_play_pause")

stop `async`

stop() -> None

Stop playback.

Invokes the media_player.media_stop Home Assistant service.

Raises:

Type	Description
`CommandError`	If Home Assistant rejects the service call.
`HTTPError`	If the REST call returns a non-2xx response.
`TimeoutError`	If the call exceeds the configured request timeout.
`ConnectionClosedError`	If the WebSocket disconnects mid-call.

Source code in src/haclient/domains/media_player.py

async def stop(self) -> None:
    """Stop playback.

    Invokes the ``media_player.media_stop`` Home Assistant service.

    Raises
    ------
    CommandError
        If Home Assistant rejects the service call.
    HTTPError
        If the REST call returns a non-2xx response.
    TimeoutError
        If the call exceeds the configured request timeout.
    ConnectionClosedError
        If the WebSocket disconnects mid-call.
    """
    await self._call_service("media_stop")

next `async`

next() -> None

Skip to the next track.

Invokes the media_player.media_next_track Home Assistant service. This method does not consult NowPlaying.next; calls to players that do not advertise skip-next support will surface as CommandError. Pre-check self.now_playing.next to avoid that.

Raises:

Type	Description
`CommandError`	If Home Assistant rejects the service call (e.g. the player does not support skip-next).
`HTTPError`	If the REST call returns a non-2xx response.
`TimeoutError`	If the call exceeds the configured request timeout.
`ConnectionClosedError`	If the WebSocket disconnects mid-call.

Source code in src/haclient/domains/media_player.py

async def next(self) -> None:
    """Skip to the next track.

    Invokes the ``media_player.media_next_track`` Home Assistant
    service. This method does **not** consult
    `NowPlaying.next`; calls to players that do not advertise
    skip-next support will surface as `CommandError`. Pre-check
    ``self.now_playing.next`` to avoid that.

    Raises
    ------
    CommandError
        If Home Assistant rejects the service call (e.g. the player
        does not support skip-next).
    HTTPError
        If the REST call returns a non-2xx response.
    TimeoutError
        If the call exceeds the configured request timeout.
    ConnectionClosedError
        If the WebSocket disconnects mid-call.
    """
    await self._call_service("media_next_track")

previous `async`

previous() -> None

Skip to the previous track.

Invokes the media_player.media_previous_track Home Assistant service. This method does not consult NowPlaying.previous; calls to players that do not advertise skip-previous support will surface as CommandError. Pre-check self.now_playing.previous to avoid that.

Raises:

Type	Description
`CommandError`	If Home Assistant rejects the service call.
`HTTPError`	If the REST call returns a non-2xx response.
`TimeoutError`	If the call exceeds the configured request timeout.
`ConnectionClosedError`	If the WebSocket disconnects mid-call.

Source code in src/haclient/domains/media_player.py

async def previous(self) -> None:
    """Skip to the previous track.

    Invokes the ``media_player.media_previous_track`` Home Assistant
    service. This method does **not** consult
    `NowPlaying.previous`; calls to players that do not advertise
    skip-previous support will surface as `CommandError`. Pre-check
    ``self.now_playing.previous`` to avoid that.

    Raises
    ------
    CommandError
        If Home Assistant rejects the service call.
    HTTPError
        If the REST call returns a non-2xx response.
    TimeoutError
        If the call exceeds the configured request timeout.
    ConnectionClosedError
        If the WebSocket disconnects mid-call.
    """
    await self._call_service("media_previous_track")

set_volume `async`

set_volume(level: float) -> None

Set the volume level.

Parameters:

Name	Type	Description	Default
`level`	`float`	Volume level between `0.0` and `1.0`.	required

Raises:

Type	Description
`ValueError`	If level is outside the `[0.0, 1.0]` range.

Source code in src/haclient/domains/media_player.py

async def set_volume(self, level: float) -> None:
    """Set the volume level.

    Parameters
    ----------
    level : float
        Volume level between ``0.0`` and ``1.0``.

    Raises
    ------
    ValueError
        If *level* is outside the ``[0.0, 1.0]`` range.
    """
    if not 0.0 <= level <= 1.0:
        raise ValueError("Volume level must be between 0.0 and 1.0")
    await self._call_service("volume_set", {"volume_level": float(level)})

mute `async`

mute(muted: bool = True) -> None

Mute or unmute the media player.

Parameters:

Name	Type	Description	Default
`muted`	`bool`	`True` to mute (default), `False` to unmute.	`True`

Source code in src/haclient/domains/media_player.py

async def mute(self, muted: bool = True) -> None:
    """Mute or unmute the media player.

    Parameters
    ----------
    muted : bool, optional
        ``True`` to mute (default), ``False`` to unmute.
    """
    await self._call_service("volume_mute", {"is_volume_muted": bool(muted)})

power_on `async`

power_on() -> None

Power the media player on.

Invokes the media_player.turn_on Home Assistant service.

Raises:

Type	Description
`CommandError`	If Home Assistant rejects the service call.
`HTTPError`	If the REST call returns a non-2xx response.
`TimeoutError`	If the call exceeds the configured request timeout.
`ConnectionClosedError`	If the WebSocket disconnects mid-call.

Source code in src/haclient/domains/media_player.py

async def power_on(self) -> None:
    """Power the media player on.

    Invokes the ``media_player.turn_on`` Home Assistant service.

    Raises
    ------
    CommandError
        If Home Assistant rejects the service call.
    HTTPError
        If the REST call returns a non-2xx response.
    TimeoutError
        If the call exceeds the configured request timeout.
    ConnectionClosedError
        If the WebSocket disconnects mid-call.
    """
    await self._call_service("turn_on")

power_off `async`

power_off() -> None

Power the media player off.

Invokes the media_player.turn_off Home Assistant service.

Raises:

Type	Description
`CommandError`	If Home Assistant rejects the service call.
`HTTPError`	If the REST call returns a non-2xx response.
`TimeoutError`	If the call exceeds the configured request timeout.
`ConnectionClosedError`	If the WebSocket disconnects mid-call.

Source code in src/haclient/domains/media_player.py

async def power_off(self) -> None:
    """Power the media player off.

    Invokes the ``media_player.turn_off`` Home Assistant service.

    Raises
    ------
    CommandError
        If Home Assistant rejects the service call.
    HTTPError
        If the REST call returns a non-2xx response.
    TimeoutError
        If the call exceeds the configured request timeout.
    ConnectionClosedError
        If the WebSocket disconnects mid-call.
    """
    await self._call_service("turn_off")

select_source `async`

select_source(source: str) -> None

Select an input source.

Parameters:

Name	Type	Description	Default
`source`	`str`	Name of the input source. Must be one of the values reported in the entity's `source_list` attribute.	required

Source code in src/haclient/domains/media_player.py

async def select_source(self, source: str) -> None:
    """Select an input source.

    Parameters
    ----------
    source : str
        Name of the input source. Must be one of the values reported
        in the entity's ``source_list`` attribute.
    """
    await self._call_service("select_source", {"source": source})

play_media `async`

play_media(media_content_type: str, media_content_id: str, **extra: Any) -> None

Play a specific media item.

Parameters:

Name	Type	Description	Default
`media_content_type`	`str`	Media content type identifier (e.g. `"music"`).	required
`media_content_id`	`str`	Media content id understood by the underlying integration.	required
`**extra`	`Any`	Additional fields forwarded verbatim to Home Assistant (e.g. `enqueue`, `announce`).	`{}`

Source code in src/haclient/domains/media_player.py

async def play_media(
    self,
    media_content_type: str,
    media_content_id: str,
    **extra: Any,
) -> None:
    """Play a specific media item.

    Parameters
    ----------
    media_content_type : str
        Media content type identifier (e.g. ``"music"``).
    media_content_id : str
        Media content id understood by the underlying integration.
    **extra : Any
        Additional fields forwarded verbatim to Home Assistant
        (e.g. ``enqueue``, ``announce``).
    """
    data: dict[str, Any] = {
        "media_content_type": media_content_type,
        "media_content_id": media_content_id,
        **extra,
    }
    await self._call_service("play_media", data)

browse_media `async`

browse_media(media_content_type: str | None = None, media_content_id: str | None = None) -> dict[str, Any]

Issue a single media_player/browse_media WebSocket command.

Parameters:

Name	Type	Description	Default
`media_content_type`	`str or None`	Restrict browsing to this content type. `None` returns the root browse node.	`None`
`media_content_id`	`str or None`	Restrict browsing to this content id. `None` returns the root browse node.	`None`

Returns:

Type	Description
`dict`	The raw result dictionary from Home Assistant.

Raises:

Type	Description
`HAClientError`	If the command fails or returns an unexpected response.

Source code in src/haclient/domains/media_player.py

async def browse_media(
    self,
    media_content_type: str | None = None,
    media_content_id: str | None = None,
) -> dict[str, Any]:
    """Issue a single ``media_player/browse_media`` WebSocket command.

    Parameters
    ----------
    media_content_type : str or None, optional
        Restrict browsing to this content type. ``None`` returns the
        root browse node.
    media_content_id : str or None, optional
        Restrict browsing to this content id. ``None`` returns the
        root browse node.

    Returns
    -------
    dict
        The raw result dictionary from Home Assistant.

    Raises
    ------
    HAClientError
        If the command fails or returns an unexpected response.
    """
    payload: dict[str, Any] = {
        "type": "media_player/browse_media",
        "entity_id": self.entity_id,
    }
    if media_content_type is not None:
        payload["media_content_type"] = media_content_type
    if media_content_id is not None:
        payload["media_content_id"] = media_content_id
    result = await self._services.ws.send_command(payload)
    if not isinstance(result, dict):
        raise HAClientError("Unexpected browse_media response")
    return result

favorites `async`

favorites(*, max_depth: int = _MAX_BROWSE_DEPTH, max_nodes: int = _MAX_BROWSE_NODES) -> list[FavoriteItem]

Return a flattened list of playable items in the media tree.

Parameters:

Name	Type	Description	Default
`max_depth`	`int`	Maximum recursion depth.	`_MAX_BROWSE_DEPTH`
`max_nodes`	`int`	Hard upper bound on total nodes visited.	`_MAX_BROWSE_NODES`

Returns:

Type	Description
`list of FavoriteItem`	The discovered playable items.

Source code in src/haclient/domains/media_player.py

async def favorites(
    self,
    *,
    max_depth: int = _MAX_BROWSE_DEPTH,
    max_nodes: int = _MAX_BROWSE_NODES,
) -> list[FavoriteItem]:
    """Return a flattened list of playable items in the media tree.

    Parameters
    ----------
    max_depth : int, optional
        Maximum recursion depth.
    max_nodes : int, optional
        Hard upper bound on total nodes visited.

    Returns
    -------
    list of FavoriteItem
        The discovered playable items.
    """
    try:
        root = await self.browse_media()
    except CommandError as err:
        _LOGGER.debug("browse_media unsupported for %s: %s", self.entity_id, err)
        return []
    except HAClientError as err:
        _LOGGER.debug("browse_media failed for %s: %s", self.entity_id, err)
        return []

    collected: list[FavoriteItem] = []
    seen: set[tuple[str, str]] = set()
    node_count = 0

    async def walk(node: dict[str, Any], depth: int, category: str | None) -> None:
        """Recurse through *node*'s children, collecting playable items.

        Updates *collected*, *seen*, and *node_count* from the
        enclosing scope. Honours the *max_depth* and *max_nodes*
        bounds and silently skips children that are missing
        playback identifiers.
        """
        nonlocal node_count
        if node_count >= max_nodes:
            return
        node_count += 1

        children = node.get("children")
        if not isinstance(children, list):
            return

        for child in children:
            if not isinstance(child, dict):
                continue
            content_id = child.get("media_content_id")
            content_type = child.get("media_content_type")
            title = child.get("title") or child.get("name") or ""
            can_play = bool(child.get("can_play"))
            can_expand = bool(child.get("can_expand"))
            thumbnail = child.get("thumbnail")
            media_class = child.get("media_class")

            if can_play and isinstance(content_id, str) and isinstance(content_type, str):
                key = (content_type, content_id)
                if key not in seen:
                    seen.add(key)
                    collected.append(
                        FavoriteItem(
                            title=str(title),
                            media_content_id=content_id,
                            media_content_type=content_type,
                            player=self,
                            thumbnail=(thumbnail if isinstance(thumbnail, str) else None),
                            category=category
                            or (media_class if isinstance(media_class, str) else None),
                            media_class=(media_class if isinstance(media_class, str) else None),
                        )
                    )

            if (
                can_expand
                and depth + 1 < max_depth
                and isinstance(content_id, str)
                and isinstance(content_type, str)
            ):
                try:
                    sub = await self.browse_media(content_type, content_id)
                except (CommandError, HAClientError) as err:
                    _LOGGER.debug(
                        "browse_media sublevel failed (%s/%s): %s",
                        content_type,
                        content_id,
                        err,
                    )
                    continue
                except asyncio.CancelledError:
                    raise
                child_category = str(title) if title else category
                await walk(sub, depth + 1, child_category)

    await walk(root, 0, None)
    return collected

Media Player

media_player

SPEC module-attribute

NowPlaying dataclass

FavoriteItem

play async

MediaPlayer

is_playing property

is_paused property

is_muted property

volume_level property

now_playing property

__init__

on_volume_change

on_mute_change

on_media_change

on_play

on_pause

on_stop

remove_granular_listener

play async

pause async

play_pause async

stop async

next async

previous async

set_volume async

mute async

power_on async

power_off async

select_source async

play_media async

browse_media async

favorites async

SPEC `module-attribute`

NowPlaying `dataclass`

play `async`

is_playing `property`

is_paused `property`

is_muted `property`

volume_level `property`

now_playing `property`

init

play `async`

pause `async`

play_pause `async`

stop `async`

next `async`

previous `async`

set_volume `async`

mute `async`

power_on `async`

power_off `async`

select_source `async`

play_media `async`

browse_media `async`

favorites `async`