From f4e106cfc995972e2bae209fa4af884b918ceb96 Mon Sep 17 00:00:00 2001 From: RyanOnTheInside <7623207+ryanontheinside@users.noreply.github.com> Date: Thu, 30 Jan 2025 07:56:28 -0500 Subject: [PATCH] draft --- README.MD | 75 ++++++++++++ base/stream_base.py | 110 +++++++++++++++++ controls/audio_controls.py | 198 +++++++++++++++++++++++++++++++ controls/audio_stream_adapter.py | 48 ++++++++ notes.txt | 88 ++++++++++++++ processors/audio_processor.py | 171 ++++++++++++++++++++++++++ requirements.txt | 5 +- 7 files changed, 694 insertions(+), 1 deletion(-) create mode 100644 base/stream_base.py create mode 100644 controls/audio_controls.py create mode 100644 controls/audio_stream_adapter.py create mode 100644 notes.txt create mode 100644 processors/audio_processor.py diff --git a/README.MD b/README.MD index 0142302..187cb42 100644 --- a/README.MD +++ b/README.MD @@ -29,6 +29,81 @@ The intention for this repository is to build a suite of nodes that can be used - **SimilarityFilter**: Filter out similar consecutive images and control downstream execution. Perfect for optimizing real-time workflows by skipping redundant processing of similar frames. - **LazyCondition**: Powerful conditional execution node that supports any input type. Uses lazy evaluation to truly skip execution of unused paths and maintains state to avoid feedback loops. +## Audio-Reactive Controls + +The RealTime Nodes pack now includes audio-reactive controls that can respond to live audio input. These nodes allow you to create dynamic animations and effects that respond to music or other audio sources. + +### Available Audio Controls + +1. **Volume Control** + - Reacts to overall audio volume (RMS) + - Smoothly interpolates between min and max values + - Perfect for size, opacity, or intensity controls + +2. **Frequency Band Control** + - Analyzes specific frequency ranges (bass, mids, treble) + - 32 frequency bands available + - Great for creating frequency-specific animations + +3. **Beat Control** + - Detects beats in the audio + - Outputs trigger values on beat detection + - Useful for synchronizing effects with music + +### Audio Features + +All audio processing is done in real-time with the following specifications: +- 48kHz sample rate +- 16-bit audio depth +- Mono channel input +- 500ms default buffer size (configurable) +- Mel-scaled frequency analysis +- Onset detection for beat tracking + +### Performance Considerations + +- Audio processing is optimized using numpy and scipy +- Shared feature cache reduces redundant calculations +- Smoothing options available for all controls +- Thread-safe implementation for real-time processing + +### Example Usage + +```python +# Connect a Volume Control to an image scale: +volume_control = VolumeControl() +volume_control.update( + processor="audio", + feature="rms", + smoothing=0.5, + min_value=0.5, + max_value=2.0 +) + +# Use frequency bands to control color: +freq_control = FrequencyBandControl() +freq_control.update( + processor="audio", + feature="spectrum", + band_start=0, # Bass frequencies + band_end=4, + smoothing=0.3, + min_value=0.0, + max_value=1.0 +) + +# Trigger effects on beats: +beat_control = BeatControl() +beat_control.update( + processor="audio", + feature="onset", + threshold=0.5, + trigger_value=1.0, + fallback_value=0.0, + smoothing=0.2 +) +``` + ## Movement Patterns 🔄 All value and motion controls support various movement patterns: diff --git a/base/stream_base.py b/base/stream_base.py new file mode 100644 index 0000000..f011ca2 --- /dev/null +++ b/base/stream_base.py @@ -0,0 +1,110 @@ +from abc import ABC, abstractmethod +import numpy as np +from typing import Dict, Any, Optional +import threading +from dataclasses import dataclass, field +from collections import defaultdict +from .control_base import ControlNodeBase + +@dataclass +class FeatureSpec: + """Specification for a stream feature""" + type: str # e.g. "FLOAT", "TENSOR", "INT" + shape: tuple = field(default_factory=tuple) # () for scalar, (N,) for vector, etc. + range: tuple = (0.0, 1.0) # min/max range for the feature + description: str = "" + +class StreamDataProcessor(ABC): + """Base class for processing any type of streaming data""" + + def __init__(self): + self._feature_cache = {} + self._cache_lock = threading.Lock() + self._registered_features = self.get_feature_specs() + + @abstractmethod + def get_feature_specs(self) -> Dict[str, FeatureSpec]: + """Return specifications for all available features""" + pass + + @abstractmethod + def process_buffer(self, buffer: Any) -> Dict[str, Any]: + """Process incoming buffer and return computed features""" + pass + + def get_feature(self, name: str) -> Optional[Any]: + """Get a feature value from cache""" + with self._cache_lock: + return self._feature_cache.get(name) + + def update_cache(self, features: Dict[str, Any]): + """Update feature cache with new values""" + with self._cache_lock: + self._feature_cache.update(features) + +class StreamProcessorRegistry: + """Global registry for stream processors""" + _processors: Dict[str, StreamDataProcessor] = {} + _feature_cache: Dict[str, Dict[str, Any]] = defaultdict(dict) + _lock = threading.Lock() + + @classmethod + def register_processor(cls, name: str, processor: StreamDataProcessor): + with cls._lock: + cls._processors[name] = processor + + @classmethod + def get_processor(cls, name: str) -> Optional[StreamDataProcessor]: + with cls._lock: + return cls._processors.get(name) + + @classmethod + def get_feature(cls, processor_name: str, feature_name: str) -> Optional[Any]: + with cls._lock: + return cls._feature_cache[processor_name].get(feature_name) + + @classmethod + def update_features(cls, processor_name: str, features: Dict[str, Any]): + with cls._lock: + cls._feature_cache[processor_name].update(features) + +class StreamControl(ControlNodeBase): + """Base class for nodes that react to streaming data""" + + def __init__(self): + super().__init__() + self._last_update = 0.0 + + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + **super().INPUT_TYPES()["required"], + "processor": ("STRING", { + "default": "audio", + "tooltip": "Type of stream processor to use" + }), + "feature": ("STRING", { + "default": "rms", + "tooltip": "Name of the feature to use" + }), + "smoothing": ("FLOAT", { + "default": 0.5, + "min": 0.0, + "max": 1.0, + "step": 0.01, + "tooltip": "Smoothing factor for feature values" + }) + } + } + + def get_feature_value(self, processor_name: str, feature_name: str, + default: Any = 0.0) -> Any: + """Get current value for a feature""" + value = StreamProcessorRegistry.get_feature(processor_name, feature_name) + return default if value is None else value + + def smooth_value(self, current: float, target: float, + smoothing: float) -> float: + """Apply smoothing to a value""" + return current * smoothing + target * (1 - smoothing) \ No newline at end of file diff --git a/controls/audio_controls.py b/controls/audio_controls.py new file mode 100644 index 0000000..1577928 --- /dev/null +++ b/controls/audio_controls.py @@ -0,0 +1,198 @@ +from ..base.stream_base import StreamControl +from ..processors.audio_processor import AudioProcessor +import numpy as np + +class AudioFeatureControl(StreamControl): + """Base class for audio-reactive controls""" + + CATEGORY = "real-time/control/audio" + + @classmethod + def IS_CHANGED(cls, **kwargs): + return float(time.time()) # Always update to get latest audio features + +class VolumeControl(AudioFeatureControl): + """Control node that reacts to audio volume (RMS)""" + + DESCRIPTION = "Outputs a value based on audio volume" + RETURN_TYPES = ("FLOAT",) + FUNCTION = "update" + + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + **super().INPUT_TYPES()["required"], + "min_value": ("FLOAT", { + "default": 0.0, + "min": -10000.0, + "max": 10000.0, + "step": 0.1 + }), + "max_value": ("FLOAT", { + "default": 1.0, + "min": -10000.0, + "max": 10000.0, + "step": 0.1 + }) + } + } + + def update(self, processor: str, feature: str, smoothing: float, + min_value: float, max_value: float, always_execute: bool = True): + # Get current RMS value + rms = self.get_feature_value(processor, "rms", 0.0) + + # Get previous state + state = self.get_state({"value": min_value}) + current = state["value"] + + # Apply smoothing + smoothed = self.smooth_value(current, rms, smoothing) + + # Map to output range + output = np.interp(smoothed, [0.0, 1.0], [min_value, max_value]) + + # Update state + self.set_state({"value": output}) + + return (float(output),) + +class FrequencyBandControl(AudioFeatureControl): + """Control node that reacts to specific frequency bands""" + + DESCRIPTION = "Outputs a value based on energy in specific frequency bands" + RETURN_TYPES = ("FLOAT",) + FUNCTION = "update" + + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + **super().INPUT_TYPES()["required"], + "band_start": ("INT", { + "default": 0, + "min": 0, + "max": 31, + "step": 1, + "tooltip": "Start index of frequency band (0-31)" + }), + "band_end": ("INT", { + "default": 4, + "min": 0, + "max": 31, + "step": 1, + "tooltip": "End index of frequency band (0-31)" + }), + "min_value": ("FLOAT", { + "default": 0.0, + "min": -10000.0, + "max": 10000.0, + "step": 0.1 + }), + "max_value": ("FLOAT", { + "default": 1.0, + "min": -10000.0, + "max": 10000.0, + "step": 0.1 + }) + } + } + + def update(self, processor: str, feature: str, smoothing: float, + band_start: int, band_end: int, min_value: float, max_value: float, + always_execute: bool = True): + # Get spectrum + spectrum = self.get_feature_value(processor, "spectrum", + np.zeros(32)) + + # Calculate average energy in band range + band_energy = np.mean(spectrum[band_start:band_end+1]) + + # Get previous state + state = self.get_state({"value": min_value}) + current = state["value"] + + # Apply smoothing + smoothed = self.smooth_value(current, band_energy, smoothing) + + # Map to output range + output = np.interp(smoothed, [0.0, 1.0], [min_value, max_value]) + + # Update state + self.set_state({"value": output}) + + return (float(output),) + +class BeatControl(AudioFeatureControl): + """Control node that detects beats in audio""" + + DESCRIPTION = "Outputs a trigger value on detected beats" + RETURN_TYPES = ("FLOAT",) + FUNCTION = "update" + + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + **super().INPUT_TYPES()["required"], + "threshold": ("FLOAT", { + "default": 0.5, + "min": 0.0, + "max": 1.0, + "step": 0.01, + "tooltip": "Onset detection threshold" + }), + "trigger_value": ("FLOAT", { + "default": 1.0, + "min": -10000.0, + "max": 10000.0, + "step": 0.1, + "tooltip": "Value to output when beat detected" + }), + "fallback_value": ("FLOAT", { + "default": 0.0, + "min": -10000.0, + "max": 10000.0, + "step": 0.1, + "tooltip": "Value to output when no beat" + }) + } + } + + def update(self, processor: str, feature: str, smoothing: float, + threshold: float, trigger_value: float, fallback_value: float, + always_execute: bool = True): + # Get onset strength + onset = self.get_feature_value(processor, "onset", 0.0) + + # Get previous state + state = self.get_state({ + "value": fallback_value, + "last_trigger_time": 0.0 + }) + + current_time = time.time() + output = fallback_value + + # Check if we should trigger + if onset > threshold: + # Ensure minimum time between triggers (100ms) + if current_time - state["last_trigger_time"] > 0.1: + output = trigger_value + state["last_trigger_time"] = current_time + + # Apply smoothing + smoothed = self.smooth_value(state["value"], output, smoothing) + state["value"] = smoothed + + # Update state + self.set_state(state) + + return (float(smoothed),) + +# Register the audio processor +StreamProcessorRegistry.register_processor( + "audio", + AudioProcessor(sample_rate=48000) +) \ No newline at end of file diff --git a/controls/audio_stream_adapter.py b/controls/audio_stream_adapter.py new file mode 100644 index 0000000..f525097 --- /dev/null +++ b/controls/audio_stream_adapter.py @@ -0,0 +1,48 @@ +from ..base.stream_base import StreamProcessorRegistry +from ..processors.audio_processor import AudioProcessor +import numpy as np + +class AudioStreamAdapter: + """Adapter node that connects ComfyStream audio to our processor""" + + CATEGORY = "real-time/control/audio" + RETURN_TYPES = ("AUDIO",) + FUNCTION = "update" + + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "audio": ("AUDIO",), + "buffer_size_ms": ("FLOAT", { + "default": 500.0, + "min": 20.0, # Minimum one frame + "max": 2000.0, # 2 seconds max + "step": 20.0, + "tooltip": "Audio buffer size in milliseconds" + }) + } + } + + def __init__(self): + self.processor = None + + def update(self, audio: np.ndarray, buffer_size_ms: float): + # Initialize processor if needed or if buffer size changed + if (self.processor is None or + self.processor.buffer_size_samples != int(buffer_size_ms * 48)): + self.processor = AudioProcessor( + sample_rate=48000, + buffer_size_ms=buffer_size_ms + ) + # Register/update the processor + StreamProcessorRegistry.register_processor("audio", self.processor) + + # Process the audio buffer + if audio is not None: + features = self.processor.process_buffer(audio) + # Update the global feature cache + StreamProcessorRegistry.update_features("audio", features) + + # Pass through the audio buffer + return (audio,) \ No newline at end of file diff --git a/notes.txt b/notes.txt new file mode 100644 index 0000000..2fde581 --- /dev/null +++ b/notes.txt @@ -0,0 +1,88 @@ +ComfyStream Audio Format Specification +Basic Properties +| Property | Value | Notes | +|----------|--------|-------| +| Sample Rate | 48kHz (48000 Hz) | Standard for WebRTC | +| Bit Depth | 16-bit signed | Range: -32768 to 32767 | +| Channels | Mono | Converted via resampler | +| Frame Size | 960 samples | 20ms chunks (48000 0.02) | +Data Flow +Input: Audio comes from WebRTC through AudioStreamTrack +Preprocessing: +Resampled to mono using av.audio.resampler.AudioResampler +Converted to s16 format +Flattened to 1D numpy array +Node Interface +When writing audio processing nodes, you'll receive: +Type: Numpy array (np.ndarray) +Shape: 1-dimensional (N,) where N depends on buffer size +Data Type: np.int16 +Default Buffer: 500ms = 24000 samples (configurable via buffer_size parameter) + + +nodes for ComfyStream interface: + +import numpy as np + +from comfystream import tensor_cache + +class LoadAudioTensor: + CATEGORY = "audio_utils" + RETURN_TYPES = ("AUDIO",) + FUNCTION = "execute" + + def __init__(self): + self.audio_buffer = np.array([]) + + @classmethod + def INPUT_TYPES(s): + return { + "required": { + "buffer_size": ("FLOAT", {"default": 500.0}) + } + } + + @classmethod + def IS_CHANGED(): + return float("nan") + + def execute(self, buffer_size): + audio = tensor_cache.audio_inputs.get(block=True) + self.audio_buffer = np.concatenate((self.audio_buffer, audio)) + + buffer_size_samples = int(buffer_size * 48) + + if self.audio_buffer.size >= buffer_size_samples: + buffered_audio = self.audio_buffer[:buffer_size_samples] + self.audio_buffer = self.audio_buffer[buffer_size_samples:] + return (buffered_audio,) + else: + return (None,) + + + +from comfystream import tensor_cache + + +class SaveAudioTensor: + CATEGORY = "audio_utils" + RETURN_TYPES = () + FUNCTION = "execute" + OUTPUT_NODE = True + + @classmethod + def INPUT_TYPES(s): + return { + "required": { + "audio": ("AUDIO",), + } + } + + @classmethod + def IS_CHANGED(s): + return float("nan") + + def execute(self, audio): + fut = tensor_cache.audio_outputs.get() + fut.set_result((audio)) + return (audio,) diff --git a/processors/audio_processor.py b/processors/audio_processor.py new file mode 100644 index 0000000..b83ed1f --- /dev/null +++ b/processors/audio_processor.py @@ -0,0 +1,171 @@ +import numpy as np +from scipy import signal +from typing import Dict, Any +from ..base.stream_base import StreamDataProcessor, FeatureSpec + +class AudioProcessor(StreamDataProcessor): + """Processor for real-time audio stream data""" + + def __init__(self, sample_rate: int = 48000, + buffer_size_ms: float = 500.0, + n_bands: int = 32, + min_freq: float = 20.0, + max_freq: float = 20000.0): + super().__init__() + self.sample_rate = sample_rate + self.buffer_size_samples = int(buffer_size_ms * (sample_rate / 1000)) + self.n_bands = n_bands + self.min_freq = min_freq + self.max_freq = max_freq + + # Pre-compute frequency bands + self.freq_bands = self._get_mel_bands() + + # Initialize FFT config - ensure it's not larger than buffer + self.n_fft = min(2048, self.buffer_size_samples) + self.hop_length = self.n_fft // 4 + + # Window function for FFT + self.window = signal.windows.hann(self.n_fft) + + # Smoothing state + self.prev_spectrum = np.zeros(self.n_bands) + self.prev_rms = 0.0 + + def get_feature_specs(self) -> Dict[str, FeatureSpec]: + """Define available audio features""" + return { + "rms": FeatureSpec( + type="FLOAT", + range=(0.0, 1.0), + description="Root mean square amplitude" + ), + "spectrum": FeatureSpec( + type="TENSOR", + shape=(self.n_bands,), + range=(0.0, 1.0), + description="Mel-scaled frequency spectrum" + ), + "onset": FeatureSpec( + type="FLOAT", + range=(0.0, 1.0), + description="Onset detection strength" + ), + "pitch": FeatureSpec( + type="FLOAT", + range=(0.0, 1.0), + description="Estimated dominant pitch" + ) + } + + def _get_mel_bands(self) -> np.ndarray: + """Create mel-scaled frequency bands""" + return librosa.mel_frequencies( + n_mels=self.n_bands, + fmin=self.min_freq, + fmax=self.max_freq, + htk=True + ) + + def _compute_spectrum(self, buffer: np.ndarray) -> np.ndarray: + """Compute mel-scaled spectrum""" + # Pad buffer if needed + if len(buffer) < self.n_fft: + buffer = np.pad(buffer, (0, self.n_fft - len(buffer))) + + # Compute STFT + stft = np.abs(librosa.stft( + buffer, + n_fft=self.n_fft, + hop_length=self.hop_length, + window=self.window, + center=False + )) + + # Convert to mel scale + mel = librosa.feature.melspectrogram( + S=stft, + sr=self.sample_rate, + n_mels=self.n_bands, + fmin=self.min_freq, + fmax=self.max_freq + ) + + # Convert to dB scale and normalize + mel_db = librosa.power_to_db(mel, ref=np.max) + mel_norm = (mel_db - mel_db.min()) / (mel_db.max() - mel_db.min()) + + # Average across time + return np.mean(mel_norm, axis=1) + + def _compute_onset(self, buffer: np.ndarray) -> float: + """Detect onset strength""" + onset_env = librosa.onset.onset_strength( + y=buffer, + sr=self.sample_rate, + hop_length=self.hop_length + ) + return float(np.mean(onset_env)) + + def _compute_pitch(self, buffer: np.ndarray) -> float: + """Estimate dominant pitch""" + pitches, magnitudes = librosa.piptrack( + y=buffer, + sr=self.sample_rate, + n_fft=self.n_fft, + hop_length=self.hop_length, + fmin=self.min_freq, + fmax=self.max_freq + ) + + # Get pitch with highest magnitude + pitch_idx = magnitudes.argmax() + return float(pitches.flatten()[pitch_idx]) + + def process_buffer(self, buffer: np.ndarray) -> Dict[str, Any]: + """Process audio buffer and compute features""" + # Validate buffer + if buffer is None: + return { + "rms": 0.0, + "spectrum": np.zeros(self.n_bands), + "onset": 0.0, + "pitch": 0.0 + } + + # Ensure buffer is the right type and normalized + if buffer.dtype != np.int16: + raise ValueError(f"Expected int16 buffer, got {buffer.dtype}") + buffer = buffer.astype(np.float32) / np.iinfo(np.int16).max + + # Ensure buffer size + if len(buffer) != self.buffer_size_samples: + if len(buffer) < self.buffer_size_samples: + buffer = np.pad(buffer, (0, self.buffer_size_samples - len(buffer))) + else: + buffer = buffer[:self.buffer_size_samples] + + # Compute RMS + rms = np.sqrt(np.mean(buffer**2)) + + # Compute spectrum + spectrum = self._compute_spectrum(buffer) + + # Apply temporal smoothing + spectrum = 0.5 * spectrum + 0.5 * self.prev_spectrum + rms = 0.5 * rms + 0.5 * self.prev_rms + + # Update previous values + self.prev_spectrum = spectrum + self.prev_rms = rms + + # Compute additional features + onset = self._compute_onset(buffer) + pitch = self._compute_pitch(buffer) + + return { + "rms": float(rms), + "spectrum": spectrum, + "onset": float(onset), + "pitch": float(pitch) + } \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 3c73c1b..90dbc2f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,5 @@ opencv-python -Pillow \ No newline at end of file +Pillow +numpy>=1.21.0 +scipy>=1.7.0 +librosa>=0.9.0 \ No newline at end of file