diff --git a/ANDROID-REVIEW/android-missing-functionality.md b/ANDROID-REVIEW/android-missing-functionality.md
index d382164..d5e8ccd 100644
--- a/ANDROID-REVIEW/android-missing-functionality.md
+++ b/ANDROID-REVIEW/android-missing-functionality.md
@@ -45,7 +45,7 @@ Python receiver engine mirroring that pattern.**
| LED transports (network/USB-serial/BLE) | ✅ | ✅ (USB via Android driver, BLE via Android bridge) | No |
| System metrics | psutil | ✅ CPU/RAM/battery/thermal via `/proc`, `/sys` (`AndroidMetricsProvider`) | No |
| **Audio capture** | WASAPI / Sounddevice | ❌ no PortAudio | **Yes** |
-| **Notification capture** | WinRT / D-Bus | ❌ listener only Win/Linux | **Yes** |
+| Notification capture | WinRT / D-Bus | ✅ NotificationListenerService → `push_notification()` | No (implemented) |
| Webcam capture | OpenCV | ❌ no OpenCV wheel | Yes (niche) |
| GPU monitoring | NVML | ❌ no NVIDIA GPU | Marginal |
| Capture from *another* Android phone | scrcpy/ADB | ❌ | Skip (redundant) |
@@ -70,7 +70,7 @@ Python receiver engine mirroring that pattern.**
media and the device's own audio. Root mode (no MediaProjection) → mic-only.
- 📄 **See `android-audio-capture-plan.md`** for the full implementation plan.
-### 🔔 Notification capture — **FEASIBLE, HIGH VALUE** ⭐ (planned)
+### 🔔 Notification capture — **IMPLEMENTED** ✅ (shipped)
- **Android is the *best* platform for this:** `NotificationListenerService` is the native,
event-push mechanism (no polling).
@@ -82,8 +82,13 @@ Python receiver engine mirroring that pattern.**
- **Permission:** user enables "Notification access" in Settings (`ACTION_NOTIFICATION_LISTENER_SETTINGS`);
no runtime-permission popup.
- **Effort:** moderate. **Value:** high.
-- 📄 **Plan approved & detailed** — see `C:\Users\Alexei\.claude\plans\deep-enchanting-muffin.md`
- (app-name parity; prompt-once permission UX).
+- ✅ **Implemented** on branch `feature/android-notification-capture`: a push-based
+ `_AndroidBackend` + module-level `push_notification()` in `os_notification_listener.py`,
+ a Kotlin `LedGrabNotificationListener` (NLS), and prompt-once permission UX. App-name
+ parity — only the resolved app label crosses the JNI boundary, never the notification
+ title/body. ⚠️ App labels can differ across OSes (Windows `display_name` / Linux D-Bus
+ `app_name` / Android `getApplicationLabel`), so desktop-configured per-app colors/filters
+ may need re-matching on Android.
### 📷 Webcam capture — **FEASIBLE, LOW VALUE**
@@ -128,7 +133,7 @@ Python receiver engine mirroring that pattern.**
| Priority | Feature | Effort | Value | New Python deps | Status |
| -------- | ------- | ------ | ----- | --------------- | ------ |
-| 1 | Notification capture | Moderate | High | None | **Plan approved** |
+| 1 | Notification capture | Moderate | High | None | **✅ Implemented** |
| 2 | Audio capture | Moderate | High | None | **Plan written** (this folder) |
| 3 | Automation: foreground-app condition | Moderate | Moderate | None | Idea |
| 4 | Webcam capture (CameraX) | Moderate | Low | None | Idea |
diff --git a/README.md b/README.md
index f2c7ef2..08135ed 100644
--- a/README.md
+++ b/README.md
@@ -105,16 +105,17 @@ LedGrab runs as a desktop / server application:
### Feature support by OS
-| Feature | Windows | Linux / macOS |
-| ------- | ------- | ------------- |
-| Screen capture | DXCam, BetterCam, WGC, MSS | MSS |
-| Webcam capture | OpenCV (DirectShow) | OpenCV (V4L2) |
-| Audio capture | WASAPI, Sounddevice | Sounddevice (PulseAudio/PipeWire) |
-| GPU monitoring | NVIDIA (nvidia-ml-py) | NVIDIA (nvidia-ml-py) |
-| Capture from Android phone | scrcpy (ADB) | scrcpy (ADB) |
-| Notification capture | WinRT | dbus (Linux) |
-| Monitor names | Friendly names (WMI) | Generic ("Display 0") |
-| Automation: window/process conditions | Supported | Partial |
+| Feature | Windows | Linux / macOS | Android TV (experimental) |
+| ------- | ------- | ------------- | ------------------------- |
+| Screen capture | DXCam, BetterCam, WGC, MSS | MSS | MediaProjection; root `screenrecord` (rooted devices) |
+| Webcam capture | OpenCV (DirectShow) | OpenCV (V4L2) | — (no OpenCV wheel) |
+| Audio capture | WASAPI, Sounddevice | Sounddevice (PulseAudio/PipeWire) | AudioPlaybackCapture (API 29+) |
+| GPU monitoring | NVIDIA (nvidia-ml-py) | NVIDIA (nvidia-ml-py) | — (CPU/RAM/battery/thermal via `/proc`) |
+| Capture from Android phone | scrcpy (ADB) | scrcpy (ADB) | — (captures its own screen instead) |
+| Notification capture | WinRT | dbus (Linux) | NotificationListenerService |
+| Monitor names | Friendly names (WMI) | Generic ("Display 0") | Single built-in display |
+| LED transports | Network, USB-serial, BLE | Network, USB-serial, BLE | Network, USB-serial (Android driver), BLE (Android bridge) |
+| Automation: window/process conditions | Supported | Partial | — |
## Requirements
diff --git a/android/app/src/main/AndroidManifest.xml b/android/app/src/main/AndroidManifest.xml
index bb74f08..05778b4 100644
--- a/android/app/src/main/AndroidManifest.xml
+++ b/android/app/src/main/AndroidManifest.xml
@@ -110,6 +110,23 @@
android:value="Root-mode screen capture for ambient LED sync. Uses /system/bin/screenrecord on rooted devices to avoid MediaProjection's persistent capture indicator overlay, which is required for the always-on ambient-lighting use case." />
+
+
+
+
+
+
+
+
+
Отображается, пока LedGrab захватывает экран.
LedGrab работает
Веб-интерфейс: %1$s
+ Захват уведомлений LedGrab
+ Разрешить доступ к уведомлениям
diff --git a/android/app/src/main/res/values-zh/strings.xml b/android/app/src/main/res/values-zh/strings.xml
index c7ffc32..719093f 100644
--- a/android/app/src/main/res/values-zh/strings.xml
+++ b/android/app/src/main/res/values-zh/strings.xml
@@ -25,4 +25,6 @@
LedGrab 捕获屏幕时显示。
LedGrab 运行中
Web界面:%1$s
+ LedGrab 通知捕获
+ 授予通知访问权限
diff --git a/android/app/src/main/res/values/strings.xml b/android/app/src/main/res/values/strings.xml
index 6741f73..83e3187 100644
--- a/android/app/src/main/res/values/strings.xml
+++ b/android/app/src/main/res/values/strings.xml
@@ -25,4 +25,6 @@
Shows while LedGrab is capturing the screen.
LedGrab Running
Web UI: %1$s
+ LedGrab notification capture
+ Grant notification access
diff --git a/server/src/ledgrab/core/processing/os_notification_listener.py b/server/src/ledgrab/core/processing/os_notification_listener.py
index 27f1280..bae2b26 100644
--- a/server/src/ledgrab/core/processing/os_notification_listener.py
+++ b/server/src/ledgrab/core/processing/os_notification_listener.py
@@ -8,6 +8,8 @@ Supported platforms:
- **Windows**: polls toast notifications via winrt UserNotificationListener
(falls back to winsdk if winrt packages are not installed)
- **Linux**: monitors org.freedesktop.Notifications via D-Bus (dbus-next)
+- **Android**: receives notifications pushed from a Kotlin NotificationListenerService
+ via Chaquopy (push-based; see push_notification() and _AndroidBackend)
"""
import asyncio
@@ -17,9 +19,10 @@ import platform
import threading
import time
from pathlib import Path
-from typing import Dict, List, Optional, Set
+from typing import Callable, Dict, List, Optional, Set
from ledgrab.utils import get_logger
+from ledgrab.utils.platform import is_linux
logger = get_logger(__name__)
@@ -30,15 +33,71 @@ _HISTORY_MAX = 50
# Module-level singleton for dependency access
_instance: Optional["OsNotificationListener"] = None
+# Push target for the Android backend — set by _AndroidBackend.start(), read by
+# push_notification(). None when the Android backend isn't running (desktop / server down).
+_android_target: Callable[[str | None], None] | None = None
+
def get_os_notification_listener() -> Optional["OsNotificationListener"]:
"""Return the global OsNotificationListener instance (or None)."""
return _instance
+def push_notification(app_name: str | None) -> None:
+ """Receive an Android notification pushed from Kotlin via Chaquopy.
+
+ Called by the LedGrabNotificationListener service through
+ ``Python.getInstance().getModule(...).callAttr("push_notification", label)``.
+ Routes the posting app's display label into the active listener's
+ ``_on_new_notification`` handler. No-op when the Android backend isn't running,
+ so a notification arriving before the server is ready (or on desktop) is safely
+ ignored.
+ """
+ # Snapshot into a local first: stop() may null _android_target concurrently, but an
+ # in-flight push then still completes against the prior callback. Do NOT collapse this
+ # into `if _android_target is not None: _android_target(...)` — that reintroduces a
+ # TOCTOU None-deref race.
+ cb = _android_target
+ if cb is None:
+ return
+ try:
+ cb(app_name)
+ except Exception as exc: # never let a JNI-side call crash the bound service
+ logger.warning("push_notification callback error: %s", exc)
+
+
# ── Platform backends ──────────────────────────────────────────────────
+class _AndroidBackend:
+ """Push-based backend — notifications arrive from Kotlin via push_notification().
+
+ Unlike the Windows/Linux backends (which poll or eavesdrop on a thread), Android
+ notifications are delivered by a Kotlin NotificationListenerService across the
+ Chaquopy JNI boundary into the module-level push_notification() receiver, so
+ start()/stop() simply register/clear the receiver target.
+ """
+
+ def __init__(self, on_notification):
+ self._on_notification = on_notification
+
+ @staticmethod
+ def probe() -> bool:
+ """Return True when running on Android (Chaquopy)."""
+ from ledgrab.utils.platform import is_android
+
+ return is_android()
+
+ def start(self) -> None:
+ global _android_target
+ _android_target = self._on_notification
+ logger.info("OS notification listener: Android backend active")
+
+ def stop(self) -> None:
+ global _android_target
+ _android_target = None
+
+
def _import_winrt_notifications():
"""Try to import WinRT notification APIs: winrt first, then winsdk fallback.
@@ -193,7 +252,9 @@ class _LinuxBackend:
@staticmethod
def probe() -> bool:
"""Return True if this backend can run on the current system."""
- if platform.system() != "Linux":
+ # is_linux() excludes Android, which also reports platform.system() == "Linux"
+ # but has no D-Bus session — defense-in-depth beyond probe ordering.
+ if not is_linux():
return False
try:
import dbus_next # noqa: F401
@@ -312,8 +373,9 @@ class OsNotificationListener:
global _instance
_instance = self
- # Try platform backends in order
- for backend_cls in (_WindowsBackend, _LinuxBackend):
+ # Try platform backends in order (Android first — it reports platform.system()
+ # == "Linux", so probing it ahead of _LinuxBackend is the robust ordering).
+ for backend_cls in (_AndroidBackend, _WindowsBackend, _LinuxBackend):
if backend_cls.probe():
self._backend = backend_cls(on_notification=self._on_new_notification)
self._backend.start()
diff --git a/server/tests/core/processing/test_os_notification_listener_android.py b/server/tests/core/processing/test_os_notification_listener_android.py
new file mode 100644
index 0000000..f539f1a
--- /dev/null
+++ b/server/tests/core/processing/test_os_notification_listener_android.py
@@ -0,0 +1,237 @@
+"""Tests for the Android push-based notification backend.
+
+These run on desktop CI (no Android device needed): ``is_android`` is
+monkeypatched and the app label is pushed directly into the module-level
+``push_notification`` receiver, exactly as the Kotlin
+``NotificationListenerService`` would across the Chaquopy bridge.
+
+Isolation (critical): the listener keeps process-global state
+(``_android_target``, ``_instance``) and persists history to a hardcoded
+``data/notification_history.json``. Every test resets those globals and
+repoints ``_HISTORY_FILE`` to ``tmp_path`` so the suite never leaks state
+between tests or clobbers the real repo data file.
+"""
+
+from datetime import datetime, timezone
+
+import pytest
+
+import ledgrab.core.processing.os_notification_listener as nl
+from ledgrab.storage.color_strip_source import NotificationColorStripSource
+
+PLATFORM_MOD = "ledgrab.utils.platform"
+
+
+# ---------------------------------------------------------------------------
+# Test doubles
+# ---------------------------------------------------------------------------
+
+
+class _FakeStream:
+ """Stub NotificationColorStripStream — records fire() calls."""
+
+ def __init__(self, accept: bool = True):
+ self._accept = accept
+ self.fired_with: list = []
+
+ def fire(self, app_name=None) -> bool:
+ self.fired_with.append(app_name)
+ return self._accept
+
+
+class _FakeStore:
+ def __init__(self, sources):
+ self._sources = sources
+
+ def get_all_sources(self):
+ return list(self._sources)
+
+
+class _FakeStreamManager:
+ def __init__(self, streams):
+ self._streams = streams
+
+ def get_streams_by_source_id(self, source_id):
+ return list(self._streams)
+
+
+def _notif_source(
+ *, source_id: str = "css_test", os_listener: bool = True
+) -> NotificationColorStripSource:
+ now = datetime.now(timezone.utc)
+ return NotificationColorStripSource.create_from_kwargs(
+ id=source_id,
+ name="Test Notification Source",
+ source_type="notification",
+ created_at=now,
+ updated_at=now,
+ os_listener=os_listener,
+ )
+
+
+# ---------------------------------------------------------------------------
+# Fixtures — module-global + disk isolation
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def nl_mod(monkeypatch, tmp_path):
+ """Reset module globals and repoint the history file to tmp_path.
+
+ ``monkeypatch.setattr`` auto-restores originals on teardown, so even though
+ ``start()``/``stop()`` rebind ``_android_target`` and ``_instance`` during a
+ test, the globals are returned to their pre-test values afterward — no
+ cross-test leakage and no write to the real repo ``data/`` dir.
+ """
+ monkeypatch.setattr(nl, "_android_target", None)
+ monkeypatch.setattr(nl, "_instance", None)
+ monkeypatch.setattr(nl, "_HISTORY_FILE", tmp_path / "notification_history.json")
+ return nl
+
+
+# ---------------------------------------------------------------------------
+# _AndroidBackend.probe()
+# ---------------------------------------------------------------------------
+
+
+def test_probe_true_on_android(nl_mod, monkeypatch):
+ monkeypatch.setattr(f"{PLATFORM_MOD}.is_android", lambda: True)
+ assert nl_mod._AndroidBackend.probe() is True
+
+
+def test_probe_false_on_desktop(nl_mod, monkeypatch):
+ monkeypatch.setattr(f"{PLATFORM_MOD}.is_android", lambda: False)
+ assert nl_mod._AndroidBackend.probe() is False
+
+
+# ---------------------------------------------------------------------------
+# push_notification() routing contract
+# ---------------------------------------------------------------------------
+
+
+def test_push_is_noop_before_start(nl_mod):
+ # _android_target is None → no callback, no exception.
+ nl_mod.push_notification("Telegram") # must not raise
+
+
+def test_push_routes_after_start_and_stops_after_stop(nl_mod):
+ received: list = []
+ backend = nl_mod._AndroidBackend(on_notification=received.append)
+
+ backend.start()
+ nl_mod.push_notification("Telegram")
+ assert received == ["Telegram"]
+
+ backend.stop()
+ nl_mod.push_notification("Signal") # no-op after stop
+ assert received == ["Telegram"]
+
+
+def test_push_swallows_callback_exception(nl_mod):
+ def boom(_app):
+ raise RuntimeError("callback exploded")
+
+ nl_mod._AndroidBackend(on_notification=boom).start()
+ # JNI entry point must never propagate — would crash the bound service.
+ nl_mod.push_notification("X")
+
+
+# ---------------------------------------------------------------------------
+# Integration — start() selects Android, push fires the stream + records history
+# ---------------------------------------------------------------------------
+
+
+def test_android_selected_push_fires_stream_and_records_history(nl_mod, monkeypatch, tmp_path):
+ monkeypatch.setattr(f"{PLATFORM_MOD}.is_android", lambda: True)
+ stream = _FakeStream(accept=True)
+ listener = nl_mod.OsNotificationListener(
+ _FakeStore([_notif_source(os_listener=True)]),
+ _FakeStreamManager([stream]),
+ )
+
+ listener.start()
+ assert listener.available is True # flips True on backend selection, not on push
+
+ nl_mod.push_notification("Telegram")
+
+ assert stream.fired_with == ["Telegram"]
+ assert listener.recent_history[0]["app"] == "Telegram"
+ assert listener.recent_history[0]["fired"] == 1
+ # history written under tmp_path — never the repo data/ dir
+ assert nl_mod._HISTORY_FILE.exists()
+ assert nl_mod._HISTORY_FILE.parent == tmp_path
+
+ listener.stop()
+
+
+def test_push_with_none_app_name_is_recorded(nl_mod, monkeypatch):
+ # The Windows (_extract_app_name) and Linux D-Bus paths can yield None;
+ # the Android path falls back to the package name, but None must still be
+ # handled end-to-end without raising.
+ monkeypatch.setattr(f"{PLATFORM_MOD}.is_android", lambda: True)
+ stream = _FakeStream(accept=True)
+ listener = nl_mod.OsNotificationListener(
+ _FakeStore([_notif_source(os_listener=True)]),
+ _FakeStreamManager([stream]),
+ )
+
+ listener.start()
+ nl_mod.push_notification(None)
+
+ assert stream.fired_with == [None]
+ assert listener.recent_history[0]["app"] is None
+ listener.stop()
+
+
+def test_get_os_notification_listener_tracks_started_instance(nl_mod, monkeypatch):
+ monkeypatch.setattr(f"{PLATFORM_MOD}.is_android", lambda: True)
+ assert nl_mod.get_os_notification_listener() is None
+
+ listener = nl_mod.OsNotificationListener(_FakeStore([]), _FakeStreamManager([]))
+ listener.start()
+ assert nl_mod.get_os_notification_listener() is listener
+ listener.stop()
+
+
+def test_source_with_os_listener_off_does_not_fire(nl_mod, monkeypatch):
+ monkeypatch.setattr(f"{PLATFORM_MOD}.is_android", lambda: True)
+ stream = _FakeStream()
+ listener = nl_mod.OsNotificationListener(
+ _FakeStore([_notif_source(os_listener=False)]),
+ _FakeStreamManager([stream]),
+ )
+
+ listener.start()
+ nl_mod.push_notification("Telegram")
+
+ assert stream.fired_with == [] # os_listener=False → skipped
+ listener.stop()
+
+
+# ---------------------------------------------------------------------------
+# Desktop regression — the probe-order change must not alter desktop selection
+# ---------------------------------------------------------------------------
+
+
+def test_android_probe_false_on_real_desktop(nl_mod, monkeypatch):
+ # With is_android() False, the new first-in-tuple backend must not be selectable.
+ monkeypatch.setattr(f"{PLATFORM_MOD}.is_android", lambda: False)
+ assert nl_mod._AndroidBackend.probe() is False
+
+
+def test_desktop_selection_unchanged_windows_wins(nl_mod, monkeypatch):
+ # Deterministically control probes and stub start() so no real polling thread spawns.
+ # Order under test is (_AndroidBackend, _WindowsBackend, _LinuxBackend): Android skipped,
+ # Windows is the first True → it must be the selected backend, exactly as before.
+ monkeypatch.setattr(nl_mod._AndroidBackend, "probe", staticmethod(lambda: False))
+ monkeypatch.setattr(nl_mod._WindowsBackend, "probe", staticmethod(lambda: True))
+ monkeypatch.setattr(nl_mod._LinuxBackend, "probe", staticmethod(lambda: False))
+ started: list = []
+ monkeypatch.setattr(nl_mod._WindowsBackend, "start", lambda self: started.append("win"))
+
+ listener = nl_mod.OsNotificationListener(_FakeStore([]), _FakeStreamManager([]))
+ listener.start()
+
+ assert listener.available is True
+ assert isinstance(listener._backend, nl_mod._WindowsBackend)
+ assert started == ["win"]