Implement audio.moving_average_filter()

Author: skoolkid

skoolkit/audio.py

@@ -27,6 +27,31 @@
 INTERRUPT_DELAY = 'InterruptDelay'
 SAMPLE_RATE = 'SampleRate'
 
+def moving_average_filter(delays, options, volume=1):
+    sample_delay = options[CLOCK_SPEED] / options[SAMPLE_RATE]
+    s0, s1 = 0, sample_delay
+    t, t1 = 0, ceil(s1)
+    bit = bits = 0
+    samples = []
+    for d in delays:
+        while True:
+            if t + d < t1:
+                if bit:
+                    bits += d
+                t += d
+                break
+            i = t1 - t
+            if bit:
+                bits += i
+            d -= i
+            samples.append(volume * bits / (t1 - s0))
+            s0 = t = t1
+            s1 += sample_delay
+            t1 = ceil(s1)
+            bits = 0
+        bit = 1 - bit
+    return samples
+
 def write_wav(audio_file, samples, sample_rate, channels=1):
     bits_per_sample = 16
     bytes_per_sample = (bits_per_sample // 8) * channels
@@ -102,7 +127,7 @@ def write_audio(self, audio_file, delays, contention=False, interrupts=False, of
         if contention or interrupts:
             self._add_contention(delays, contention, interrupts, offset, options)
         if ma_filter:
-            samples = self._moving_average_filter(delays, options)
+            samples = moving_average_filter(delays, options)
         else:
             samples = self._delays_to_samples(delays, options)
         write_wav(audio_file, samples, options[SAMPLE_RATE])
@@ -164,31 +189,6 @@ def _add_contention(self, delays, contention, interrupts, cycle, options):
                     d_offset += f_duration - cycle
                     cycle = 0
 
-    def _moving_average_filter(self, delays, options):
-        sample_delay = options[CLOCK_SPEED] / options[SAMPLE_RATE]
-        s0, s1 = 0, sample_delay
-        t, t1 = 0, ceil(s1)
-        bit = bits = 0
-        samples = []
-        for d in delays:
-            while True:
-                if t + d < t1:
-                    if bit:
-                        bits += d
-                    t += d
-                    break
-                i = t1 - t
-                if bit:
-                    bits += i
-                d -= i
-                samples.append(bits / (t1 - s0))
-                s0 = t = t1
-                s1 += sample_delay
-                t1 = ceil(s1)
-                bits = 0
-            bit = 1 - bit
-        return samples
-
     def _delays_to_samples(self, delays, options):
         sample_delay = options[CLOCK_SPEED] / options[SAMPLE_RATE]
         samples = []

skoolkit/ay.py

@@ -16,7 +16,8 @@
 
 from math import ceil
 
-from skoolkit.audio import CLOCK_SPEED, FRAME_DURATION, SAMPLE_RATE, write_wav
+from skoolkit.audio import (CLOCK_SPEED, FRAME_DURATION, SAMPLE_RATE,
+                            moving_average_filter, write_wav)
 from skoolkit.simutils import CLOCK_SPEEDS, FRAME_DURATIONS
 
 AY_CLOCK_RATE = 1773400
@@ -224,7 +225,7 @@ def write_audio(self, audio_file, audio_log, options):
         ay_samples = ay.render(ay_log, frame_duration, ay_res, ay_volume)
         if options.beeper and beeper_log:
             delays = [t1 - t0 for t0, t1 in zip(beeper_log, beeper_log[1:])]
-            beeper_samples = self._render_beeper(delays, volume)
+            beeper_samples = moving_average_filter(delays, self.options, volume)
             samples = self._combine(ay_samples, beeper_samples)
         else:
             samples = ay_samples
@@ -240,32 +241,6 @@ def _parse_log(self, audio_log):
                 beeper_log.append(record[0])
         return ay_log, beeper_log
 
-    def _render_beeper(self, delays, volume):
-        # Apply a moving average filter
-        sample_delay = self.options[CLOCK_SPEED] / self.options[SAMPLE_RATE]
-        s0, s1 = 0, sample_delay
-        t, t1 = 0, ceil(s1)
-        bit = bits = 0
-        samples = []
-        for d in delays:
-            while True:
-                if t + d < t1:
-                    if bit:
-                        bits += d
-                    t += d
-                    break
-                i = t1 - t
-                if bit:
-                    bits += i
-                d -= i
-                samples.append(volume * bits / (t1 - s0))
-                s0 = t = t1
-                s1 += sample_delay
-                t1 = ceil(s1)
-                bits = 0
-            bit = 1 - bit
-        return samples
-
     def _combine(self, ay_samples, beeper_samples):
         samples = []
         ay_len = len(ay_samples) // 2