feat(android): on-device OS notification capture (NotificationListenerService)
Add an Android backend to os_notification_listener.py so notifications on the experimental Android-TV build drive the existing NotificationColorStripSource LED effects (flash/pulse/sweep, per-app colors + sounds) at app-name parity with the Windows/Linux backends. A Kotlin NotificationListenerService forwards the posting app's display label across the Chaquopy JNI boundary into a new push-based _AndroidBackend + module-level push_notification() receiver; the existing color-strip pipeline, per-app colors/filters, and history endpoint are reused unchanged. - Python: _AndroidBackend (probed first), push_notification() receiver, _LinuxBackend.probe() hardened with is_linux() to exclude Android (which also reports platform.system() == "Linux"). - Android: LedGrabNotificationListener NLS — serial single-thread executor, full crash isolation around Python.getInstance(), label-only forwarding (never notification title/body), ongoing/group-summary/self-package noise filtering. Manifest service exported + gated by BIND_NOTIFICATION_LISTENER_SERVICE (no new uses-permission). - UX: prompt-once notification-access + manual "Grant notification access" button wired into the D-pad focus chain (computed from visible controls); en/ru/zh strings. - Tests: 11 isolated unit tests — module-global + tmp_path history isolation, push routing contract, callback-exception swallowing, None app-name, and a desktop-regression lock on backend selection order. - Docs: README OS-support Android column (notification + audio cells), ANDROID-REVIEW status flipped to Implemented. Zero new Python deps; no build.gradle.kts / Chaquopy pip changes.
This commit is contained in:
@@ -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 |
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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." />
|
||||
</service>
|
||||
|
||||
<!-- Notification capture — a NotificationListenerService bound by
|
||||
system_server. exported="true" is REQUIRED here (the system binds
|
||||
it cross-process) and intentionally diverges from CaptureService
|
||||
(exported="false"); access is gated by the system-held
|
||||
BIND_NOTIFICATION_LISTENER_SERVICE permission, so no new
|
||||
<uses-permission> is needed. The user grants access via
|
||||
Settings > Notification access (opened from MainActivity). -->
|
||||
<service
|
||||
android:name=".LedGrabNotificationListener"
|
||||
android:label="@string/notification_listener_label"
|
||||
android:exported="true"
|
||||
android:permission="android.permission.BIND_NOTIFICATION_LISTENER_SERVICE">
|
||||
<intent-filter>
|
||||
<action android:name="android.service.notification.NotificationListenerService" />
|
||||
</intent-filter>
|
||||
</service>
|
||||
|
||||
<!-- Autostart — fires on device boot (and package replace).
|
||||
On rooted devices, launches CaptureService directly so capture
|
||||
resumes without the user tapping Start. Unrooted devices are
|
||||
|
||||
@@ -0,0 +1,97 @@
|
||||
package com.ledgrab.android
|
||||
|
||||
import android.app.Notification
|
||||
import android.service.notification.NotificationListenerService
|
||||
import android.service.notification.StatusBarNotification
|
||||
import android.util.Log
|
||||
import com.chaquo.python.Python
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
import java.util.concurrent.Executors
|
||||
|
||||
/**
|
||||
* Captures posted OS notifications and forwards the posting app's display
|
||||
* label to the Python notification pipeline, where the existing
|
||||
* `NotificationColorStripSource` fires its one-shot LED effect.
|
||||
*
|
||||
* Direction is Kotlin -> Python via the process-global Chaquopy instance
|
||||
* (NOT a per-[CaptureService] [PythonBridge]): `system_server` binds this
|
||||
* service independently of [CaptureService], so it resolves Python itself.
|
||||
* The Python receiver (`os_notification_listener.push_notification`) is a
|
||||
* no-op whenever the server/listener isn't running, so a notification
|
||||
* arriving before — or after — a capture session is safely ignored.
|
||||
*/
|
||||
class LedGrabNotificationListener : NotificationListenerService() {
|
||||
|
||||
// Serial executor: the Python receiver does a (non-concurrency-safe) history
|
||||
// disk write and may play a sound, so pushes must not overlap. Off the main
|
||||
// looper to keep the system service responsive.
|
||||
private val pushExecutor = Executors.newSingleThreadExecutor()
|
||||
|
||||
// packageName -> resolved human-readable label. Matches the app_name the
|
||||
// Windows/Linux backends pass, so per-app colors/filters keep working.
|
||||
// Naturally bounded by the number of notification-posting apps (tens) and
|
||||
// cleared with the process — no eviction needed.
|
||||
private val labelCache = ConcurrentHashMap<String, String>()
|
||||
|
||||
override fun onNotificationPosted(sbn: StatusBarNotification?) {
|
||||
val notification = sbn ?: return
|
||||
|
||||
// The Python server (and thus the listener) only exists during a capture
|
||||
// session. isRunning is a coarse early-out — the authoritative gate is the
|
||||
// Python receiver's None-check — but it avoids needless JNI churn here.
|
||||
if (!CaptureService.isRunning) return
|
||||
|
||||
// Filter notifications that should never drive an effect:
|
||||
// - ongoing (media transport, downloads): not user-facing "alerts"
|
||||
// - group summaries: duplicate their child notifications
|
||||
// - our own foreground-service notification: would self-trigger
|
||||
if (notification.isOngoing) return
|
||||
if ((notification.notification.flags and Notification.FLAG_GROUP_SUMMARY) != 0) return
|
||||
if (notification.packageName == packageName) return
|
||||
|
||||
val label = resolveAppLabel(notification.packageName)
|
||||
|
||||
pushExecutor.execute {
|
||||
try {
|
||||
Python.getInstance()
|
||||
.getModule(PY_MODULE)
|
||||
.callAttr("push_notification", label)
|
||||
} catch (t: Throwable) {
|
||||
// Never crash a system-bound service. Python.getInstance() throws
|
||||
// IllegalStateException if Python.start() hasn't run (e.g. the
|
||||
// service was bound at boot before the app process initialized).
|
||||
// Log at debug — the label is potentially sensitive on a shared TV.
|
||||
Log.d(TAG, "push_notification failed: ${t.message}")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Resolve (and cache) a package's human-readable label; fall back to the package name. */
|
||||
private fun resolveAppLabel(pkg: String): String {
|
||||
labelCache[pkg]?.let { return it }
|
||||
val resolved = runCatching {
|
||||
val info = packageManager.getApplicationInfo(pkg, 0)
|
||||
packageManager.getApplicationLabel(info).toString()
|
||||
}.getOrDefault(pkg)
|
||||
labelCache[pkg] = resolved
|
||||
return resolved
|
||||
}
|
||||
|
||||
override fun onListenerConnected() {
|
||||
Log.i(TAG, "Notification listener connected")
|
||||
}
|
||||
|
||||
override fun onListenerDisconnected() {
|
||||
Log.i(TAG, "Notification listener disconnected")
|
||||
}
|
||||
|
||||
override fun onDestroy() {
|
||||
pushExecutor.shutdown()
|
||||
super.onDestroy()
|
||||
}
|
||||
|
||||
companion object {
|
||||
private const val TAG = "LedGrabNotifListener"
|
||||
private const val PY_MODULE = "ledgrab.core.processing.os_notification_listener"
|
||||
}
|
||||
}
|
||||
@@ -24,6 +24,7 @@ import android.widget.ImageView
|
||||
import android.widget.LinearLayout
|
||||
import android.widget.ScrollView
|
||||
import android.widget.TextView
|
||||
import androidx.core.app.NotificationManagerCompat
|
||||
import androidx.core.content.ContextCompat
|
||||
import androidx.core.splashscreen.SplashScreen.Companion.installSplashScreen
|
||||
import com.google.zxing.BarcodeFormat
|
||||
@@ -55,6 +56,8 @@ class MainActivity : Activity() {
|
||||
private const val REQUEST_POST_NOTIFICATIONS = 1002
|
||||
private const val REQUEST_RECORD_AUDIO = 1003
|
||||
private const val QR_SIZE_PX = 560
|
||||
private const val NOTIF_PREFS = "ledgrab_notif"
|
||||
private const val KEY_NOTIF_ACCESS_PROMPTED = "notif_access_prompted"
|
||||
}
|
||||
|
||||
// Stopped-state views (always inflated).
|
||||
@@ -64,6 +67,7 @@ class MainActivity : Activity() {
|
||||
private lateinit var versionText: TextView
|
||||
private lateinit var autostartCheck: CheckBox
|
||||
private lateinit var autostartPrefs: AutostartPrefs
|
||||
private lateinit var grantNotificationButton: Button
|
||||
|
||||
// Running-state views (lazy-inflated via ViewStub).
|
||||
private lateinit var runningPanelStub: ViewStub
|
||||
@@ -107,6 +111,7 @@ class MainActivity : Activity() {
|
||||
toggleButton = findViewById(R.id.toggle_button)
|
||||
versionText = findViewById(R.id.version_text)
|
||||
autostartCheck = findViewById(R.id.autostart_check)
|
||||
grantNotificationButton = findViewById(R.id.grant_notification_button)
|
||||
|
||||
val versionName = packageManager.getPackageInfo(packageName, 0).versionName
|
||||
versionText.text = getString(R.string.version_prefix, versionName ?: "?")
|
||||
@@ -127,8 +132,10 @@ class MainActivity : Activity() {
|
||||
autostartCheck.visibility = View.GONE
|
||||
}
|
||||
|
||||
grantNotificationButton.setOnClickListener { openNotificationListenerSettings() }
|
||||
toggleButton.setOnClickListener { startCapture() }
|
||||
|
||||
updateNotificationAccessUi()
|
||||
updateUI()
|
||||
}
|
||||
|
||||
@@ -149,12 +156,16 @@ class MainActivity : Activity() {
|
||||
|
||||
override fun onResume() {
|
||||
super.onResume()
|
||||
if (!::stoppedPanel.isInitialized) return
|
||||
// Restart the pulse if we returned to the foreground while the
|
||||
// service is still running. The running panel's view may have
|
||||
// been recreated; ensureRunningPanelInflated already keys off
|
||||
// the field reference.
|
||||
if (CaptureService.isRunning && ::stoppedPanel.isInitialized) {
|
||||
// service is still running. The running panel's view may have been
|
||||
// recreated; ensureRunningPanelInflated already keys off the field
|
||||
// reference. When stopped, refresh the notification-access button —
|
||||
// the user may have just granted/revoked access in Settings.
|
||||
if (CaptureService.isRunning) {
|
||||
updateUI()
|
||||
} else {
|
||||
updateNotificationAccessUi()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -197,6 +208,7 @@ class MainActivity : Activity() {
|
||||
|
||||
private fun startRootCaptureService() {
|
||||
ensureNotificationPermission()
|
||||
ensureNotificationListenerAccess()
|
||||
ContextCompat.startForegroundService(this, CaptureService.createRootIntent(this))
|
||||
updateUI()
|
||||
}
|
||||
@@ -216,6 +228,7 @@ class MainActivity : Activity() {
|
||||
|
||||
private fun startCaptureService(resultCode: Int, resultData: Intent) {
|
||||
ensureNotificationPermission()
|
||||
ensureNotificationListenerAccess()
|
||||
ensureAudioPermission()
|
||||
val intent = CaptureService.createIntent(this, resultCode, resultData)
|
||||
ContextCompat.startForegroundService(this, intent)
|
||||
@@ -493,4 +506,60 @@ class MainActivity : Activity() {
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/** Whether the user has granted notification-listener access to this app. */
|
||||
private fun isNotificationAccessGranted(): Boolean =
|
||||
NotificationManagerCompat.getEnabledListenerPackages(this).contains(packageName)
|
||||
|
||||
/** Open the system Notification-access screen (manual affordance / re-grant). */
|
||||
private fun openNotificationListenerSettings() {
|
||||
runCatching {
|
||||
startActivity(Intent(Settings.ACTION_NOTIFICATION_LISTENER_SETTINGS))
|
||||
}.onFailure { Log.w(TAG, "Notification-access settings unavailable: ${it.message}") }
|
||||
}
|
||||
|
||||
/**
|
||||
* Prompt-once-then-remember: the first time capture starts without
|
||||
* notification-listener access, open the settings screen so the user can
|
||||
* grant it — then never nag again (the manual "Grant notification access"
|
||||
* button stays available). Fire-and-forget like [ensureNotificationPermission].
|
||||
*/
|
||||
private fun ensureNotificationListenerAccess() {
|
||||
if (isNotificationAccessGranted()) return
|
||||
val prefs = getSharedPreferences(NOTIF_PREFS, MODE_PRIVATE)
|
||||
if (prefs.getBoolean(KEY_NOTIF_ACCESS_PROMPTED, false)) return
|
||||
prefs.edit().putBoolean(KEY_NOTIF_ACCESS_PROMPTED, true).apply()
|
||||
openNotificationListenerSettings()
|
||||
}
|
||||
|
||||
/**
|
||||
* Show the "Grant notification access" button only while access is missing,
|
||||
* then re-wire the D-pad focus chain. Called on create and on resume
|
||||
* (access can change in Settings while we're backgrounded).
|
||||
*/
|
||||
private fun updateNotificationAccessUi() {
|
||||
if (!::grantNotificationButton.isInitialized) return
|
||||
grantNotificationButton.visibility =
|
||||
if (isNotificationAccessGranted()) View.GONE else View.VISIBLE
|
||||
wireStoppedFocusChain()
|
||||
}
|
||||
|
||||
/**
|
||||
* Link the visible stopped-panel controls into a single up/down D-pad chain.
|
||||
* Both optional controls (the grant-access button and the root-only autostart
|
||||
* checkbox) may be GONE, so the chain is computed from whatever is visible —
|
||||
* a static nextFocus pointing at a GONE view would strand the focus on a TV
|
||||
* remote.
|
||||
*/
|
||||
private fun wireStoppedFocusChain() {
|
||||
val chain = listOfNotNull(
|
||||
toggleButton,
|
||||
grantNotificationButton.takeIf { it.visibility == View.VISIBLE },
|
||||
autostartCheck.takeIf { it.visibility == View.VISIBLE },
|
||||
)
|
||||
chain.forEachIndexed { i, view ->
|
||||
view.nextFocusUpId = (chain.getOrNull(i - 1) ?: view).id
|
||||
view.nextFocusDownId = (chain.getOrNull(i + 1) ?: view).id
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -66,6 +66,21 @@
|
||||
android:focusableInTouchMode="true"
|
||||
android:nextFocusDown="@+id/autostart_check" />
|
||||
|
||||
<!-- Shown only while notification-listener access is missing. The D-pad
|
||||
focus chain is wired at runtime (wireStoppedFocusChain) because this
|
||||
button and the autostart checkbox are both conditionally visible. -->
|
||||
<Button
|
||||
android:id="@+id/grant_notification_button"
|
||||
style="@style/Widget.LedGrab.Button.Secondary"
|
||||
android:layout_width="320dp"
|
||||
android:layout_height="56dp"
|
||||
android:layout_marginTop="20dp"
|
||||
android:text="@string/btn_grant_notification_access"
|
||||
android:textSize="18sp"
|
||||
android:focusable="true"
|
||||
android:focusableInTouchMode="true"
|
||||
android:visibility="gone" />
|
||||
|
||||
<CheckBox
|
||||
android:id="@+id/autostart_check"
|
||||
android:layout_width="wrap_content"
|
||||
|
||||
@@ -25,4 +25,6 @@
|
||||
<string name="notification_channel_description">Отображается, пока LedGrab захватывает экран.</string>
|
||||
<string name="notification_title">LedGrab работает</string>
|
||||
<string name="notification_text">Веб-интерфейс: %1$s</string>
|
||||
<string name="notification_listener_label">Захват уведомлений LedGrab</string>
|
||||
<string name="btn_grant_notification_access">Разрешить доступ к уведомлениям</string>
|
||||
</resources>
|
||||
|
||||
@@ -25,4 +25,6 @@
|
||||
<string name="notification_channel_description">LedGrab 捕获屏幕时显示。</string>
|
||||
<string name="notification_title">LedGrab 运行中</string>
|
||||
<string name="notification_text">Web界面:%1$s</string>
|
||||
<string name="notification_listener_label">LedGrab 通知捕获</string>
|
||||
<string name="btn_grant_notification_access">授予通知访问权限</string>
|
||||
</resources>
|
||||
|
||||
@@ -25,4 +25,6 @@
|
||||
<string name="notification_channel_description">Shows while LedGrab is capturing the screen.</string>
|
||||
<string name="notification_title">LedGrab Running</string>
|
||||
<string name="notification_text">Web UI: %1$s</string>
|
||||
<string name="notification_listener_label">LedGrab notification capture</string>
|
||||
<string name="btn_grant_notification_access">Grant notification access</string>
|
||||
</resources>
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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"]
|
||||
Reference in New Issue
Block a user