cycle_detection.py

import wave

import numpy as np
import matplotlib.pyplot as plt


class LowPassFilter:
    def __init__(self, init=0.0, decay=0.9):
        self.b = 1 - decay
        self.y = init
        self.prev_sample = 0.0

    def process(self, sample):
        self.y += self.b * (0.5 * (sample + self.prev_sample) - self.y)
        self.prev_sample = sample
        return self.y
    
class SlewLimiter:
    def __init__(self, init=0.0, attack=0.9, decay=0.99):
        self.rise = 1 - attack
        self.fall = 1 - decay
        self.y = init

    def process(self, sample):
        if sample > self.y:
            self.y += self.rise * (sample - self.y)
        else:
            self.y += self.fall * (sample - self.y)
        return self.y

class CycleDetector:
    def __init__(self):
        self.attack = 0.9
        self.decay = 0.995
        self.env_high = SlewLimiter(attack=self.attack, decay=self.decay)
        self.env_low  = SlewLimiter(attack=self.decay, decay=self.attack)
        
        self.high_state = False
        self.low_state = False
        self.cycle_state = False
        
        self.cycle_counter = 0
        self.cycle_length = 100

    def process(self, sample):
        env_high_value = self.env_high.process(max(sample, 0))
        env_low_value  = self.env_low.process(min(sample, 0))

        high_off_to_on = False
        if not self.high_state and sample > env_high_value:
            self.high_state = True
            high_off_to_on = True
        if self.high_state and sample < env_high_value:
            self.high_state = False

        low_off_to_on = False
        if not self.low_state and sample < env_low_value:
            low_off_to_on = True
            self.low_state = True
        if self.low_state and sample > env_low_value:
            self.low_state = False

        if not self.cycle_state and high_off_to_on:
            self.cycle_state = True
            self.cycle_length = self.cycle_counter
            self.cycle_counter = 0
        if self.cycle_state and low_off_to_on:
            self.cycle_state = False
        
        self.cycle_counter += 1
        return self.cycle_length


def process(samples, samplerate):
    lpf = LowPassFilter(decay=0.99)
    cyc = CycleDetector()

    processed_history = np.zeros(samples.size)
    frq_history = np.zeros(samples.size)
    for i in range(samples.size):
        processed_history[i] = lpf.process(samples[i])
        cyc_length = cyc.process(processed_history[i])
        frq_history[i] = samplerate / cyc_length

    plt.plot(500 * samples, label="samples")
    plt.plot(300 * processed_history, label="processed")
    plt.plot(frq_history, label="freq")
    plt.ylim(0, 500)
    plt.legend()
    plt.show()

if __name__ == "__main__":
    with open("test.wav", "rb") as wav_file:
        wav = wave.open(wav_file)
        length = wav.getnframes()
        buffer = wav.readframes(length)
        samples = np.frombuffer(buffer, dtype='<h').astype(float) / 32767

        process(samples, wav.getframerate())