Espressif analogbufio implementation

[milindmovasha]

Hello team, this is a PR for implementation of analogbuio module for espressif port. I have tested the code on ESP32S2 board adafruit_feather_esp32s2. The code supports other version of ESP32 chip however it is not tested. Please note this code is compatible with ESP IDF version 4.4.x. There have been changes in ESP IDF 5.0 implementation of ADC DMA and the module will have to be updated when CircuitPython moves to ESP IDF 5.0

Test code:

import time
import board
import analogbufio
import array
import supervisor
import gc

SAMPLE_RATE = const(40000)
CAPTURE_TIME_SECONDS = const(20)
ADC_PIN = board.A5
#For ESP32 and ESP32S2 set SAMPLE_SIZE_IN_BYTES to 2
#For ESP32C3, ESP32H2 and ESP32S3 set SAMPLE_SIZE_IN_BYTES to 4
SAMPLE_SIZE_IN_BYTES = 2

_TICKS_PERIOD = const(1<<29)
_TICKS_MAX = const(_TICKS_PERIOD-1)
_TICKS_HALFPERIOD = const(_TICKS_PERIOD//2)

def ticks_diff(ticks1, ticks2):
    "Compute the signed difference between two ticks values, assuming that they are within 2**28 ticks"
    diff = (ticks1 - ticks2) & _TICKS_MAX
    diff = ((diff + _TICKS_HALFPERIOD) & _TICKS_MAX) - _TICKS_HALFPERIOD
    return diff

def ram_available(sample_rate, capture_time_seconds, sample_size_in_bytes):
    ram_required = sample_rate*capture_time_seconds*sample_size_in_bytes
    gc.collect()
    ram_available = gc.mem_free()
    return (ram_required < ram_available)

def collect_samples(adc_pin, sample_rate, capture_time_seconds, sample_size_in_bytes):
    try:
        numSamples = sample_rate*capture_time_seconds
        print("Reading",numSamples,"samples from",adc_pin,"pin")
        if (sample_size_in_bytes == 2):
            mybuffer = array.array('H', [0]) * numSamples
        else:
            mybuffer = array.array('L', [0]) * numSamples
        adcbuf = analogbufio.BufferedIn(adc_pin, sample_rate=sample_rate)
        start_time = supervisor.ticks_ms()
        numCapturedSamples = adcbuf.readinto(mybuffer)
        end_time = supervisor.ticks_ms()
        adcbuf.deinit()
        if(numCapturedSamples == numSamples):
            timetaken = ticks_diff(end_time,start_time)
            print("Number of samples read:",numCapturedSamples)
            print("Time required",timetaken,"ms")
            print("Desired sampling rate:",sample_rate,"/s")
            print("Computed sampling rate:",(numCapturedSamples*1000)/timetaken,"/s")
        else:
            print("Unable to capture all samples")
    except (ValueError, RuntimeError) as e:
        print("Exception", e)

def main():
    print("\n=================")
    if ram_available(SAMPLE_RATE,CAPTURE_TIME_SECONDS,SAMPLE_SIZE_IN_BYTES):
        collect_samples(ADC_PIN,SAMPLE_RATE,CAPTURE_TIME_SECONDS,SAMPLE_SIZE_IN_BYTES)
    else:
        print("Not enough RAM available to collect required samples")
    print("=================\n")

main()

Sample output:

=================
Reading 800000 samples from board.A5 pin
Number of samples read: 800000
Time required 19869 ms
Desired sampling rate: 40000 /s
Computed sampling rate: 40263.7 /s
=================

[tannewt]

Thank you for this! One structural comment.

[dhalbert]

Thanks for working on this! Additional comments to @tannewt's.

[dhalbert]

BTW, see https://github.com/adafruit/circuitpython/actions/runs/4205312574/jobs/7297211968 pre-commit section for formatting changes. Also, we use spaces after commas in arg lists.

[milindmovasha]

I have committed the changes as per all the review comments received so far. Except for the translation related pre-commit failure everything seems to be fine. Please let me know how to resolve translation related error and also additional review comments if any.

Following is the updated test code and output:

Test code:

import time
import board
import analogbufio
import array
import supervisor
import gc

SAMPLE_RATE = const(40000)
CAPTURE_TIME_SECONDS = const(2)
ADC_PIN = board.A5
SAMPLE_SIZE_IN_BYTES = 2

_TICKS_PERIOD = const(1<<29)
_TICKS_MAX = const(_TICKS_PERIOD-1)
_TICKS_HALFPERIOD = const(_TICKS_PERIOD//2)

def ticks_diff(ticks1, ticks2):
    "Compute the signed difference between two ticks values, assuming that they are within 2**28 ticks"
    diff = (ticks1 - ticks2) & _TICKS_MAX
    diff = ((diff + _TICKS_HALFPERIOD) & _TICKS_MAX) - _TICKS_HALFPERIOD
    return diff

def ram_available(sample_rate, capture_time_seconds, sample_size_in_bytes):
    ram_required = sample_rate*capture_time_seconds*sample_size_in_bytes
    gc.collect()
    ram_available = gc.mem_free()
    return (ram_required < ram_available)

def collect_samples(adc_pin, sample_rate, capture_time_seconds, sample_size_in_bytes):
    try:
        numSamples = sample_rate*capture_time_seconds
        print("Reading",numSamples,"samples from",adc_pin,"pin")
        mybuffer = array.array('H', [0]) * numSamples
        adcbuf = analogbufio.BufferedIn(adc_pin, sample_rate=sample_rate)
        start_time = supervisor.ticks_ms()
        numCapturedSamples = adcbuf.readinto(mybuffer)
        end_time = supervisor.ticks_ms()
        adcbuf.deinit()
        if(numCapturedSamples == numSamples):
            timetaken = ticks_diff(end_time,start_time)
            print("Number of samples read:",numCapturedSamples)
            print("Time required",timetaken,"ms")
            print("Desired sampling rate:",sample_rate,"/s")
            print("Computed sampling rate:",(numCapturedSamples*1000)/timetaken,"/s")
        else:
            print("Unable to capture all samples")
    except (ValueError, RuntimeError) as e:
        print("Exception", e)

def main():
    print("\n=================")
    if ram_available(SAMPLE_RATE,CAPTURE_TIME_SECONDS,SAMPLE_SIZE_IN_BYTES):
        collect_samples(ADC_PIN,SAMPLE_RATE,CAPTURE_TIME_SECONDS,SAMPLE_SIZE_IN_BYTES)
    else:
        print("Not enough RAM available to collect required samples")
    print("=================\n")

main()

Output:

=================
Reading 80000 samples from board.A5 pin
Number of samples read: 80000
Time required 1987 ms
Desired sampling rate: 40000 /s
Computed sampling rate: 40261.7 /s
=================

[dhalbert]

For the translate issue, at the top level of the repo, do make translate. That will change locals/circuitpython.pot. Commit and push that.

You can run pre-commit locally to save the round-trips to GitHub. See https://learn.adafruit.com/building-circuitpython/build-circuitpython#install-pre-commit-3096511 and the links it references.

[dhalbert]

Thanks for persevering with all the formatting changes!

[dhalbert]

@milindmovasha If I understand correctly, you are allowing one-byte arrays or two-byte arrays to be passed in, but you always write two bytes. Is that correct?

Normally if a result is 16-bit words, we would design an API to require an array.array('H', ...) (or maybe larger), instead of spreading the result over two bytes. That is why I was asking for further checking of the allowed type of the passed-buffer.

If the previous analogbufio implementation is doing something similar, we should fix it as well.

[milindmovasha]

@milindmovasha If I understand correctly, you are allowing one-byte arrays or two-byte arrays to be passed in, but you always write two bytes. Is that correct?

Normally if a result is 16-bit words, we would design an API to require an array.array('H', ...) (or maybe larger), instead of spreading the result over two bytes. That is why I was asking for further checking of the allowed type of the passed-buffer.

If the previous analogbufio implementation is doing something similar, we should fix it as well.

@dhalbert yes, your understanding is correct. I will restrict the buffer type to array.array('H', ...)

[milindmovasha]

I have restricted the buffer type to array.array('H', ...) as part of commit 09f84e3

[dhalbert]

Looks good re my comments! I'm making a simple grammar change myself.

addressed

[dhalbert]

Changed my mind; "an array" to just "array", for space reasons, and to make translation easier.