diff --git a/CLAUDE.md b/CLAUDE.md index ec00d07..1f427b3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -6,7 +6,7 @@ A custom mastering toolkit that provides metrics to evaluate audio masterings th ### Core features - **Audio Analysis**: Uses librosa to analyze audio files (MP3/WAV/FLAC support) -- **Power Visualization**: Generates colorized power magnitude graphs over time +- **Pluggable Metrics**: Switchable visualizations (RMS Power, Waveform, LUFS; DR next) via a `Metric` ABC - **Metadata Extraction**: Reads ID3 tags from MP3 files for better file identification - **Modular GUI Architecture**: Complete PyQt5 interface with drag-and-drop and file dialog support - **Font Management**: Comprehensive CJK-compatible font system with user-provided font support @@ -29,7 +29,8 @@ A custom mastering toolkit that provides metrics to evaluate audio masterings th #### `analysis_results_manager.py` - Background threading for audio analysis -- Results caching and management +- Caches both the loaded `AudioFile` and per-metric `compute()` output, so + metric/font switches re-render from cache without reloading librosa - Progress tracking and error handling #### `audio_visualization_widget.py` @@ -40,7 +41,18 @@ A custom mastering toolkit that provides metrics to evaluate audio masterings th - Unified font control system with clustered interface - Auto-detection of custom fonts from `fonts/` directory - System font discovery and CJK compatibility -- Auto-regeneration of plots when fonts change +- Font changes trigger a cheap re-render of the cached metric data + +#### `plot_control_widget.py` +- Metric selector dropdown driven by the `metrics.METRICS` registry +- Houses the `Refresh Plot` button (foundation for upcoming style controls) + +#### `metrics.py` +- Pluggable `Metric` ABC: `compute(audio_file) -> data` (heavy, worker thread) + and `render(data, file_path) -> Figure` (cheap, GUI thread) +- Current registry: `RMSPowerMetric`, `WaveformMetric`, `LUFSMetric` + (BS.1770 short-term + integrated, via pyloudnorm) — drop in new ones (DR, + spectrum) by appending an instance to `METRICS` #### `master_core.py` - Defines the `AudioFile` class: librosa loading, rolling RMS power, BPM detection @@ -57,22 +69,20 @@ A custom mastering toolkit that provides metrics to evaluate audio masterings th ### GUI features - **File management**: Drag-and-drop and file dialog for audio selection -- **Font control**: Unified font selector with size control and plot regeneration +- **Font control**: Unified font selector with size control +- **Plot control**: Metric selector + refresh-plot button - **Analysis display**: Real-time visualization with metadata panels - **Modular architecture**: Self-contained widgets for easy layout management ## Future development plans ### Short-term (urgent) -1. **Plot control widget cluster** - - Move 'Refresh Plot' into dedicated plot/graph widget cluster - - Add metric selection widget (choose which analysis to display) - - Implement plot style controller (colormap, line vs bar, etc.) - - Prepare foundation for mastering comparison features +1. **Plot control widget cluster** *(metric selector + Refresh Plot done; still TODO)* + - Plot style controller (colormap, line vs bar, etc.) + - Foundation for mastering comparison features ### Short-term (not urgent) -1. **Enhanced metrics** - - LUFS loudness measurement implementation +1. **Enhanced metrics** *(plug new ones into `metrics.METRICS`)* - Dynamic range measurement (DR meter) - Peak-to-average ratio analysis - Frequency spectrum analysis diff --git a/README.md b/README.md index 42c8263..ec1a6fa 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ Developed with Claude Code assistance. ### Roadmap See [CLAUDE.md](CLAUDE.md) for the full development roadmap. Near-term: -plot-control widget cluster, LUFS, dynamic range, interactive axis controls. +dynamic range, plot-style controls, interactive axis controls. ## Quick start @@ -40,19 +40,27 @@ selector. The directory is gitignored to avoid bundling licensed font data. See [CJK_FONTS.md](CJK_FONTS.md) for details. ## Dependencies -`librosa`, `numpy`, `matplotlib`, `mutagen`, `PyQt5` — all pinned through -`uv.lock`. Python 3.10+. +`librosa`, `numpy`, `matplotlib`, `mutagen`, `pyloudnorm`, `PyQt5` — all pinned +through `uv.lock`. Python 3.10+. ## Architecture | Module | Responsibility | | --- | --- | | `main.py` | `MainWindow` + the `ujm` entry point | -| `analysis_results_manager.py` | Background `QThread` worker, result cache | +| `analysis_results_manager.py` | Background `QThread` worker, result + metric-data cache | | `master_core.py` | `AudioFile`: librosa loading, RMS rolling window, BPM | -| `plotting_engine.py` | Matplotlib `Figure` builder for the power graph | +| `metrics.py` | Pluggable `Metric` ABC + registry (RMS Power, Waveform, …) | | `audio_visualization_widget.py` | Embedded `FigureCanvasQTAgg` host | | `font_manager.py` | Custom + system CJK font discovery, matplotlib/Qt config | -| `font_control_widget.py` | Font picker, size slider, refresh-plot button | +| `font_control_widget.py` | Font picker + size slider | +| `plot_control_widget.py` | Metric selector + refresh-plot button | | `logger_setup.py` | CLI log-level parsing + custom TRACE level | | `setup_fonts.py` | Diagnostic utility (run standalone) | + +### Adding a metric + +Subclass `Metric` in `metrics.py`, implement `compute(audio_file) -> data` (the +heavy part, runs on the worker thread) and `render(data, file_path) -> Figure` +(cheap, runs on the GUI thread). Register the instance in the `METRICS` dict at +the bottom of the file — it shows up in the dropdown automatically. diff --git a/analysis_results_manager.py b/analysis_results_manager.py index 54612c2..9309779 100644 --- a/analysis_results_manager.py +++ b/analysis_results_manager.py @@ -4,180 +4,241 @@ Manages analysis queue and coordinates between components. """ from PyQt5.QtCore import QObject, pyqtSignal, QThread -from dataclasses import dataclass -from typing import Optional +from dataclasses import dataclass, field +from typing import Any, Optional import os import logging from master_core import AudioFile -from plotting_engine import PlottingEngine +from font_manager import safe_title +from metrics import METRICS, DEFAULT_METRIC_ID, Metric @dataclass class AnalysisResult: - """Container for audio analysis results.""" - file_path: str - song_name: str - bpm: float - max_amplitude: float - avg_amplitude: float - times: list - rms_array: list - analysis_successful: bool = True - error_message: str = "" + """Container for audio analysis results.""" + file_path: str + audio_file: AudioFile + song_name: str + bpm: float + max_amplitude: float + avg_amplitude: float + metric_data: dict[str, Any] = field(default_factory=dict) + analysis_successful: bool = True + error_message: str = "" + + def metadata_text(self) -> str: + return ( + f"Track: {safe_title(self.song_name)}\n" + f"BPM: {self.bpm:.1f}\n" + f"Max Amplitude: {self.max_amplitude:.3f}\n" + f"Avg Amplitude: {self.avg_amplitude:.3f}" + ) class AudioAnalysisWorker(QThread): - """ - Worker thread for audio analysis to prevent GUI freezing. - Performs heavy librosa operations in background. - """ - - # Signals for communicating with main thread - progressUpdate = pyqtSignal(str, int) # message, percentage - analysisCompleted = pyqtSignal(str, object) # file_path, AnalysisResult - analysisError = pyqtSignal(str, str) # file_path, error_message - - def __init__(self, file_path: str, window: int = 10, hop: int = 2): - super().__init__() - self.file_path = file_path - self.window = window - self.hop = hop - self.logger = logging.getLogger(__name__) - - def run(self): - """Main thread execution - performs audio analysis.""" - try: - self.logger.info(f"Starting analysis of: {os.path.basename(self.file_path)}") - self.progressUpdate.emit("Loading audio file...", 10) - - # Create AudioFile and load audio data - audio_file = AudioFile(self.file_path) - self.progressUpdate.emit("Audio loaded, detecting tempo...", 30) - - # BPM is already calculated in __init__, now do RMS analysis - self.progressUpdate.emit("Computing RMS power levels...", 60) - audio_file.get_energy_levels_over_time(window=self.window, hop=self.hop) - - self.progressUpdate.emit("Finalizing analysis...", 90) - - # Extract analysis results - result = AnalysisResult( - file_path=self.file_path, - song_name=audio_file.song_name, - bpm=audio_file.get_bpm(), - max_amplitude=audio_file.max_amplitude, - avg_amplitude=audio_file.avg_amplitude, - times=audio_file.get_times(), - rms_array=audio_file.rms_array, - analysis_successful=True - ) - - self.progressUpdate.emit("Analysis complete!", 100) - self.logger.info(f"Analysis completed: {os.path.basename(self.file_path)} (BPM: {result.bpm:.1f})") - - # Emit success signal - self.analysisCompleted.emit(self.file_path, result) - - except Exception as e: - error_msg = f"Analysis failed: {str(e)}" - self.logger.error(f"Analysis error for {self.file_path}: {error_msg}") - self.analysisError.emit(self.file_path, error_msg) + """Worker thread that loads audio and computes a single metric.""" + + progressUpdate = pyqtSignal(str, int) # message, percentage + analysisCompleted = pyqtSignal(str, object) # file_path, AnalysisResult + analysisError = pyqtSignal(str, str) # file_path, error_message + + def __init__(self, file_path: str, metric: Metric): + super().__init__() + self.file_path = file_path + self.metric = metric + self.logger = logging.getLogger(__name__) + + def run(self): + try: + self.logger.info(f"Starting analysis of: {os.path.basename(self.file_path)}") + self.progressUpdate.emit("Loading audio file...", 10) + + audio_file = AudioFile(self.file_path) + self.progressUpdate.emit("Audio loaded, detecting tempo...", 30) + + self.progressUpdate.emit(f"Computing {self.metric.display_name}...", 60) + metric_data = {self.metric.id: self.metric.compute(audio_file)} + + self.progressUpdate.emit("Finalizing analysis...", 90) + + result = AnalysisResult( + file_path=self.file_path, + audio_file=audio_file, + song_name=audio_file.song_name, + bpm=audio_file.get_bpm(), + max_amplitude=audio_file.max_amplitude, + avg_amplitude=audio_file.avg_amplitude, + metric_data=metric_data, + analysis_successful=True, + ) + + self.progressUpdate.emit("Analysis complete!", 100) + self.logger.info( + f"Analysis completed: {os.path.basename(self.file_path)} (BPM: {result.bpm:.1f})" + ) + self.analysisCompleted.emit(self.file_path, result) + + except Exception as e: + error_msg = f"Analysis failed: {str(e)}" + self.logger.error(f"Analysis error for {self.file_path}: {error_msg}") + self.analysisError.emit(self.file_path, error_msg) + + +class MetricComputeWorker(QThread): + """Worker thread that computes a single metric against an already-loaded AudioFile.""" + + completed = pyqtSignal(str, str, object) # file_path, metric_id, data + failed = pyqtSignal(str, str, str) # file_path, metric_id, error_message + + def __init__(self, file_path: str, audio_file: AudioFile, metric: Metric): + super().__init__() + self.file_path = file_path + self.audio_file = audio_file + self.metric = metric + self.logger = logging.getLogger(__name__) + + def run(self): + try: + self.logger.info( + f"Computing {self.metric.display_name} for {os.path.basename(self.file_path)}" + ) + data = self.metric.compute(self.audio_file) + self.completed.emit(self.file_path, self.metric.id, data) + except Exception as e: + msg = f"{self.metric.display_name} compute failed: {e}" + self.logger.error(msg) + self.failed.emit(self.file_path, self.metric.id, str(e)) class AnalysisResultsManager(QObject): + """Manages audio file analysis and coordinates between processing and GUI.""" + + # Full-analysis (load + initial metric) signals. + analysisStarted = pyqtSignal(str) + analysisCompleted = pyqtSignal(str, object) + analysisError = pyqtSignal(str, str) + progressUpdate = pyqtSignal(str, int) + + # Metric-only signals (used for switches after analysis has completed). + metricComputeStarted = pyqtSignal(str, str) # file_path, metric_id + metricReady = pyqtSignal(str, str) # file_path, metric_id + metricComputeError = pyqtSignal(str, str, str) # file_path, metric_id, error + + def __init__(self): + super().__init__() + self.results_cache: dict[str, AnalysisResult] = {} + self.current_worker: Optional[AudioAnalysisWorker] = None + self.metric_workers: dict[tuple[str, str], MetricComputeWorker] = {} + self.logger = logging.getLogger(__name__) + + def analyze_file(self, file_path: str, metric_id: str = DEFAULT_METRIC_ID): + """Kick off background analysis for the given file and metric.""" + if not os.path.exists(file_path): + error_msg = f"File not found: {file_path}" + self.logger.error(error_msg) + self.analysisError.emit(file_path, error_msg) + return + + metric = METRICS.get(metric_id) + if metric is None: + error_msg = f"Unknown metric: {metric_id}" + self.logger.error(error_msg) + self.analysisError.emit(file_path, error_msg) + return + + if self.current_worker and self.current_worker.isRunning(): + self.logger.info("Stopping previous analysis to start new one") + self.current_worker.quit() + self.current_worker.wait() + + self.analysisStarted.emit(file_path) + self.logger.info( + f"Queuing analysis: {os.path.basename(file_path)} ({metric.display_name})" + ) + + self.current_worker = AudioAnalysisWorker(file_path, metric) + self.current_worker.progressUpdate.connect(self.progressUpdate.emit) + self.current_worker.analysisCompleted.connect(self._on_worker_completed) + self.current_worker.analysisError.connect(self.analysisError.emit) + self.current_worker.start() + + def _on_worker_completed(self, file_path: str, result: AnalysisResult): + self.results_cache[file_path] = result + self.analysisCompleted.emit(file_path, result) + + def request_metric(self, file_path: str, metric_id: str) -> bool: + """Ensure the metric's data exists for the file; emit metricReady when ready. + + Returns True if the data was already cached (metricReady emitted synchronously) + or successfully kicked off (will emit later). Returns False if the file hasn't + been analysed yet or the metric id is unknown — in that case the caller + should wait for analysisCompleted or correct the metric id. """ - Manages audio file analysis and coordinates between processing and GUI. - Now uses background threads to prevent GUI freezing. + result = self.results_cache.get(file_path) + if result is None: + return False + + metric = METRICS.get(metric_id) + if metric is None: + self.logger.warning(f"Unknown metric requested: {metric_id}") + return False + + if metric_id in result.metric_data: + # Cached — emit immediately so the caller can re-render. + self.metricReady.emit(file_path, metric_id) + return True + + key = (file_path, metric_id) + existing = self.metric_workers.get(key) + if existing is not None and existing.isRunning(): + self.logger.debug(f"Metric compute already in flight: {metric_id} for {os.path.basename(file_path)}") + return True + + worker = MetricComputeWorker(file_path, result.audio_file, metric) + worker.completed.connect(self._on_metric_completed) + worker.failed.connect(self._on_metric_failed) + self.metric_workers[key] = worker + self.metricComputeStarted.emit(file_path, metric_id) + worker.start() + return True + + def _on_metric_completed(self, file_path: str, metric_id: str, data: object): + result = self.results_cache.get(file_path) + if result is not None: + result.metric_data[metric_id] = data + self.metric_workers.pop((file_path, metric_id), None) + self.metricReady.emit(file_path, metric_id) + + def _on_metric_failed(self, file_path: str, metric_id: str, error_message: str): + self.metric_workers.pop((file_path, metric_id), None) + self.metricComputeError.emit(file_path, metric_id, error_message) + + def get_metric_figure(self, file_path: str, metric_id: str): + """Render a Figure from cached metric data. Returns None if not cached. + + Never triggers compute — call `request_metric` first and listen for + `metricReady` if you need on-demand computation. """ - - # Signals for GUI communication - analysisStarted = pyqtSignal(str) # file_path - analysisCompleted = pyqtSignal(str, object) # file_path, AnalysisResult - analysisError = pyqtSignal(str, str) # file_path, error_message - progressUpdate = pyqtSignal(str, int) # message, percentage - - def __init__(self): - super().__init__() - self.results_cache = {} # Store analysis results - self.plotting_engine = PlottingEngine() - self.current_worker = None # Track active worker thread - self.logger = logging.getLogger(__name__) - - def analyze_file(self, file_path: str, window: int = 10, hop: int = 2): - """ - Analyze an audio file using background thread to prevent GUI freezing. - - Args: - file_path: Path to audio file - window: RMS analysis window size in seconds - hop: Analysis hop size in seconds - """ - if not os.path.exists(file_path): - error_msg = f"File not found: {file_path}" - self.logger.error(error_msg) - self.analysisError.emit(file_path, error_msg) - return - - # Stop any existing worker - if self.current_worker and self.current_worker.isRunning(): - self.logger.info("Stopping previous analysis to start new one") - self.current_worker.quit() - self.current_worker.wait() - - # Emit analysis started signal - self.analysisStarted.emit(file_path) - self.logger.info(f"Queuing analysis: {os.path.basename(file_path)}") - - # Create and start worker thread - self.current_worker = AudioAnalysisWorker(file_path, window, hop) - - # Connect worker signals - self.current_worker.progressUpdate.connect(self.progressUpdate.emit) - self.current_worker.analysisCompleted.connect(self._on_worker_completed) - self.current_worker.analysisError.connect(self.analysisError.emit) - - # Start the background analysis - self.current_worker.start() - - def _on_worker_completed(self, file_path: str, result: AnalysisResult): - """Handle completion of worker thread analysis.""" - # Cache the result - self.results_cache[file_path] = result - - # Forward the signal to GUI - self.analysisCompleted.emit(file_path, result) - - def get_analysis_figure(self, file_path: str): - """ - Get matplotlib figure for a previously analyzed file. - - Returns: - matplotlib.figure.Figure or None - """ - if file_path not in self.results_cache: - return None - - result = self.results_cache[file_path] - return self.plotting_engine.create_power_analysis_figure( - result.times, result.rms_array, result.file_path - ) - - def get_metadata_text(self, file_path: str) -> str: - """Get formatted metadata text for a file.""" - if file_path not in self.results_cache: - return "No analysis data available" - - result = self.results_cache[file_path] - return self.plotting_engine.create_metadata_display_text( - result.song_name, result.bpm, - result.max_amplitude, result.avg_amplitude - ) - - def clear_cache(self): - """Clear all cached analysis results.""" - self.results_cache.clear() - - def is_file_analyzed(self, file_path: str) -> bool: - """Check if a file has been analyzed.""" - return file_path in self.results_cache \ No newline at end of file + result = self.results_cache.get(file_path) + if result is None: + return None + metric = METRICS.get(metric_id) + if metric is None: + return None + data = result.metric_data.get(metric_id) + if data is None: + return None + return metric.render(data, file_path) + + def get_metadata_text(self, file_path: str) -> str: + result = self.results_cache.get(file_path) + if result is None: + return "No analysis data available" + return result.metadata_text() + + def clear_cache(self): + self.results_cache.clear() + + def is_file_analyzed(self, file_path: str) -> bool: + return file_path in self.results_cache diff --git a/font_control_widget.py b/font_control_widget.py index 191eb6a..d0a1eb7 100644 --- a/font_control_widget.py +++ b/font_control_widget.py @@ -5,8 +5,8 @@ Self-contained widget for easy layout management. import logging from typing import List, Dict, Optional -from PyQt5.QtWidgets import (QWidget, QComboBox, QVBoxLayout, QHBoxLayout, - QLabel, QSlider, QPushButton, QGroupBox) +from PyQt5.QtWidgets import (QWidget, QComboBox, QVBoxLayout, QHBoxLayout, + QLabel, QSlider, QGroupBox) from PyQt5.QtCore import pyqtSignal, Qt from font_manager import get_font_manager @@ -19,13 +19,11 @@ class FontControlWidget(QWidget): Provides clustered interface for: - Font selection (unified for Qt and matplotlib) - Qt font size adjustment - - Plot regeneration controls """ - + # Signals fontChanged = pyqtSignal(str, str) # (font_name, font_type) fontSizeChanged = pyqtSignal(int) # font_size - plotRefreshRequested = pyqtSignal() # manual refresh request def __init__(self, parent=None): """Initialize the font control widget.""" @@ -59,11 +57,7 @@ class FontControlWidget(QWidget): # Font size section size_section = self._create_font_size_section() group_layout.addWidget(size_section) - - # Control buttons section - button_section = self._create_button_section() - group_layout.addWidget(button_section) - + layout.addWidget(group_box) def _create_font_selector_section(self) -> QWidget: @@ -119,20 +113,6 @@ class FontControlWidget(QWidget): return section - def _create_button_section(self) -> QWidget: - """Create the control buttons section.""" - section = QWidget() - layout = QHBoxLayout(section) - layout.setContentsMargins(0, 0, 0, 0) - - # Refresh plot button - self.refresh_button = QPushButton("Refresh Plot") - self.refresh_button.setToolTip("Regenerate current plot with new font settings") - self.refresh_button.clicked.connect(self.on_refresh_plot_clicked) - layout.addWidget(self.refresh_button) - - return section - def refresh_font_list(self): """Refresh the list of available fonts.""" self.logger.debug("Refreshing font list...") @@ -243,11 +223,6 @@ class FontControlWidget(QWidget): # Emit signal for external listeners self.fontSizeChanged.emit(size) - def on_refresh_plot_clicked(self): - """Handle manual plot refresh button click.""" - self.logger.info("Manual plot refresh requested") - self.plotRefreshRequested.emit() - def _apply_font_change(self, font_name: str, font_type: str): """Apply the font change to both Qt and matplotlib.""" try: diff --git a/main.py b/main.py index 7b88382..67356cf 100644 --- a/main.py +++ b/main.py @@ -11,6 +11,7 @@ from analysis_results_manager import AnalysisResultsManager from logger_setup import setup_logging, parse_log_args from font_manager import initialize_fonts, get_font_manager from font_control_widget import FontControlWidget +from plot_control_widget import PlotControlWidget class MainWindow(QMainWindow): @@ -63,8 +64,13 @@ class MainWindow(QMainWindow): self.font_control = FontControlWidget() self.font_control.fontChanged.connect(self.on_font_changed) self.font_control.fontSizeChanged.connect(self.on_font_size_changed) - self.font_control.plotRefreshRequested.connect(self.on_plot_refresh_requested) layout.addWidget(self.font_control) + + # Plot control cluster (metric selector + refresh) + self.plot_control = PlotControlWidget() + self.plot_control.metricChanged.connect(self.on_metric_changed) + self.plot_control.plotRefreshRequested.connect(self.on_plot_refresh_requested) + layout.addWidget(self.plot_control) # File list self.file_list_label = QLabel("Analyzed Files:") @@ -99,6 +105,9 @@ class MainWindow(QMainWindow): self.analysis_manager.analysisCompleted.connect(self.on_analysis_completed) self.analysis_manager.analysisError.connect(self.on_analysis_error) self.analysis_manager.progressUpdate.connect(self.on_progress_update) + self.analysis_manager.metricComputeStarted.connect(self.on_metric_compute_started) + self.analysis_manager.metricReady.connect(self.on_metric_ready) + self.analysis_manager.metricComputeError.connect(self.on_metric_compute_error) def dragEnterEvent(self, event): """Handle drag enter event for file drops.""" @@ -124,7 +133,7 @@ class MainWindow(QMainWindow): # TODO: Add support for multiple file queue file_path = audio_files[0] self.logger.info(f"Starting analysis of dropped file: {os.path.basename(file_path)}") - self.analysis_manager.analyze_file(file_path) + self.analysis_manager.analyze_file(file_path, self.plot_control.current_metric_id()) else: self.visualization_widget.set_status("No audio files detected in drop") self.logger.warning("No supported audio files found in drop") @@ -140,7 +149,7 @@ class MainWindow(QMainWindow): if file_path: # User selected a file (didn't cancel) self.logger.info(f"File selected via dialog: {os.path.basename(file_path)}") - self.analysis_manager.analyze_file(file_path) + self.analysis_manager.analyze_file(file_path, self.plot_control.current_metric_id()) def on_analysis_started(self, file_path): """Called when analysis starts.""" @@ -150,30 +159,28 @@ class MainWindow(QMainWindow): def on_analysis_completed(self, file_path, result): """Called when analysis completes successfully.""" filename = os.path.basename(file_path) - + # Add to file list if not already there - existing_items = [self.file_list.item(i).text() + existing_items = [self.file_list.item(i).text() for i in range(self.file_list.count())] if filename not in existing_items: item = QListWidgetItem(filename) item.setData(Qt.UserRole, file_path) # Store full path self.file_list.addItem(item) - - # Get and display the analysis figure - figure = self.analysis_manager.get_analysis_figure(file_path) - if figure: - self.visualization_widget.display_figure_direct(figure) - + # Update metadata display metadata_text = self.analysis_manager.get_metadata_text(file_path) self.metadata_display.setText(metadata_text) - + # Select the analyzed file in the list for i in range(self.file_list.count()): item = self.file_list.item(i) if item.data(Qt.UserRole) == file_path: self.file_list.setCurrentItem(item) break + + # Render the currently-selected metric (cached, or async-compute it) + self._render_or_request(file_path) def on_analysis_error(self, file_path, error_message): """Called when analysis fails.""" @@ -189,54 +196,82 @@ class MainWindow(QMainWindow): def on_file_selected(self, item): """Called when a file is selected from the list.""" file_path = item.data(Qt.UserRole) - - # Display the analysis figure - figure = self.analysis_manager.get_analysis_figure(file_path) - if figure: - self.visualization_widget.display_figure_direct(figure) - + # Update metadata display metadata_text = self.analysis_manager.get_metadata_text(file_path) self.metadata_display.setText(metadata_text) + + # Render the currently-selected metric (cached, or async-compute it) + self._render_or_request(file_path) def on_font_changed(self, font_name: str, font_type: str): """Called when font selection changes.""" self.logger.info(f"Font changed via GUI: {font_name} ({font_type})") - # Auto-regenerate current plot with new font - self._regenerate_current_plot() - + # Cheap re-render — cached metric data, redraws under the new font. + self._render_or_request(self._current_file_path()) + def on_font_size_changed(self, font_size: int): """Called when Qt font size changes.""" self.logger.info(f"Qt font size changed via GUI: {font_size}pt") # Qt font size doesn't affect matplotlib plots, so no regeneration needed - + + def on_metric_changed(self, metric_id: str): + """Called when the metric selector changes.""" + self.logger.info(f"Metric changed via GUI: {metric_id}") + self._render_or_request(self._current_file_path()) + def on_plot_refresh_requested(self): """Called when manual plot refresh is requested.""" self.logger.info("Manual plot refresh requested via GUI") - self._regenerate_current_plot() - - def _regenerate_current_plot(self): - """Regenerate the current plot with updated font settings.""" - try: - # Get the currently selected file - current_item = self.file_list.currentItem() - if not current_item: - self.logger.debug("No file selected for plot regeneration") - return - - file_path = current_item.data(Qt.UserRole) - if not file_path: - self.logger.debug("No file path found for current selection") - return - - self.logger.info(f"Regenerating plot for: {os.path.basename(file_path)}") - - # Re-analyze the file to regenerate plots with new font - self.analysis_manager.analyze_file(file_path) - - except Exception as e: - self.logger.error(f"Error regenerating plot: {e}") - self.visualization_widget.set_status(f"Error regenerating plot: {e}") + self._render_or_request(self._current_file_path()) + + def on_metric_compute_started(self, file_path: str, metric_id: str): + """Called when an off-thread metric compute starts.""" + if file_path != self._current_file_path(): + return # selection moved on; status bar shouldn't lie + from metrics import METRICS + metric = METRICS.get(metric_id) + display = metric.display_name if metric else metric_id + self.visualization_widget.set_status(f"Computing {display}...") + + def on_metric_ready(self, file_path: str, metric_id: str): + """Called when metric data is available (cached hit or async finish).""" + if file_path != self._current_file_path(): + return # stale — user moved on + if metric_id != self.plot_control.current_metric_id(): + return # user already switched to a different metric + figure = self.analysis_manager.get_metric_figure(file_path, metric_id) + if figure: + self.visualization_widget.display_figure_direct(figure) + + def on_metric_compute_error(self, file_path: str, metric_id: str, error_message: str): + self.logger.error(f"Metric compute failed ({metric_id} / {os.path.basename(file_path)}): {error_message}") + if file_path == self._current_file_path(): + self.visualization_widget.set_status(f"Error computing {metric_id}: {error_message}") + + def _current_file_path(self): + item = self.file_list.currentItem() + return item.data(Qt.UserRole) if item else None + + def _render_or_request(self, file_path): + """Render the current metric from cache, or kick off async compute if missing. + + Falls back to a full analyse_file if the file hasn't been processed yet + (e.g. font change on an empty session — defensive). + """ + if not file_path: + return + metric_id = self.plot_control.current_metric_id() + figure = self.analysis_manager.get_metric_figure(file_path, metric_id) + if figure: + self.visualization_widget.display_figure_direct(figure) + return + # Not cached yet — try async compute if the file has been loaded. + if self.analysis_manager.is_file_analyzed(file_path): + self.analysis_manager.request_metric(file_path, metric_id) + else: + # No AudioFile yet either; kick off a full analysis with this metric. + self.analysis_manager.analyze_file(file_path, metric_id) def main(): diff --git a/metrics.py b/metrics.py new file mode 100644 index 0000000..a05bb8a --- /dev/null +++ b/metrics.py @@ -0,0 +1,231 @@ +""" +Pluggable analysis metrics. + +A `Metric` knows how to compute a series from an `AudioFile` and how to render +that series into a matplotlib `Figure`. Compute is the heavy step (runs on the +worker thread); render is cheap and reruns on font / refresh. + +To add a metric: subclass `Metric`, implement `compute` and `render`, and +register the instance in `METRICS` at the bottom of this file. +""" + +from __future__ import annotations + +import os +import warnings +from abc import ABC, abstractmethod +from typing import Any + +import numpy as np +import matplotlib.colors as mcolors +import matplotlib.cm as cm +from matplotlib.figure import Figure +import pyloudnorm as pyln + +from font_manager import safe_title +from master_core import AudioFile + + +class Metric(ABC): + """A pluggable analysis metric.""" + + id: str + display_name: str + + @abstractmethod + def compute(self, audio_file: AudioFile) -> Any: + """Compute and return the metric's data from a loaded AudioFile. + + The returned object is cached and later passed to `render`. This is the + heavy step and runs on the worker thread. + """ + + @abstractmethod + def render(self, data: Any, file_path: str, figsize=(10, 4)) -> Figure: + """Render a Figure from precomputed data. Cheap; runs on the GUI thread.""" + + +class RMSPowerMetric(Metric): + id = "rms_power" + display_name = "RMS Power" + + def __init__(self, window: int = 10, hop: int = 2): + self.window = window + self.hop = hop + + def compute(self, audio_file: AudioFile): + audio_file.get_energy_levels_over_time(window=self.window, hop=self.hop) + return { + "times": audio_file.get_times(), + "rms_array": audio_file.rms_array, + } + + def render(self, data, file_path, figsize=(10, 4)) -> Figure: + times = data["times"] + rms_array = data["rms_array"] + + # Adaptive colour scale: bump headroom for loud masters. + maxpower = 0.6 if np.max(rms_array) > 0.3 else 0.3 + norm = mcolors.Normalize(vmin=0, vmax=maxpower) + cmap = cm.autumn + + fig = Figure(figsize=figsize, facecolor="white") + ax = fig.add_subplot(111) + ax.set_ylim(0., maxpower) + for i in range(len(times) - 1): + ax.fill_between( + times[i:i + 2], 0, rms_array[0][i], + color=cmap(norm(rms_array[0][i])), edgecolor="none", + ) + sm = cm.ScalarMappable(cmap=cmap, norm=norm) + sm.set_array([]) + fig.colorbar(sm, ax=ax, label="RMS Power") + ax.set_ylabel("Power") + ax.set_xlabel("Time (seconds)") + ax.set_title(safe_title(os.path.basename(file_path))) + fig.tight_layout() + return fig + + +class WaveformMetric(Metric): + """Raw mono waveform with a min/max envelope downsample for plotting speed.""" + + id = "waveform" + display_name = "Waveform" + + def __init__(self, target_columns: int = 4000): + self.target_columns = target_columns + + def compute(self, audio_file: AudioFile): + y = audio_file.y_mono + sr = audio_file.sr + n = len(y) + if n <= self.target_columns: + times = np.arange(n) / sr + return {"times": times, "lo": y, "hi": y} + + chunk = n // self.target_columns + trimmed = y[: chunk * self.target_columns] + reshaped = trimmed.reshape(self.target_columns, chunk) + lo = reshaped.min(axis=1) + hi = reshaped.max(axis=1) + times = (np.arange(self.target_columns) * chunk + chunk / 2) / sr + return {"times": times, "lo": lo, "hi": hi} + + def render(self, data, file_path, figsize=(10, 4)) -> Figure: + times = data["times"] + lo = data["lo"] + hi = data["hi"] + + fig = Figure(figsize=figsize, facecolor="white") + ax = fig.add_subplot(111) + ax.fill_between(times, lo, hi, color="#3a7ad6", linewidth=0) + ax.axhline(0, color="black", linewidth=0.5, alpha=0.3) + # Fixed full-scale range with a touch of headroom for float-wav signals. + ax.set_ylim(-1.1, 1.1) + ax.set_xlim(times[0], times[-1]) + ax.set_ylabel("Amplitude") + ax.set_xlabel("Time (seconds)") + ax.set_title(safe_title(os.path.basename(file_path))) + fig.tight_layout() + return fig + + +class LUFSMetric(Metric): + """ITU-R BS.1770 loudness: short-term (3 s) time series + integrated value. + + Powered by pyloudnorm. The time series slides `meter.integrated_loudness` + across the track because pyloudnorm doesn't expose a per-block series. + Slightly redundant work, but the per-call cost is small. + """ + + id = "lufs" + display_name = "LUFS" + + # Short-term as defined by EBU R128 / BS.1770: 3-second window. + WINDOW_S = 3.0 + HOP_S = 0.5 + SILENCE_FLOOR = -70.0 # BS.1770 absolute gate + + def compute(self, audio_file: AudioFile): + y = audio_file.y_mono.astype(np.float64, copy=False) + sr = audio_file.sr + meter = pyln.Meter(sr) + + # pyloudnorm warns on clipping and on too-short audio; we handle both. + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + integrated = self._safe_integrated(meter, y) + + window_n = int(self.WINDOW_S * sr) + hop_n = int(self.HOP_S * sr) + + if len(y) < window_n: + # Track shorter than 3 s — just one data point at the centre. + times = np.array([len(y) / (2.0 * sr)]) + lufs = np.array([integrated if np.isfinite(integrated) else self.SILENCE_FLOOR]) + else: + n_windows = 1 + (len(y) - window_n) // hop_n + lufs = np.empty(n_windows) + for i in range(n_windows): + start = i * hop_n + lufs[i] = self._safe_integrated(meter, y[start:start + window_n]) + times = (np.arange(n_windows) * hop_n + window_n / 2.0) / sr + + lufs = np.where(np.isfinite(lufs), lufs, self.SILENCE_FLOOR) + lufs = np.clip(lufs, self.SILENCE_FLOOR, 0.0) + + return { + "times": times, + "lufs": lufs, + "integrated": float(integrated), + } + + @staticmethod + def _safe_integrated(meter: "pyln.Meter", segment: np.ndarray) -> float: + try: + return float(meter.integrated_loudness(segment)) + except (ValueError, FloatingPointError): + return float("-inf") + + def render(self, data, file_path, figsize=(10, 4)) -> Figure: + times = data["times"] + lufs = data["lufs"] + integrated = data["integrated"] + + fig = Figure(figsize=figsize, facecolor="white") + ax = fig.add_subplot(111) + ax.plot(times, lufs, color="#2a9d8f", linewidth=1.4, label="Short-term (3 s)") + + if np.isfinite(integrated): + ax.axhline( + integrated, color="#e76f51", linestyle="--", linewidth=1.5, + label=f"Integrated: {integrated:.1f} LUFS", + ) + + # Streaming target reference (Spotify normalises to -14 LUFS). + ax.axhline(-14.0, color="gray", linestyle=":", linewidth=0.8, alpha=0.6) + ax.text( + times[-1], -14.0, " -14 LUFS (streaming target)", + va="center", ha="left", fontsize=8, alpha=0.6, + ) + + ax.set_ylim(-50.0, 0.0) + ax.set_xlim(times[0], times[-1]) + ax.set_ylabel("LUFS") + ax.set_xlabel("Time (seconds)") + ax.set_title(safe_title(os.path.basename(file_path))) + ax.grid(True, alpha=0.3) + ax.legend(loc="lower right", fontsize=8) + fig.tight_layout() + return fig + + +METRICS: dict[str, Metric] = { + m.id: m for m in ( + RMSPowerMetric(), + WaveformMetric(), + LUFSMetric(), + ) +} +DEFAULT_METRIC_ID = "rms_power" diff --git a/plot_control_widget.py b/plot_control_widget.py new file mode 100644 index 0000000..81c76ff --- /dev/null +++ b/plot_control_widget.py @@ -0,0 +1,61 @@ +""" +Plot control widget: pick which metric to display and refresh the current plot. + +Mirrors FontControlWidget's clustered-groupbox style so the two sit naturally +next to each other in the left panel. +""" + +import logging +from PyQt5.QtWidgets import ( + QWidget, QVBoxLayout, QHBoxLayout, QLabel, QComboBox, QPushButton, QGroupBox, +) +from PyQt5.QtCore import pyqtSignal + +from metrics import METRICS, DEFAULT_METRIC_ID + + +class PlotControlWidget(QWidget): + """Metric selector + manual plot refresh.""" + + metricChanged = pyqtSignal(str) # metric_id + plotRefreshRequested = pyqtSignal() + + def __init__(self, parent=None): + super().__init__(parent) + self.logger = logging.getLogger(__name__) + self.initUI() + + def initUI(self): + layout = QVBoxLayout(self) + layout.setContentsMargins(5, 5, 5, 5) + + group_box = QGroupBox("Plot") + group_layout = QVBoxLayout(group_box) + + group_layout.addWidget(QLabel("Metric:")) + self.metric_combo = QComboBox() + for metric_id, metric in METRICS.items(): + self.metric_combo.addItem(metric.display_name, metric_id) + default_idx = self.metric_combo.findData(DEFAULT_METRIC_ID) + if default_idx >= 0: + self.metric_combo.setCurrentIndex(default_idx) + self.metric_combo.currentIndexChanged.connect(self._on_metric_changed) + group_layout.addWidget(self.metric_combo) + + button_row = QHBoxLayout() + self.refresh_button = QPushButton("Refresh Plot") + self.refresh_button.setToolTip("Re-render the current plot with current settings") + self.refresh_button.clicked.connect(self.plotRefreshRequested.emit) + button_row.addWidget(self.refresh_button) + group_layout.addLayout(button_row) + + layout.addWidget(group_box) + + def _on_metric_changed(self, _index: int): + metric_id = self.metric_combo.currentData() + if metric_id: + self.logger.info(f"Metric changed: {metric_id}") + self.metricChanged.emit(metric_id) + + def current_metric_id(self) -> str: + return self.metric_combo.currentData() or DEFAULT_METRIC_ID diff --git a/plotting_engine.py b/plotting_engine.py deleted file mode 100644 index dd85e37..0000000 --- a/plotting_engine.py +++ /dev/null @@ -1,80 +0,0 @@ -""" -Audio visualization plotting engine. -Separates plotting logic from audio processing for clean GUI integration. -""" - -import numpy as np -import matplotlib.pyplot as plt -import matplotlib.colors as mcolors -import matplotlib.cm as cm -from matplotlib.figure import Figure -import os -from font_manager import safe_title - - -class PlottingEngine: - """Handles all matplotlib visualization logic for audio analysis.""" - - @staticmethod - def create_power_analysis_figure(times, rms_array, file_path, figsize=(10, 4)): - """ - Creates a matplotlib Figure for power analysis visualization. - - Args: - times: Array of time points - rms_array: RMS power values over time - file_path: Path to the audio file for title - figsize: Figure size tuple - - Returns: - matplotlib.figure.Figure: Ready-to-embed figure - """ - # Determine color scale based on headroom detection - local_max_power = np.max(rms_array) - if local_max_power > 0.3: - norm = mcolors.Normalize(vmin=0, vmax=0.6) - maxpower = 0.6 - else: - norm = mcolors.Normalize(vmin=0, vmax=0.3) - maxpower = 0.3 - - # Create figure and axis - fig = Figure(figsize=figsize, facecolor='white') - ax = fig.add_subplot(111) - - # Color map - cmap = cm.autumn - - # Plot power levels as colored bars - ax.set_ylim(0., maxpower) - for i in range(len(times)-1): - ax.fill_between(times[i:i+2], 0, rms_array[0][i], - color=cmap(norm(rms_array[0][i])), edgecolor='none') - - # Add colorbar - sm = cm.ScalarMappable(cmap=cmap, norm=norm) - sm.set_array([]) - cbar = fig.colorbar(sm, ax=ax, label='RMS Power') - - # Labels and title - ax.set_ylabel('Power') - ax.set_xlabel('Time (seconds)') - ax.set_title(safe_title(os.path.basename(file_path))) - - # Tight layout for better appearance in GUI - fig.tight_layout() - - return fig - - @staticmethod - def create_metadata_display_text(song_name, bpm, max_amplitude, avg_amplitude): - """ - Creates formatted text for metadata display. - - Returns: - str: Formatted metadata text - """ - return f"""Track: {safe_title(song_name)} -BPM: {bpm:.1f} -Max Amplitude: {max_amplitude:.3f} -Avg Amplitude: {avg_amplitude:.3f}""" \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 43bebd2..c33e5ef 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,6 +11,7 @@ dependencies = [ "numpy", "matplotlib", "mutagen", + "pyloudnorm", "PyQt5>=5.15.10", # 5.15.2 is the only pyqt5-qt5 release with a Windows wheel; later # versions are Linux/macOS only. @@ -26,9 +27,10 @@ py-modules = [ "analysis_results_manager", "audio_visualization_widget", "master_core", - "plotting_engine", + "metrics", "font_manager", "font_control_widget", + "plot_control_widget", "logger_setup", "setup_fonts", ] diff --git a/uv.lock b/uv.lock index 82f6f29..f59ba63 100644 --- a/uv.lock +++ b/uv.lock @@ -1187,6 +1187,21 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0c/c3/44f3fbbfa403ea2a7c779186dc20772604442dde72947e7d01069cbe98e3/pycparser-3.0-py3-none-any.whl", hash = "sha256:b727414169a36b7d524c1c3e31839a521725078d7b2ff038656844266160a992", size = 48172, upload-time = "2026-01-21T14:26:50.693Z" }, ] +[[package]] +name = "pyloudnorm" +version = "0.2.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, + { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, + { name = "scipy", version = "1.15.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, + { name = "scipy", version = "1.17.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/23/00/f915eaa75326f4209941179c2b93ac477f2040e4aeff5bb21d16eb8058f9/pyloudnorm-0.2.0.tar.gz", hash = "sha256:8bf597658ea4e1975c275adf490f6deb5369ea409f2901f939915efa4b681b16", size = 14037, upload-time = "2026-01-04T11:43:35.265Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/aa/b6/65a49a05614b2548edbba3aab118f2ebe7441dfd778accdcdce9f6567f20/pyloudnorm-0.2.0-py3-none-any.whl", hash = "sha256:9bb69afb904f59d007a7f9ba3d75d16fb8aeef35c44d6df822a9f192d69cf13f", size = 10879, upload-time = "2026-01-04T11:43:34.534Z" }, +] + [[package]] name = "pyparsing" version = "3.3.2" @@ -1642,6 +1657,7 @@ dependencies = [ { name = "mutagen" }, { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, + { name = "pyloudnorm" }, { name = "pyqt5" }, { name = "pyqt5-qt5", marker = "sys_platform == 'win32'" }, ] @@ -1652,6 +1668,7 @@ requires-dist = [ { name = "matplotlib" }, { name = "mutagen" }, { name = "numpy" }, + { name = "pyloudnorm" }, { name = "pyqt5", specifier = ">=5.15.10" }, { name = "pyqt5-qt5", marker = "sys_platform == 'win32'", specifier = "==5.15.2" }, ]