feat(config_flow,strings): CONF_LOGCAT_EVENTS options toggle (default off, stability-report wording) (P5-3)

custom_components/nvidia_shield/config_flow.py

@@ -25,6 +25,7 @@ from .const import (
     CONF_CHANNEL_DIGIT_DELAY,
     CONF_CHANNEL_SEND_ENTER,
     CONF_JELLYFIN_ENTITY,
+    CONF_LOGCAT_EVENTS,
     CONF_PAIRING_CODE,
     CONF_POWER_ENTITY,
     CONF_PROTOCOL_ENABLED,
@@ -587,6 +588,11 @@ class NvidiaShieldOptionsFlow(config_entries.OptionsFlow):
         current_enter = self._options.get(CONF_CHANNEL_SEND_ENTER, False)
         schema_dict[vol.Optional(CONF_CHANNEL_SEND_ENTER, default=current_enter)] = bool
 
+        # P5-3: the live events push is opt-in (default off; the user
+        # tests stability on their device and reports)
+        current_logcat = self._options.get(CONF_LOGCAT_EVENTS, False)
+        schema_dict[vol.Optional(CONF_LOGCAT_EVENTS, default=current_logcat)] = bool
+
         # P4-2a: the raw adb_command service is opt-in (default off)
         current_flag = self._options.get(CONF_ADB_COMMAND_ENABLED, False)
         schema_dict[vol.Optional(CONF_ADB_COMMAND_ENABLED, default=current_flag)] = bool

custom_components/nvidia_shield/ha_adb_common/__init__.py

@@ -13,6 +13,7 @@ from .const import (
 from .foreground import parse_foreground
 from .intent import PlaybackIntent
 from .keys import ensure_adb_key
+from .logcat_events import foreground_from_logcat
 from .media_session import (
     MediaSessionInfo,
     parse_all_media_sessions,
@@ -36,6 +37,7 @@ __all__ = [
     "parse_application_label",
     "parse_audio_playback",
     "parse_foreground",
+    "foreground_from_logcat",
     "parse_wakefulness",
     "parse_media_session",
     "running_apps_from_sessions",

custom_components/nvidia_shield/ha_adb_common/logcat_events.py

@@ -0,0 +1,79 @@
+"""Live events push (P5-3 / C-STREAM): logcat foreground trigger.
+
+Parses Android logcat's ActivityTaskManager ``START`` records and extracts
+the ``cmp=package`` part so the connection manager can stamp the foreground
+AppState instantly (sub-heartbeat latency). Logcat carries lifecycle only —
+no PlaybackState — so the hybrid split (ADB stays the playback/scene state
+source) is untouched: the stream never reads playback state.
+"""
+
+import re
+
+_LOG_RECORD_RE = re.compile(r"^\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}\.\d+")
+_CMP_RE = re.compile(r"cmp=([\w.]+)(?:/[^\s,}]+)?")
+_START_HINT_RE = re.compile(r"ActivityTaskManager:.*?START\b")
+
+
+def start_records(lines: list[str]) -> list[str]:
+    """Split raw logcat lines into START-record buffers (wrapped lines joined).
+
+    A record opens at a line that matches a log timestamp and mentions
+    ActivityTaskManager + START; subsequent lines that do NOT look like a
+    fresh log record are treated as continuations (the intent JSON can wrap
+    across lines). Anything else closes/open the next record.
+    """
+    records: list[str] = []
+    current: list[str] = []
+    for line in lines:
+        if _LOG_RECORD_RE.match(line) and _START_HINT_RE.search(line):
+            if current:
+                records.append("\n".join(current))
+            current = [line]
+        elif current and not _LOG_RECORD_RE.match(line):
+            current.append(line)  # wrapped intent text
+        else:
+            if current:
+                records.append("\n".join(current))
+            current = []
+    if current:
+        records.append("\n".join(current))
+    return records
+
+
+def foreground_from_logcat(lines: list[str]) -> str | None:
+    """The foreground package from the LAST valid START record, or None.
+
+    Non-START lines are ignored entirely; records whose intent carries no
+    ``cmp=`` (e.g. service starts) yield nothing; a explicit ``null`` cmp
+    is skipped too. Multi-user prefixes (``START u0``/``u10``...) are
+    tolerated — the user prefix is never part of the cmp value.
+    """
+    package: str | None = None
+    for record in start_records(lines):
+        match = _CMP_RE.search(record)
+        if not match:
+            continue
+        candidate = match.group(1)
+        if candidate and candidate != "null":
+            package = candidate  # last valid START wins
+    return package
+
+
+def foreground_packages(lines: list[str]) -> list[str]:
+    """Every distinct valid START package in order (test seam)."""
+    seen: list[str] = []
+    for record in start_records(lines):
+        match = _CMP_RE.search(record)
+        if not match:
+            continue
+        candidate = match.group(1)
+        if candidate and candidate != "null" and candidate not in seen:
+            seen.append(candidate)
+    return seen
+
+
+__all__ = [
+    "foreground_from_logcat",
+    "foreground_packages",
+    "start_records",
+]