diff --git a/src/dsp/pipeline.py b/src/dsp/pipeline.py index 57532b1..0798894 100644 --- a/src/dsp/pipeline.py +++ b/src/dsp/pipeline.py @@ -319,6 +319,16 @@ class AudioPipeline: # Smoothing factor: ~50ms time constant at 48kHz/256 block self._vu_alpha: float = np.exp(-BLOCK_SIZE / (0.05 * SAMPLE_RATE)) + # ── Tuner / pitch detection state ──────────────────────────────── + self._tuner_frequency: float = 0.0 # detected fundamental freq (Hz) + self._tuner_note: str = "--" # closest note name + self._tuner_cents: float = 0.0 # cent deviation from closest note + self._tuner_string: int = -1 # string number (1-6) or -1 if not matched + self._tuner_confidence: float = 0.0 # 0.0 to 1.0 + # Pitch detection buffer (keep last N samples for analysis) + self._pitch_buffer: np.ndarray = np.array([], dtype=np.float32) + self._pitch_buffer_max: int = 2048 # ~43ms at 48kHz + logger.info("Audio pipeline initialized (block=%d, sr=%d)", BLOCK_SIZE, SAMPLE_RATE) @@ -370,6 +380,26 @@ class AudioPipeline: Mono mode: shape (N,) — processed output. 4CM mode: shape (2, N) — [send_out, return_out]. """ + # ── Tuner mode: mute output, keep input tracking for pitch detection ── + if self._tuner_enabled: + # Still track input level for tuner display + if audio_in.ndim == 1: + in_rms = np.sqrt(np.mean(audio_in ** 2) + _EPS) + self._input_level = ( + self._input_level * self._vu_alpha + + in_rms * (1.0 - self._vu_alpha) + ) + else: + ch0 = audio_in[0, :] + in_rms = np.sqrt(np.mean(ch0 ** 2) + _EPS) + self._input_level = ( + self._input_level * self._vu_alpha + + in_rms * (1.0 - self._vu_alpha) + ) + # Run pitch detection on input + self._detect_pitch(audio_in) + return np.zeros_like(audio_in) + if self._bypassed: return audio_in * self._master_volume @@ -378,6 +408,139 @@ class AudioPipeline: else: return self._process_mono(audio_in) + # ── Pitch detection for tuner ────────────────────────────────────────────── + + # Standard tuning frequencies (E2, A2, D3, G3, B3, E4) — guitar strings + _STRING_FREQS = [82.41, 110.0, 146.83, 196.0, 246.94, 329.63] + _STRING_NAMES = ["E", "A", "D", "G", "B", "e"] + + # Note names in chromatic order (C = 0) + _NOTE_NAMES = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"] + + def _detect_pitch(self, audio_in: np.ndarray) -> None: + """Run pitch detection on the input buffer using autocorrelation. + + Updates ``self._tuner_frequency``, ``self._tuner_note``, + ``self._tuner_cents``, and ``self._tuner_string``. + + Args: + audio_in: Input audio block — mono (N,) or stereo (2, N). + """ + # Extract mono channel + if audio_in.ndim == 2: + signal = audio_in[0, :].copy() + else: + signal = audio_in.copy() + + # Append to rolling pitch buffer + self._pitch_buffer = np.concatenate([self._pitch_buffer, signal]) + if len(self._pitch_buffer) > self._pitch_buffer_max: + self._pitch_buffer = self._pitch_buffer[-self._pitch_buffer_max:] + + # Need enough signal for meaningful analysis + if len(self._pitch_buffer) < 512: + self._tuner_confidence = 0.0 + return + + # Simple autocorrelation pitch detection + buf = self._pitch_buffer + # Remove DC offset + buf = buf - np.mean(buf) + + # Check if there's enough amplitude + rms = np.sqrt(np.mean(buf ** 2)) + if rms < 0.002: # Silence threshold + self._tuner_confidence = 0.0 + self._tuner_frequency = 0.0 + self._tuner_note = "--" + self._tuner_cents = 0.0 + self._tuner_string = -1 + return + + # Autocorrelation: find the fundamental period + # Search lag range: 30 to 1024 samples (46.9Hz to 1600Hz at 48kHz) + min_lag = int(SAMPLE_RATE / 1600) # ~30 + max_lag = min(int(SAMPLE_RATE / 50), len(buf) // 2) # ~960 + + if max_lag <= min_lag: + self._tuner_confidence = 0.0 + return + + corr = np.correlate(buf, buf, mode='full') + # Take only the second half (positive lags) + corr = corr[len(corr) // 2:] + + # Normalize by energy at each lag (YIN-style) + energy = np.cumsum(buf ** 2) + energy = energy[max(1, min_lag):max_lag + len(buf) - len(corr) + min_lag + 1] + # Fallback: use simple autocorrelation + lag_slice = corr[min_lag:max_lag + 1] + + # Find the first peak in the autocorrelation + diffs = np.diff(lag_slice) + # Look for zero crossings in diff (peaks: positive→negative) + peaks = [] + for i in range(1, len(diffs)): + if diffs[i-1] > 0 and diffs[i] <= 0: + peaks.append((min_lag + i, lag_slice[i])) + + if not peaks: + self._tuner_confidence = 0.0 + return + + # Pick the strongest peak + best_lag, best_val = max(peaks, key=lambda x: x[1]) + + # Confidence based on relative peak strength + noise_floor = np.mean(np.abs(corr[min_lag:max_lag + 1])) + confidence = best_val / (noise_floor + 1e-10) + # Parabolic interpolation for sub-sample accuracy + if best_lag > min_lag and best_lag < max_lag: + idx = best_lag - min_lag + if 0 < idx < len(lag_slice) - 1: + y0, y1, y2 = lag_slice[idx-1], lag_slice[idx], lag_slice[idx+1] + if y0 + y2 - 2 * y1 != 0: + correction = (y0 - y2) / (2 * (y0 + y2 - 2 * y1)) + best_lag = best_lag + correction + + # Fundamental frequency + freq = SAMPLE_RATE / best_lag if best_lag > 0 else 0 + + # Clip confidence to 0-1 range + self._tuner_confidence = min(1.0, max(0.0, confidence / 10.0)) + + if freq < 30 or freq > 1600: + self._tuner_confidence = 0.0 + return + + self._tuner_frequency = freq + + # ── Convert frequency to note name ── + # A4 = 440Hz, MIDI note 69 + midi_note = 12 * np.log2(freq / 440.0) + 69 + midi_rounded = round(midi_note) + cents = int(100 * (midi_note - midi_rounded)) + + # Clamp to valid MIDI range + if midi_rounded < 0 or midi_rounded > 127: + self._tuner_note = "--" + self._tuner_confidence = 0.0 + return + + octave = (midi_rounded // 12) - 1 + note_idx = midi_rounded % 12 + note_name = self._NOTE_NAMES[note_idx] + self._tuner_note = f"{note_name}{octave}" + self._tuner_cents = cents + + # ── Guess guitar string ── + self._tuner_string = -1 + for si, sf in enumerate(self._STRING_FREQS): + # +/- 3 semitones from the string's fundamental + if abs(freq - sf) / sf < 0.2: + self._tuner_string = si + 1 + break + def _process_mono(self, audio_in: np.ndarray) -> np.ndarray: """Process a mono block through the full chain (all blocks).""" # Update input VU level (RMS with envelope smoothing)