Measuring current with the ADC

[lbergman]

I have scoured the forums to no avail. In short, I have 4 AC current sources I was hoping to use something like the split core current transformer (SCT-013-020) from YHDC. SCT Spec sheet with the GPy. I am being told by an Electrical Engineer that both MicroPython and the Pycom boards have limitations with accuracy that cannot be overcome making this type of measurement impossible. It seems to me, while I am no electrical engineer, that there are projects in existence that use the same or similar ADC to do the exact same thing. Specifically, the openenergymonitor.org project. The only info on the ESP32 I could find seem to suggest a max sample rate of 6KHz for the entire ADC or in my case, 1500 samples a second. I believe that number to be way more accurate than I require.

I only need to be accurate down to about .25 amps with about 1/10 of a second. At 50Hz and 60Hz that means at minimum I would need at least 60Hz x 10 * 4

The question is: Can this be done?

I just received my first GPy this morning and have yet to receive my CT's. I am hoping someone has done this and can tell me it will work.

[robert-hh]

@lbergman Ok, looking at the current transformer, it is specified with at 20A range. A resolution of .25 A means abiout 7 bit resolution. Even if the ESP32 performs below the expectations, it is good for 8 bit. So you can simply get a 12 bit value and ignore the lower four bits, or run it in 9 bit mode.

The current transformer delivers an AC output voltage, the ADC expects DC. You need some circuitry to cope with that. The simplest solution would be to put one side of the current transformer at Vcc/2, and the other side into the ADC input. Set the ADC to 11db attenuation and add a protection resistor at the input. What you then get from the ADC is the sine wave of the current with on offset of about range/2. The difference between max and min should tell you the actual current. 400 samples/sec should be possible. The call to the ADC takes about 100µs.
Edit:
I found a sample connection. It says Arduino, but read LoPy instead. Use 3.3V instead of the 5V shown here. C1 should be something like 1000µF. The burden Resistor is already built into the unit your ordered.

[lbergman]

@robert-hh Thanks for your replay. I saw that example on the site I mentioned and thought it might work. I really appreciate you clarifying what is needed and the 3.3 vs 5 volts.

[robert-hh]

@lbergman Since I found the topic interesting, I did some testing myself, using a function generator as input source. Finding: the ADC is not good, but about usable for your purpose. It is noisy and nonlinear, but some of the nois seen also results from the varying IRQ response latency of the firmware. Some hints.

• Devices with a rev1 chip perform better, so use this one. But there is a big variation between individual devices. But with respect to noise the Pycom boards are better than, let's say, the Wemos boards.
• Add a low impedance capacitor (ceramic) of like 10nF as close as possible to the ADC input between input and GND, with short leads. That reduces the noise.
• If possible, use the 6dB attenuation mode. That requires an additional 2:1 voltage divider, but gives better linearity. You will need some analog filtering anyhow to protect the inputs of the ESP32 from spikes. This is not shown in the schematics below. The sample code below uses 11db attenuation.
• My test code is below. It uses a timer for acquisition to get regular sampling, and uses the sin2 + cos2 approach to determine the amplitude. The value for noise is what you get with no signal.

#
import math
import array
from machine import ADC, Timer, idle
import utime
samples = 600
_BUFFERSIZE = const(64)

#
# acquire ADC values at a fixed rate/s, which is given in
# the constructor. The second argument is the size
# of a ringbuffer. 
#
class Acquire:
    def __init__(self, samples, buffersize, pin):
        self.put = 0
        self.get = 0
        self.buffersize = buffersize
        self.buffer = array.array("h", 0 for _ in range(self.buffersize))
        self.adc = ADC(bits=9)
        self.apin = self.adc.channel(pin=pin, attn=ADC.ATTN_11DB)
        self._alarm = Timer.Alarm(self.read_adc, 1/samples, periodic=True)

    def stop(self):
        self._alarm.cancel()

    def read_adc(self, alarm):
        self.buffer[self.put] = self.apin()
        self.put = (self.put + 1) % self.buffersize

    def next(self):
        while self.get == self.put:
            idle()
        # calculate the moving average over the last three values:
        self.value = (self.buffer[self.get] +
                      self.buffer[(self.get - 1) % self.buffersize] +
                      self.buffer[(self.get - 2) % self.buffersize]) / 3
        self.get = (self.get + 1) % self.buffersize
        return self.value
