ideoforms / signalflow

A sound synthesis framework for Python, designed for clear and concise expression of complex musical ideas
https://signalflow.dev
MIT License
166 stars 14 forks source link

Pitch shift #120

Open paralin opened 2 weeks ago

paralin commented 2 weeks ago

How can I pitch shift a signal up or down with this library?

Say I want to pitch the right channel up:

def pitch_shift_right(input_file, output_file):
    graph = AudioGraph(output_device=AudioOut_Dummy(2))

    buffer = SignalFlowBuffer(input_file)
    duration = buffer.duration

    base_audio = BufferPlayer(buffer, loop=Node(False))

    left = ChannelSelect(base_audio, 0)
    right = ChannelSelect(base_audio, 1)

    # TODO: how do I pitch shift the right channel here?

    output = ChannelArray([left, right])

    output.play()
    output_buffer = Buffer(2, graph.sample_rate * int(duration))
    graph.render_to_buffer(output_buffer)
    output_buffer.save(output_file)
    graph.destroy()

What do I do on the "TODO" line to make the pitch shift on right?

Thanks!

paralin commented 2 weeks ago

I now realize this is harder than it sounds.

This works:

import wave
import numpy as np

# Open the input WAV file
with wave.open('input.wav', 'r') as wr:
    # Get the parameters of the input file
    params = wr.getparams()
    nchannels, sampwidth, framerate, nframes = params[:4]

    # Set the parameters for the output file
    output_params = list(params)
    output_params[3] = 0  # The number of samples will be set by writeframes
    output_params = tuple(output_params)

    # Open the output WAV file
    with wave.open('output.wav', 'w') as ww:
        ww.setparams(output_params)

        # Set the processing parameters
        frame_rate = 20
        chunk_size = framerate // frame_rate
        num_chunks = nframes // chunk_size
        pitch_shift = 8 // frame_rate

        # Process the audio in chunks
        for _ in range(num_chunks):
            # Read a chunk of audio data
            chunk = wr.readframes(chunk_size)
            data = np.frombuffer(chunk, dtype=np.int16)

            # Split the data into left and right channels
            left = data[0::2]
            right = data[1::2]

            # Perform pitch shifting on the right channel
            right_freq = np.fft.rfft(right)
            right_freq = np.roll(right_freq, pitch_shift)
            right_freq[:pitch_shift] = 0
            right_shifted = np.fft.irfft(right_freq)

            # Combine the left and right channels
            output_data = np.column_stack((left, right_shifted)).ravel().astype(np.int16)

            # Write the output data to the output file
            ww.writeframes(output_data.tobytes())

But how can I implement this with signalflow? Thanks!

paralin commented 2 weeks ago

This works too, also looking for how to do this with signalflow. Thanks again.

import numpy as np
import scipy.io.wavfile as wv
from scipy.signal import hilbert

def frequency_shift(data_array, shift_amount, sample_rate):
    analytic_signal = hilbert(data_array)
    instantaneous_phase = np.unwrap(np.angle(analytic_signal))
    new_phase = instantaneous_phase + 2.0 * np.pi * shift_amount / sample_rate * np.arange(data_array.size)
    new_signal = np.abs(analytic_signal) * np.cos(new_phase)
    return new_signal

# The normalization function
def normalize(data_array):
    return np.int16(data_array/np.max(np.abs(data_array)) * 32767)

def apply_effect(input_file, output_file, target_frequency=8.0):
    # Load the audio file using scipy
    sample_rate, data = wv.read(input_file)

    right_shift = float(target_frequency) / 2.0
    left_shift = -1.0 * right_shift

    # Split the stereo audio into left and right channels
    left = data[:, 0]
    right = data[:, 1]

    # Perform frequency shift on the right channel
    right_shifted = frequency_shift(right, right_shift, sample_rate)

    # Normalize the output to prevent clipping
    right_shifted = normalize(right_shifted)

    # Perform frequency shift on the left channel
    left_shifted = frequency_shift(left, left_shift, sample_rate)

    # Normalize the output to prevent clipping
    left_shifted = normalize(left_shifted)

    # Merge the left and right_shifted channels
    output_data = np.column_stack((left_shifted, right_shifted))

    # Write the output to file
    wv.write(output_file, sample_rate, output_data.astype(np.int16))