"""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. ``playing``/``paused``/``buffering`` are independent flags mapped from the Android PlaybackState integer: 2 = paused, 3 = playing, 6 = buffering (t55). 6 is the stock-Android buffering value and the mapping is INFERRED from the ``state=PlaybackState {state=N, ...}`` line shape, not from a captured Shield dump: no fixture in this repo contains state=6, and no real device was available. Nothing here is vendor-specific either way — every integer is read straight out of the dump. """ title: str | None = None artist: str | None = None album: str | None = None package: str | None = None playing: bool = False paused: bool = False buffering: 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 buffering, then paused, then others (t55). """ if not dump: return [] sessions: list[MediaSessionInfo] = [] current_pkg: str | None = None in_active = False # Android PlaybackState: 0=unknown, 2=paused, 3=playing, 6=buffering current_state = 0 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.buffering = current_state == 6 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 buffering (mid-playback), then paused, then # others — a buffering session outranks a paused one (t55). def sort_key(s: MediaSessionInfo) -> int: if s.playing: return 0 if s.buffering: return 1 if s.paused: return 2 return 3 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 buffering (state=6) 4. The session that is paused (state=2) 5. 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 > buffering > paused > other return sessions[0]