dmatest: vary sampling frequency based on signal processing

programs/dmatest_v2.py

@@ -3,14 +3,12 @@
 # which is  Copyright (c) 2021 Jeremy P Bentham and licensed under the Apache License, Version 2.0
 
 import array
-import math
+import asyncio
+import gc
 
-import machine
 import rp2
-import asyncio
 import uctypes
 from uctypes import BF_POS, BF_LEN, UINT32, BFUINT32, struct
-import gc
 
 ADC_BASE = 0x4004C000
 
@@ -61,7 +59,8 @@ def t(x):
 
 SAMPLES = 1000
 SAMPLING_RATE_HZ = 22000
-
+SAMPLING_FREQ_TO_HIGH = -2
+SAMPLING_FREQ_TO_LOW = -1
 
 class AdcDevice:
     def __init__(self, device_nr, rate=SAMPLING_RATE_HZ, samples=SAMPLES):
@@ -92,6 +91,9 @@ class AdcDevice:
             self._rate = rate
             self.__adc_device.DIV_REG = (48000000 // self._rate - 1) << 8
 
+    def get_sampling_rate(self):
+        return self._rate
+
     def set_sample_count(self, samples):
         async with self.__lock:
             self._sample_count = samples
@@ -191,10 +193,13 @@ def calculate_frequency(voltages_centered, sampling_rate):
         consecutive_sign += 1
 
     if len(counts) < 3:
-        return -1
+        return SAMPLING_FREQ_TO_HIGH
 
     counts = counts[1:-1]
-    f = sampling_rate / (2 * sum(counts) / len(counts))
+    count_mean = sum(counts) / len(counts)
+    if count_mean < 5:
+        return SAMPLING_FREQ_TO_LOW
+    f = sampling_rate / (2 * count_meant)
     count = None
     return f
 
@@ -239,6 +244,23 @@ async def main():
         voltage_mean = sum(voltages) / SAMPLES
         voltages_centered = [v - voltage_mean for v in voltages]
         freq = calculate_frequency(voltages_centered, sampling_rate=SAMPLING_RATE_HZ)
+        if freq == SAMPLING_FREQ_TO_HIGH:
+            rate = device.get_sampling_rate()
+            if rate < 3:
+                print(f"Cannot reduce sampling frequency rate further than {rate}")
+                continue
+            device.set_sampling_rate(rate / 2.0)
+            print(f"Reducing sampling rate to {device.get_sampling_rate()}")
+            continue
+        elif freq == SAMPLING_FREQ_TO_LOW:
+            rate = device.get_sampling_rate()
+            if rate > 100000:
+                print(f"Cannot reduce increase frequency rate further than {rate}")
+                continue
+            device.set_sampling_rate(rate * 2)
+            print(f"Increasing sampling rate to {device.get_sampling_rate()}")
+            continue
+
         v_to_v = calculate_peak_to_peak(voltages_centered)
         print(f"{freq:0.2f} Hz, {v_to_v:0.2f} V peak to peak")
         gc.collect()