USB connection notifications added.

I38.md

@@ -271,6 +271,15 @@ I38 provides audio feedback for many actions.
 - Mode sounds confirm when you enter or leave a mode.
 - Workspace sounds confirm workspace changes.
 - Fullscreen sounds confirm fullscreen changes.
+- Optional USB alerts play a rising sound on connection and its reverse on removal.
+
+Enable USB alerts separately during setup. With `python-pyudev` installed, I38
+uses SoX for sounds and Speech Dispatcher for spoken device names, falling back
+to "USB device connected" or "USB device disconnected" when a product name is
+unavailable. Disconnects use cached names. A wireless receiver may report its
+own name instead of the keyboard or mouse paired with it. Existing devices are
+recorded silently at startup; alerts indicate USB attachment, not mounting or
+device readiness.
 
 ## Application Menu and Running Programs
 

README.md

@@ -42,9 +42,11 @@ Optional features use these packages when installed:
 - playerctl: music controls
 - python-pillow: For OCR
 - python-pytesseract: For OCR
+- python-pyudev: [optional] for USB connection and removal alerts.
 - remind: [optional] For reminder notifications, Requires notify-daemon and notify-send for automatic reminders.
 - scrot: For OCR
 - sox: for sounds.
+- speech-dispatcher: [optional] provides `spd-say` for spoken USB alerts and other I38 announcements.
 - tesseract: For OCR
 - xprintidle: [optional] For screen lock autolock on X11/i3
 - i38lock: [optional] Secure accessible screen locker. On Stormux, install it from the repository. It requires Cthulhu or Orca 50 or newer for screen reader feedback.
@@ -76,6 +78,46 @@ During setup, the screen reader picker includes Cthulhu, Orca, and None. If no s
 
 During setup, I38 can configure one screen lock method. The legacy I38 PIN lock is a privacy screen only. I38Lock is a true PAM-backed screen locker, available at https://git.stormux.org/storm/i38lock and packaged in the Stormux repository. I38 offers I38Lock only when `i38lock` is installed and the screen-reader setup can support it: Cthulhu, or Orca 50 or newer. I38 does not enable the PIN lock and I38Lock at the same time.
 
+## USB Connection Alerts (Optional)
+
+Setup offers USB connection sounds and spoken device names separately from window
+event sounds. Default configuration mode enables USB alerts. Install
+`python-pyudev`, plus SoX for sounds and Speech Dispatcher (`spd-say`) for speech.
+If either output command is absent, the available output still works.
+
+Connections play a 0.65-second rising, pulsing triangle tone; removals play the
+same sound reversed. Alerts use the device's product and manufacturer names,
+with `USB device connected` or `USB device disconnected` when no product name is
+available. Names are cached so unplugged devices can still be identified.
+Wireless receivers may expose only their receiver name, not the names of paired
+keyboards or mice. These alerts describe USB attachment, not successful mounting
+or readiness for use.
+
+The listener observes udev events without changing rules, claiming devices, or
+mounting storage. It filters out USB interface events, silently records devices
+already attached at startup, and allows one listener per X display. It survives
+i3 restarts and exits when i3 logs out. Speech uses `spd-say -P important -w`
+without globally cancelling other speech.
+
+After updating, regenerate the I38 configuration and enable the USB option;
+`./i38.sh -u` alone copies scripts without adding the startup entry. Saved
+preferences use `usbNotifications=0` to enable alerts and `usbNotifications=1`
+to disable them. To try the source listener before regeneration, run
+`python3 scripts/usb_notifications.py` in a terminal within your i3 session,
+then press Control+C when finished. Stop that foreground trial before testing
+the installed listener.
+
+Run the focused checks with
+`python3 -B -m unittest discover -s tests -p 'test_usb_notifications.py' -v`.
+For human review, connect and remove a named USB device, an unnamed device if
+available, and a multifunction device: expect one alert per USB device, including
+the cached name on removal. Devices behind a hub each generate their own alert.
+Restart i3 and check that subsequent hotplug events are announced once and that
+devices already attached do not produce startup alerts. Confirm normal keyboard,
+mouse, storage mounting, existing window/mode sounds, and other speech still
+work. Check the USB setup option both enabled and disabled. Automated checks do
+not establish audible sound or spoken output.
+
 ## GTK Application Sound Themes (Optional)
 
 This section is completely optional and separate from I38's window manager sounds. If you want GTK-based applications (like file managers, terminal emulators, text editors, etc.) to play sounds for their own events (button presses, menu navigation, dialog boxes, etc.), you can configure a GTK sound theme.

extra/arch-linux.sh