#
# calculate a moving average
# The constructors argument is the window size
#
class Average:
    def __init__(self, size):
        self.size = size
        self.buffer = array.array("f", 0.0 for _ in range (size))
        self.put = 0
        self.sum = 0.0

    def avg(self, value):
        self.sum -= self.buffer[self.put]
        self.buffer[self.put] = value
        self.put = (self.put + 1) % self.size
        self.sum += value
        return self.sum / self.size


def run(noise=0.6):
    acq = Acquire(samples, _BUFFERSIZE, 'P20')
    avg_offset = Average(samples)
    avg_res = Average(128)

    for _ in range(samples):
        offset = avg_offset.avg(acq.next())
    print("Offset = ",offset)

    noise_sq = noise * noise

    try:
        prev = acq.next() - offset
        i = 0
        while True:
            act = acq.next()
            offset = avg_offset.avg(act)
            act -= offset
            # estimate 1st order derivate as dy/dt, dt is implicitely set to 1
            deriv = (act - prev)
            prev = act
            # get sin**2 + cos**2 = actual value
            # and feed it through the average machine
            res = avg_res.avg(math.sqrt(act * act + deriv * deriv))
            i += 1
            if i == samples:
                # remove the noise floor
                res = math.sqrt(abs(res*res - noise_sq))
                print("{:6.2f} ".format(res))
                i = 0
    except KeyboardInterrupt:
        acq.stop()
        print("Handler stopped")

run()

[robert-hh]

@robert-hh One additional not: by processing the output of the adc.value() call with a simple (3 tap) digital low-pass the noise could be reduces further. One has to find the balance between filtering characteristics and time it needs to be processed.
A good web site for calculating the filter parameters is this one: http://t-filter.engineerjs.com/

[robert-hh]

@lbergman I have updated the code below. It copes now for an offset drift and uses a simple low pass filter for the output of the ADC. Assuming a 20A range, the resolution is about 100 mA, depending on the applied smoothing (the length of avg_res).

[robert-hh]

@robert-hh Another variant for measuring, which determines the difference between a minimal and maximal value over a window of samples, which should be at least a period of the signal to be tested, e.g. 20 ms for a 50Hz net. More simple than the last example, and immune to offset drifts. Still, the ADC is bad. The noise is a little bit lower if WiFi is switched off, but that only matters at small signals.

#
import math
import array
from machine import ADC, Timer, idle
samples = 600
freq = 50
_BUFFERSIZE = const(64)

#
# acquire ADC values at a fixed rate/s, which is given in
# the constructor. The second argument is the size
# of a ringbuffer, the third a string with the Pin name
#
class Acquire:
    def __init__(self, samples, buffersize, pin):
        self.put = 0
        self.get = 0
        self.buffersize = buffersize
        self.buffer = array.array("h", 0 for _ in range(self.buffersize))
        self.adc = ADC(bits=9)
        self.apin = self.adc.channel(pin=pin, attn=ADC.ATTN_11DB)
        self._alarm = Timer.Alarm(self.read_adc, 1/samples, periodic=True)

    def stop(self):
        self._alarm.cancel()

    def read_adc(self, alarm):
        self.buffer[self.put] = self.apin.value()
        self.put = (self.put + 1) % self.buffersize

    def next(self, period):
        vmin = 4096
        vmax = 0
        for _ in range(period):
            while self.get == self.put:
                idle()
            # calculate the moving average over the last three values:
            value = (self.buffer[self.get] +
                     self.buffer[(self.get - 1) % self.buffersize] +
                     self.buffer[(self.get - 2) % self.buffersize]) / 3
            vmin = min(value, vmin)
            vmax = max(value, vmax)
            self.get = (self.get + 1) % self.buffersize
        return vmax - vmin
#
# calculate a moving average
# The constructors argument is the window size
#
class Average:
    def __init__(self, size):
        self.size = size
        self.buffer = array.array("f", 0.0 for _ in range (size))
        self.put = 0
        self.sum = 0.0

    def avg(self, value):
        self.sum -= self.buffer[self.put]
        self.buffer[self.put] = value
        self.put = (self.put + 1) % self.size
        self.sum += value
        return self.sum / self.size


def run(noise=1.2):
    acq = Acquire(samples, _BUFFERSIZE, 'P20')
    avg_res = Average(freq)

    noise_sq = noise * noise

    try:
        i = 0
        while True:
            res = avg_res.avg(acq.next(samples/freq))
            i += 1
            if i == freq:
                # remove the noise floor
                res = math.sqrt(abs(res*res - noise_sq))
                print("{:6.2f} ".format(res/20.2))
                i = 0
    except KeyboardInterrupt:
        acq.stop()
        print("Handler stopped")

run()