jonathan/ISR
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: remove dead code from isr.py

Browse Source 
- Drop unused struct import, AudioDevice.extra/description fields
- Remove unused get_preferred_backend() and the backend priority
  machinery (only one backend exists)
- Deduplicate SoundcardRecorder.close_current_file via super()
- Remove duplicate config-exists check in main()
- Simplify --list-devices output: drop dead monitor grouping and the
  nonexistent pipewire backend example

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
jonathan
2026-06-10 11:41:37 +02:00
parent 4539ff78fa
commit 792f2b1fd5
1 changed files with 18 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
isr.py
+13 -66
View File
@@ -10,7 +10,6 @@ import os
import sys
import time
import wave
import struct
import signal
import logging
import threading
@@ -18,7 +17,7 @@ import configparser
import subprocess
import shutil
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional, Dict, Any, List, Callable
@@ -57,8 +56,6 @@ class AudioDevice:
backend: str # Backend name ('alsa')
is_default: bool = False # Is system default
is_monitor: bool = False # Is a monitor/loopback source
description: str = "" # Extended description
extra: Dict[str, Any] = field(default_factory=dict)
def __str__(self):
flags = []
@@ -74,7 +71,6 @@ class AudioBackend(ABC):
"""Abstract base for audio capture backends."""
name: str = "base"
priority: int = 0 # Higher = preferred
@classmethod
@abstractmethod
@@ -99,7 +95,6 @@ class ALSABackend(AudioBackend):
"""ALSA backend using arecord (raw PCM output, no sound server required)."""
name = "alsa"
priority = 5 # Lowest priority — direct hardware access, use when no sound server runs
@classmethod
def is_available(cls) -> bool:
@@ -251,19 +246,12 @@ class AudioSystem:
def get_backend(self, name: str) -> Optional[AudioBackend]:
return self._backends.get(name)
def get_preferred_backend(self) -> Optional[AudioBackend]:
"""Get the highest priority available backend."""
if not self._backends:
return None
return max(self._backends.values(), key=lambda b: b.__class__.priority)
def list_all_devices(self) -> List[AudioDevice]:
"""List devices from all available backends."""
all_devices = []
seen_names = set()
# Get devices from backends in priority order
for cls in sorted(self._backend_classes, key=lambda c: -c.priority):
for cls in self._backend_classes:
if cls.name in self._backends:
for dev in cls.list_devices():
# Deduplicate by name (same device may appear in multiple backends)
@@ -733,16 +721,9 @@ class SoundcardRecorder(BaseRecorder):
self.audio_buffer.clear()
def close_current_file(self):
"""Close current recording file."""
"""Flush any buffered audio, then close the current recording file."""
self._flush_buffer_to_file()
if self.current_file:
try:
self.current_file.close()
self.logger.info(f"[{self.name}] Closed: {self.current_filename}")
except Exception as e:
self.logger.error(f"[{self.name}] Error closing file: {e}")
self.current_file = None
self.current_filename = None
super().close_current_file()
def record(self):
"""Main recording loop."""
@@ -990,22 +971,15 @@ def list_audio_devices():
print(" ISR Audio Device Discovery")
print("=" * 70)
# Check available backends
available_backends = []
if ALSABackend.is_available():
available_backends.append(('alsa', 'ALSA (arecord)', 5))
if not available_backends:
if not ALSABackend.is_available():
print("\n No audio backends available!")
print("\n Install one of:")
print(" sudo apt install alsa-utils (ALSA, always available on Linux)")
print("\n Install ALSA utilities:")
print(" sudo apt install alsa-utils")
print()
return
print("\n Available Backends:")
for name, label, priority in sorted(available_backends, key=lambda x: -x[2]):
marker = " (preferred)" if priority == max(b[2] for b in available_backends) else ""
print(f" - {label}{marker}")
print(" - ALSA (arecord)")
# Initialize audio system and list devices
audio_system = AudioSystem(logger)
@@ -1016,30 +990,10 @@ def list_audio_devices():
print()
return
# Group by type
monitors = [d for d in devices if d.is_monitor]
inputs = [d for d in devices if not d.is_monitor]
if inputs:
print("\n Input Devices:")
print(" " + "-" * 68)
for dev in inputs:
flags = []
if dev.is_default:
flags.append("DEFAULT")
flag_str = f" [{', '.join(flags)}]" if flags else ""
print(f"\n {dev.name}{flag_str}")
print(f" ID: {dev.id} | Backend: {dev.backend}")
print(f" Channels: {dev.channels} | Sample Rate: {dev.sample_rate} Hz")
if monitors:
print("\n Monitor/Loopback Sources:")
print(" " + "-" * 68)
for dev in monitors:
flags = ["MONITOR"]
if dev.is_default:
flags.append("DEFAULT")
flag_str = f" [{', '.join(flags)}]"
for dev in devices:
flag_str = " [DEFAULT]" if dev.is_default else ""
print(f"\n {dev.name}{flag_str}")
print(f" ID: {dev.id} | Backend: {dev.backend}")
print(f" Channels: {dev.channels} | Sample Rate: {dev.sample_rate} Hz")
@@ -1048,10 +1002,9 @@ def list_audio_devices():
print(" Configuration Examples:")
print("-" * 70)
print(" device = default # Use system default input")
print(" device = monitor # Use first monitor/loopback source")
print(" device = <name> # Match by partial name")
print(" device = <id> # Use exact backend ID")
print(" backend = pipewire # Force specific backend")
print(" device = <id> # Use exact ALSA ID, e.g. hw:0,0")
print(" device = <pcm> # Any ALSA PCM name from asound.conf, e.g. shared_mic")
print("=" * 70 + "\n")
@@ -1061,17 +1014,11 @@ def main():
list_audio_devices()
return
# Get config file
# Get config file (RecorderManager exits with a usage message if it's missing)
config_file = 'config.ini'
if len(sys.argv) > 1:
config_file = sys.argv[1]
if not os.path.exists(config_file):
print(f"Error: Configuration file '{config_file}' not found!")
print("Usage: python ISR.py [config.ini]")
print(" python ISR.py --list-devices")
sys.exit(1)
# Docker sends SIGTERM before SIGKILL — treat it the same as Ctrl+C
def _sigterm(sig, frame):
raise KeyboardInterrupt()