224 lines
7.4 KiB
Python
224 lines
7.4 KiB
Python
"""Android media session parsing.
|
|
|
|
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.
|
|
|
|
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
|
|
album: str | None = None
|
|
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]:
|
|
"""Parse `dumpsys media_session` output for ALL active media sessions.
|
|
|
|
Returns a list of every active session that has metadata or a known
|
|
playback state, sorted: playing first, then paused, then others.
|
|
"""
|
|
if not dump:
|
|
return []
|
|
|
|
sessions: 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"):
|
|
stripped = line.strip()
|
|
|
|
if "package=" in stripped:
|
|
pkg = stripped.split("package=", 1)
|
|
if len(pkg) > 1:
|
|
# New session block — save the previous one if valid
|
|
if in_active and current_meta:
|
|
current_meta.package = current_pkg
|
|
sessions.append(current_meta)
|
|
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:
|
|
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=")
|
|
if desc_start < 0:
|
|
continue
|
|
desc = stripped[desc_start + 12:]
|
|
parts = [p.strip() for p in desc.split(", ")]
|
|
|
|
info = 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]
|
|
if len(parts) >= 2 and parts[1] not in ("null", ""):
|
|
info.artist = parts[1]
|
|
if len(parts) >= 3 and parts[2] not in ("null", ""):
|
|
info.album = parts[2]
|
|
|
|
if info.title:
|
|
current_meta = info
|
|
|
|
# Don't forget the last session
|
|
if in_active and current_meta:
|
|
current_meta.package = current_pkg
|
|
sessions.append(current_meta)
|
|
|
|
# Sort: playing first, then paused, then others
|
|
def sort_key(s: MediaSessionInfo) -> int:
|
|
if s.playing:
|
|
return 0
|
|
if s.paused:
|
|
return 1
|
|
return 2
|
|
|
|
sessions.sort(key=sort_key)
|
|
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:
|
|
"""Parse `dumpsys media_session` output for the best active session.
|
|
|
|
Returns the best match from all active sessions:
|
|
1. The session whose package matches the foreground app (if provided)
|
|
2. The session that is currently playing (state=3)
|
|
3. The session that is paused (state=2)
|
|
4. Any active session with metadata
|
|
"""
|
|
sessions = parse_all_media_sessions(dump)
|
|
if not sessions:
|
|
return None
|
|
|
|
# Prefer the session matching the foreground app
|
|
if foreground_package:
|
|
for s in sessions:
|
|
if s.package == foreground_package:
|
|
return s
|
|
|
|
# Already sorted: playing > paused > other
|
|
return sessions[0]
|