From 8f27618e3dc44ec3a8950c98073ffc4841c37aa0 Mon Sep 17 00:00:00 2001 From: James Mackay Date: Thu, 8 Oct 2026 08:38:35 +0100 Subject: [PATCH] feat(media_session): parse PlaybackState position/speed/updated (uptime-ms) + parse_audio_playback section (MVP plan Phase 1, R9/R2) --- .../ha_adb_common/media_session.py | 111 +++++++++++++++++- tests/test_media_session.py | 109 ++++++++++++++++- 2 files changed, 214 insertions(+), 6 deletions(-) diff --git a/custom_components/nvidia_shield/ha_adb_common/media_session.py b/custom_components/nvidia_shield/ha_adb_common/media_session.py index e700af4..88ecbd6 100644 --- a/custom_components/nvidia_shield/ha_adb_common/media_session.py +++ b/custom_components/nvidia_shield/ha_adb_common/media_session.py @@ -4,12 +4,20 @@ Parses `dumpsys media_session` output from ADB to extract now-playing information from any app using Android's MediaSession API. """ +import re from dataclasses import dataclass @dataclass class MediaSessionInfo: - """Parsed media session data from Android.""" + """Parsed media session data from Android. + + Playback position fields come from the session's PlaybackState line: + ``position`` is the playback offset in ms as of ``updated`` — which is a + **device uptime-ms** timestamp (verified on Shield Android 11, R9), not a + wallclock value. Use :meth:`position_at_uptime` with the device's current + uptime to extrapolate the live position. + """ title: str | None = None artist: str | None = None @@ -17,6 +25,56 @@ class MediaSessionInfo: package: str | None = None playing: bool = False paused: bool = False + position: float | None = None + speed: float | None = None + updated: int | None = None + + def position_at_uptime(self, now_uptime_ms: float) -> float | None: + """Extrapolate the live playback position at device uptime ``now_uptime_ms``. + + Mirrors HA core's convention: ``position + (now - updated) * speed``. + Returns None when the session carries no position sample. + """ + if self.position is None or self.updated is None: + return None + speed = self.speed if self.speed is not None else 1.0 + return self.position + (now_uptime_ms - self.updated) * speed + + +_PLAYBACK_STATE_RE = re.compile(r"PlaybackState\s*\{([^}]*)\}") + + +def _parse_playback_state(content: str) -> dict[str, str]: + """Parse the ``key=value`` pairs inside a PlaybackState block. + + Splitting on ", " (rather than searching for ``position=`` anywhere in + the line) keeps the ``buffered position`` field from being mistaken for + the real ``position``. + """ + fields: dict[str, str] = {} + for part in content.split(", "): + key, _, value = part.partition("=") + if value: + fields[key.strip()] = value.strip() + return fields + + +def _parse_float(value: str | None) -> float | None: + if value is None: + return None + try: + return float(value) + except ValueError: + return None + + +def _parse_int(value: str | None) -> int | None: + if value is None: + return None + try: + return int(value) + except ValueError: + return None def parse_all_media_sessions(dump: str) -> list[MediaSessionInfo]: @@ -32,6 +90,9 @@ def parse_all_media_sessions(dump: str) -> list[MediaSessionInfo]: current_pkg: str | None = None in_active = False current_state = 0 # 0=unknown, 2=paused, 3=playing + current_position: float | None = None + current_speed: float | None = None + current_updated: int | None = None current_meta: MediaSessionInfo | None = None for line in dump.split("\n"): @@ -47,16 +108,25 @@ def parse_all_media_sessions(dump: str) -> list[MediaSessionInfo]: current_pkg = pkg[1].strip() in_active = False current_state = 0 + current_position = None + current_speed = None + current_updated = None current_meta = None if "active=true" in stripped: in_active = True if in_active and "state=PlaybackState" in stripped: - if "state=3" in stripped: - current_state = 3 - elif "state=2" in stripped: - current_state = 2 + match = _PLAYBACK_STATE_RE.search(stripped) + if match: + fields = _parse_playback_state(match.group(1)) + try: + current_state = int(fields.get("state", 0)) + except ValueError: + current_state = 0 + current_position = _parse_float(fields.get("position")) + current_speed = _parse_float(fields.get("speed")) + current_updated = _parse_int(fields.get("updated")) if in_active and "metadata: size=" in stripped and "description=" in stripped: desc_start = stripped.find("description=") @@ -69,6 +139,9 @@ def parse_all_media_sessions(dump: str) -> list[MediaSessionInfo]: info.package = current_pkg info.playing = current_state == 3 info.paused = current_state == 2 + info.position = current_position + info.speed = current_speed + info.updated = current_updated if len(parts) >= 1 and parts[0] not in ("null", ""): info.title = parts[0] @@ -97,6 +170,34 @@ def parse_all_media_sessions(dump: str) -> list[MediaSessionInfo]: return sessions +def parse_audio_playback(dump: str) -> list[str]: + """Parse the "Audio playback (lastly played comes first)" section. + + Returns the package names that recently played audio, in the dump's own + order (most recently played first). This is a zero-cost hint from the same + ``dumpsys media_session`` call: an app shows up here even when it publishes + no MediaSession (e.g. TiviMate, R2). Callers may filter system processes + such as ``com.google.android.apps.mediashell`` (the Chromecast receiver). + """ + if not dump: + return [] + + packages: list[str] = [] + in_audio = False + for line in dump.split("\n"): + stripped = line.strip() + if not in_audio: + in_audio = "Audio playback" in stripped + continue + marker = "packages=" + if marker not in stripped: + continue + pkg_part = stripped.split(marker, 1)[1] + for pkg in pkg_part.replace(",", " ").split(): + if pkg and pkg not in packages: + packages.append(pkg) + return packages + def parse_media_session( dump: str, foreground_package: str | None = None ) -> MediaSessionInfo | None: diff --git a/tests/test_media_session.py b/tests/test_media_session.py index 40068cc..bc6f808 100644 --- a/tests/test_media_session.py +++ b/tests/test_media_session.py @@ -6,7 +6,12 @@ capture was produced and which expectations it pins. from pathlib import Path -from ha_adb_common.media_session import parse_all_media_sessions, parse_media_session +from ha_adb_common.media_session import ( + MediaSessionInfo, + parse_all_media_sessions, + parse_audio_playback, + parse_media_session, +) FIXTURES = Path(__file__).parent / "fixtures" @@ -51,3 +56,105 @@ class TestWholphinSession: assert best is not None assert best.package == "com.github.damontecres.wholphin" assert best.playing is True + + +class TestPlaybackStatePosition: + """PlaybackState carries position (ms), speed, and uptime-ms `updated` (R9).""" + + def test_wholphin_position_fields_parsed(self): + sessions = parse_all_media_sessions(_fixture("media_session_wholphin.txt")) + assert len(sessions) == 1 + session = sessions[0] + assert session.position == 17237 + assert session.speed == 1.0 + assert session.updated == 1408392 + + def test_buffered_position_is_not_mistaken_for_position(self): + """`buffered position=70070` sits in the same block; only the real + `position=17237` may be read.""" + sessions = parse_all_media_sessions(_fixture("media_session_wholphin.txt")) + assert sessions[0].position != 70070 + + def test_position_at_uptime_extrapolates_with_speed(self): + session = MediaSessionInfo( + position=17237, speed=1.0, updated=1408392, playing=True + ) + assert session.position_at_uptime(1408392 + 1000) == 18237 + assert session.position_at_uptime(1408392) == 17237 + + def test_position_at_uptime_none_without_sample(self): + assert MediaSessionInfo().position_at_uptime(1_000_000) is None + + def test_paused_session_parses_state_and_position(self): + dump = ( + "Sessions Stack - have 1 sessions:\n" + " Some app session com.example.player/Session (userId=0)\n" + " package=com.example.player\n" + " active=true\n" + " state=PlaybackState {state=2, position=4200, buffered position=9000, " + "speed=0.0, updated=500, actions=1, custom actions=[], active item id=0, " + "error=null}\n" + " metadata: size=2, description=Song Title, Some Artist\n" + " Audio playback (lastly played comes first)\n" + " uid=10100 packages=com.example.player\n" + ) + sessions = parse_all_media_sessions(dump) + assert len(sessions) == 1 + session = sessions[0] + assert session.paused is True + assert session.playing is False + assert session.position == 4200 + assert session.speed == 0.0 + assert session.updated == 500 + assert session.title == "Song Title" + assert session.artist == "Some Artist" + + def test_stateless_playbackstate_line_ignored(self): + """A non-PlaybackState `state=` line must not crash or fabricate state.""" + dump = ( + " Some app session com.example.player/Session (userId=0)\n" + " package=com.example.player\n" + " active=true\n" + " state=null\n" + " metadata: size=1, description=Only Title\n" + ) + sessions = parse_all_media_sessions(dump) + assert len(sessions) == 1 + assert sessions[0].playing is False + assert sessions[0].paused is False + assert sessions[0].position is None + + +class TestAudioPlaybackSection: + """The "Audio playback (lastly played comes first)" section lists apps + that recently played audio even with no MediaSession (R2).""" + + def test_tivimate_shows_up_without_a_session(self): + packages = parse_audio_playback(_fixture("media_session_tivimate.txt")) + assert packages == [ + "ar.tvplayer.tv", + "com.google.android.apps.mediashell", + ] + + def test_order_is_dump_order_most_recent_first(self): + packages = parse_audio_playback(_fixture("media_session_tivimate.txt")) + assert packages[0] == "ar.tvplayer.tv" + + def test_wholphin_audio_playback(self): + assert parse_audio_playback(_fixture("media_session_wholphin.txt")) == [ + "com.github.damontecres.wholphin" + ] + + def test_lines_before_the_section_are_ignored(self): + """`... package:` listener lines precede the section and must not leak + into the audio-playback package list.""" + dump = ( + " Media key listener package:\n" + " OnMediaKeyEventDispatchedListener: added 0 listener(s)\n" + " Audio playback (lastly played comes first)\n" + " uid=10117 packages=ar.tvplayer.tv\n" + ) + assert parse_audio_playback(dump) == ["ar.tvplayer.tv"] + + def test_empty_dump(self): + assert parse_audio_playback("") == []