@@ -193,6 +193,10 @@ add_optional_dependencies() {
     command_exists playerctl || add_install_package playerctl
     command_exists scrot || add_install_package scrot
     command_exists sox || add_install_package sox
+    command_exists spd-say || add_install_package speech-dispatcher
+    if ! python_import_works 'import pyudev'; then
+        add_install_package python-pyudev
+    fi
     command_exists udiskie || add_install_package udiskie
     command_exists xclip || add_install_package xclip
     command_exists xprintidle || add_install_package xprintidle

i38.sh

@@ -260,6 +260,7 @@ set_default_configuration() {
         brlapi=0
     fi
     sounds=0
+    usbNotifications=0
     screenlockMethod="none"
     screenlockPinHash=""
     screenlockAutolockSeconds=0
@@ -536,6 +537,7 @@ dex="$dex"
 batteryAlert="${batteryAlert:-1}"
 brlapi="$brlapi"
 sounds="$sounds"
+usbNotifications="${usbNotifications:-1}"
 
 # Screen lock
 screenlockMethod="${screenlockMethod:-none}"
@@ -1066,6 +1068,14 @@ if [[ -z "$sounds" ]]; then
         sounds=1
     fi
 fi
+if [[ -z "$usbNotifications" ]]; then
+    configChanged=1
+    if yesno "Do you want USB connection sounds and spoken device names? Requires python-pyudev, sox for sounds, and speech-dispatcher for announcements."; then
+        usbNotifications=0
+    else
+        usbNotifications=1
+    fi
+fi
 if [[ $defaultConfig -eq 0 && -z "$screenlockPinHash" ]]; then
     screenlockPinFile="${i3Path}/.screenpin"
     if [[ -f "$screenlockPinFile" ]]; then
@@ -1575,6 +1585,9 @@ cat << EOF >> "${i3Path}/config"
 $(if [[ $sounds -eq 0 ]]; then
     echo "exec_always --no-startup-id ${i3Path}/scripts/sound.py"
 fi
+if [[ ${usbNotifications:-1} -eq 0 ]]; then
+    echo "exec_always --no-startup-id python3 \"${i3Path}/scripts/usb_notifications.py\""
+fi
 if command -v xrdb &> /dev/null; then
     echo "exec --no-startup-id xrdb -merge ${i3Path}/I38.Xresources"
 fi

scripts/usb_notifications.py

@@ -0,0 +1,229 @@
+#!/usr/bin/env python3
+
+# This file is part of I38, licensed under the GNU GPL, version 3 or later.
+
+"""Announce USB hotplug events without claiming or configuring devices."""
+
+import collections
+import fcntl
+import hashlib
+import os
+from pathlib import Path
+import queue
+import shutil
+import signal
+import subprocess
+import sys
+import threading
+import time
+
+
+CONNECT_SOUND = (
+    'play', '-q', '-r', '44100', '-c', '1', '-n',
+    'synth', '0.65', 'triangle', '360:1190', 'tremolo', '14', '85',
+    'fade', 'h', '0.01', '0.65', '0.1', 'gain', '-14',
+)
+
+
+def clean_name(value):
+    if isinstance(value, bytes):
+        value = value.decode('utf-8', errors='replace')
+    value = value or ''
+    value = ''.join(char if char.isprintable() else ' ' for char in value)
+    return ' '.join(value.replace('_', ' ').split())[:160]
+
+
+def read_attribute(device, name):
+    try:
+        return clean_name(device.attributes.get(name))
+    except OSError:
+        # Sysfs attributes can disappear before the removal event arrives.
+        return ''
+
+
+def device_name(device):
+    model = (clean_name(device.properties.get('ID_MODEL_FROM_DATABASE'))
+             or read_attribute(device, 'product')
+             or clean_name(device.properties.get('ID_MODEL')))
+    if not model:
+        return 'USB device'
+    vendor = (read_attribute(device, 'manufacturer')
+              or clean_name(device.properties.get('ID_VENDOR_FROM_DATABASE'))
+              or clean_name(device.properties.get('ID_VENDOR')))
+    if vendor and vendor.casefold() not in model.casefold():
+        return '{} {}'.format(vendor, model)[:160]
+    return model
+
+
+class UsbDevices:
+    def __init__(self, devices):
+        # Seed silently so devices already present are named when unplugged.
+        self.names = {device.device_path: device_name(device)
+                      for device in devices if self.is_usb_device(device)}
+        self.removedPaths = collections.deque(maxlen=64)
+
+    @staticmethod
+    def is_usb_device(device):
+        return (device.subsystem == 'usb'
+                and device.device_type == 'usb_device')
+
+    def handle_event(self, device):
+        if not self.is_usb_device(device):
+            return None
+        action = device.action
+        devicePath = device.device_path
+        if action in ('add', 'change'):
+            wasPresent = devicePath in self.names
+            name = device_name(device)
+            if name != 'USB device' or not wasPresent:
+                self.names[devicePath] = name
+            if action != 'add' or wasPresent:
+                return None
+            if devicePath in self.removedPaths:
+                self.removedPaths.remove(devicePath)
+            return action, '{} connected'.format(self.names[devicePath])
+        if action == 'remove':
+            if devicePath in self.removedPaths:
+                return None
+            name = self.names.pop(devicePath, None) or device_name(device)
+            self.removedPaths.append(devicePath)
+            return action, '{} disconnected'.format(name)
+        return None
+
+
+class Announcer:
+    def __init__(self, stopEvent):
+        self.stopEvent = stopEvent
+        self.playCommand = shutil.which('play')
+        self.speechCommand = shutil.which('spd-say')
+        self.events = queue.Queue(maxsize=32)
+
+    def enqueue(self, event):
+        if event is None or self.stopEvent.is_set():
+            return
+        try:
+            self.events.put_nowait(event)
+        except queue.Full:
+            # Prefer recent events during a burst, without blocking udev reads.
+            try:
+                self.events.get_nowait()
+            except queue.Empty:
+                pass
+            self.events.put_nowait(event)
+
+    def run_command(self, command, timeout):
+        if self.stopEvent.is_set():
+            return
+        try:
+            with subprocess.Popen(command, stdout=subprocess.DEVNULL,
+                                  stderr=subprocess.DEVNULL) as process:
+                deadline = time.monotonic() + timeout
+                while process.poll() is None:
+                    if self.stopEvent.wait(0.05) or time.monotonic() >= deadline:
+                        process.terminate()
+                        try:
+                            process.wait(timeout=1)
+                        except subprocess.TimeoutExpired:
+                            process.kill()
+                        return
+                if process.returncode:
+                    print('USB notifications: {} exited with status {}'.format(
+                        command[0], process.returncode), file=sys.stderr)
+        except OSError as error:
+            print('USB notifications: {}'.format(error), file=sys.stderr)
+
+    def run(self):
+        while not self.stopEvent.is_set():
+            try:
+                action, message = self.events.get(timeout=0.2)
+            except queue.Empty:
+                continue
+            if self.playCommand:
+                command = [self.playCommand] + list(CONNECT_SOUND[1:])
+                if action == 'remove':
+                    command.append('reverse')
+                self.run_command(command, 5)
+            if self.speechCommand:
+                # Do not use -C: hotplug must not globally cancel other speech.
+                self.run_command([self.speechCommand, '-P', 'important', '-w',
+                                  '--', message], 15)
+
+
+def acquire_lock():
+    displayName = os.environ.get('DISPLAY', '')
+    if not displayName:
+        raise RuntimeError('an i3 X11 session is required (DISPLAY is unset)')
+    runtimeDir = Path(os.environ.get('XDG_RUNTIME_DIR')
+                      or Path.home() / '.cache' / 'i38')
+    runtimeDir.mkdir(mode=0o700, parents=True, exist_ok=True)
+    sessionId = hashlib.sha256(displayName.encode('utf-8')).hexdigest()[:16]
+    lockPath = runtimeDir / ('i38-usb-{}.lock'.format(sessionId))
+    descriptor = os.open(str(lockPath), os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW,
+                         0o600)
+    lockFile = os.fdopen(descriptor, 'w')
+    try:
+        fcntl.flock(lockFile, fcntl.LOCK_EX | fcntl.LOCK_NB)
+    except BlockingIOError:
+        lockFile.close()
+        return None
+    # Leave the inode in place so concurrent launches cannot lock a new file.
+    return lockFile
+
+
+def main():
+    try:
+        import i3ipc
+        import pyudev
+    except ImportError as error:
+        print('USB notifications require python-pyudev and python-i3ipc: {}'.format(
+            error), file=sys.stderr)
+        return 1
+
+    try:
+        lockFile = acquire_lock()
+        if lockFile is None:
+            return 0
+        with lockFile:
+            connection = i3ipc.Connection(auto_reconnect=True)
+            stopEvent = threading.Event()
+            announcer = Announcer(stopEvent)
+            if not announcer.playCommand and not announcer.speechCommand:
+                print('USB notifications require play or spd-say.', file=sys.stderr)
+                return 1
+            context = pyudev.Context()
+            monitor = pyudev.Monitor.from_netlink(context)
+            monitor.filter_by(subsystem='usb', device_type='usb_device')
+            # Start buffering before enumeration to avoid losing a removal.
+            monitor.start()
+            devices = UsbDevices(context.list_devices(
+                subsystem='usb', DEVTYPE='usb_device'))
+            observer = pyudev.MonitorObserver(
+                monitor, callback=lambda device: announcer.enqueue(
+                    devices.handle_event(device)))
+            worker = threading.Thread(target=announcer.run, daemon=True)
+
+            def stop_listener(*unused):
+                stopEvent.set()
+                connection.main_quit()
+
+            # Reconnect through i3 restarts; exit with the actual desktop session.
+            connection.on('shutdown::exit', stop_listener)
+            signal.signal(signal.SIGTERM, stop_listener)
+            signal.signal(signal.SIGINT, stop_listener)
+            try:
+                observer.start()
+                worker.start()
+                connection.main()
+            finally:
+                stopEvent.set()
+                observer.stop()
+                if worker.is_alive():
+                    worker.join(timeout=2)
+        return 0
+    except (OSError, RuntimeError) as error:
+        print('USB notifications: {}'.format(error), file=sys.stderr)
+        return 1
+
+
+if __name__ == '__main__':
+    sys.exit(main())