feat(media_session): parse PlaybackState position/speed/updated (uptime-ms) + parse_audio_playback section (MVP plan Phase 1, R9/R2)

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:

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("") == []