Return to Answer

Suggested

This comes with a lot of caveats. Since you don't have test usage, I haven't been able to test it, so I don't know whether it's valid. You're going to want to develop a numerical test suite to ensure that it's calculating the right thing. I also assumed that there's no need to hard-code for 3rd- or 6th-order filters, so just added an n. Finally: I don't have micropython, so this is written naively, assuming that standard Python usage is valid.

from array import array
from math import pi, acos
from typing import Sequence, Iterable


class EWMA:
    def __init__(self, coeff: Sequence[float], initial_value: float, n: int = None):
        nc = len(coeff)
        if n:
            self.n = n
            if nc > n:
                raise ValueError(f'len(coeff)={nc} > n={n}')
        else:
            self.n = nc  # default to the length of the coefficients

        # ewma states for coefficient optimization
        self.last_ewma = array('f', (0 for _ in range(self.n)))

        # default coefficients to 1.0 so the order can be from 0 - n
        # since cascade elements will pass input signal to output with a=1
        self.coeff = array('f', (1 for _ in range(self.n)))
        self.coeff[:nc] = coeff

        self.states = array('f', (0 for _ in range(1 + self.n)))

    def preload(self, value: float):
        self.states[:] = value

    @staticmethod
    def ewma(alpha: float, this: float, last: float) -> float:
        """
        calculate single EWMA element
        :param alpha: filter coefficient
        :param this: current input sample
        :param last: last output sample from this stage (feedback)
        :return: EWMA result
        """
        return alpha*this + (1 - alpha)*last

    def calculate(self, input_value: float) -> float:
        """
        calculate nth order cascade ewma
        :param input_value: Raw input sample
        :return: output of nth cascade element
        """
        self.states[0] = input_value
        for i, (c, s) in enumerate(zip(self.coeff, self.states[:-1])):
            self.states[i + 1] = self.ewma(c, s, self.states[i + 1])

        return self.get_last_output()

    def get_last_output(self) -> float:
        return self.states[-1]

    def model_ewma_preload(self, v: float):
        self.last_ewma[:] = v

    def model_ewma(self, y0: float, coeffs: Sequence[float]) -> float:
        """
        ewma nth order for IIR Model Fitting via SciPy Optimize
        :param y0: The input value
        :param coeffs: Sequence of coefficients
        :return: IIR output
        """
        prev = y0
        for i, (c, e) in enumerate(zip(coeffs, self.last_ewma)):
            new = self.ewma(c, prev, e)
            self.last_ewma[i] = new
            prev = new
        return prev

    def get_cutoff(self, fs: float = 1) -> array:
        return array(
            'f',
            (
                fs*pi/2 * acos(1 - c**2/2/(1 - c))
                for c in self.coeff
            )
        )
    
    def apply_to_data(self, data: Iterable[float]) -> array:
        return array('f', (self.calculate(d) for d in data))

Copy a list into a slice

This code:

for c in range(0, len(coeff)):
    if(c >= len(self.coeff)):
        print(f'EWMA Coefficients Length Mismatch! len(coeff) = {len(coeff)}, max is 6')
        break
    self.coeff[c] = coeff[c]

has a couple of problems. It's fine to validate the length of coeff, but it shouldn't be done like this. Also, don't do an element-by-element copy. Instead:

N = 6
self.coeff = [1]*N

if len(coeff) > N:
   raise ValueError(f'EWMA Coefficients Length Mismatch! {len(coeff)} > {N}')
self.coeff[:len(coeff)] = coeff

This:

    self.states = [0, 0, 0, 0, 0, 0, 0]
    self.states[0] = initialValue
    self.states[1] = initialValue
    self.states[2] = initialValue
    self.states[3] = initialValue
    self.states[4] = initialValue
    self.states[5] = initialValue
    self.states[6] = initialValue

should be

self.states = [initialValue] * (N+1)

And so on.

Docstrings

Move your function documentation into """triple quotes""" at the first line inside of your function.

Never try/except

This breaks Ctrl+C quit, and is too broad to be useful. Narrow your caught exception type.

Don't repeat yourself

get_cutoff needs to be rewritten as a loop over N values.

Don't cast unnecessarily

This:

def ewma(self, alpha: float, this: float, last: float) -> float:
    return (float(alpha)*float(this)) + ((1.0-float(alpha))*float(last))

already assumes that the inputs are floats - so don't call float again. Drop all of your casts.

General

Once you've cleaned up your usage of lists, you should really consider moving to numpy the array module. It'll execute more quickly.

Copy a list into a slice

This code:

for c in range(0, len(coeff)):
    if(c >= len(self.coeff)):
        print(f'EWMA Coefficients Length Mismatch! len(coeff) = {len(coeff)}, max is 6')
        break
    self.coeff[c] = coeff[c]

has a couple of problems. It's fine to validate the length of coeff, but it shouldn't be done like this. Also, don't do an element-by-element copy. Instead:

N = 6
self.coeff = [1]*N

if len(coeff) > N:
   raise ValueError(f'EWMA Coefficients Length Mismatch! {len(coeff)} > {N}')
self.coeff[:len(coeff)] = coeff

This:

    self.states = [0, 0, 0, 0, 0, 0, 0]
    self.states[0] = initialValue
    self.states[1] = initialValue
    self.states[2] = initialValue
    self.states[3] = initialValue
    self.states[4] = initialValue
    self.states[5] = initialValue
    self.states[6] = initialValue

should be

self.states = [initialValue] * (N+1)

And so on.

Docstrings

Move your function documentation into """triple quotes""" at the first line inside of your function.

Never try/except

This breaks Ctrl+C quit, and is too broad to be useful. Narrow your caught exception type.

Don't repeat yourself

get_cutoff needs to be rewritten as a loop over N values.

Don't cast unnecessarily

This:

def ewma(self, alpha: float, this: float, last: float) -> float:
    return (float(alpha)*float(this)) + ((1.0-float(alpha))*float(last))

already assumes that the inputs are floats - so don't call float again. Drop all of your casts.

Probable bug

def model_ewma6(self, y0, a, b, c, d, e, f):
    y1 = self.ewma(a, y0, self.last_ewma3[0])

Seems that you're using the wrong array here. Also - why are you hard-coding for 3rd- or 6th-order filters? Can you not just accept N as a parameter?

General

Once you've cleaned up your usage of lists, you should really consider moving to numpy the array module. It'll execute more quickly.

Copy a list into a slice

This code:

for c in range(0, len(coeff)):
    if(c >= len(self.coeff)):
        print(f'EWMA Coefficients Length Mismatch! len(coeff) = {len(coeff)}, max is 6')
        break
    self.coeff[c] = coeff[c]

has a couple of problems. It's fine to validate the length of coeff, but it shouldn't be done like this. Also, don't do an element-by-element copy. Instead:

N = 6
self.coeff = [1]*N

if len(coeff) > N:
   raise ValueError(f'EWMA Coefficients Length Mismatch! {len(coeff)} > {N}')
self.coeff[:len(coeff)] = coeff

This:

    self.states = [0, 0, 0, 0, 0, 0, 0]
    self.states[0] = initialValue
    self.states[1] = initialValue
    self.states[2] = initialValue
    self.states[3] = initialValue
    self.states[4] = initialValue
    self.states[5] = initialValue
    self.states[6] = initialValue

should be

self.states = [initialValue] * (N+1)

And so on.

Docstrings

Move your function documentation into """triple quotes""" at the first line inside of your function.

Never try/except

This breaks Ctrl+C quit, and is too broad to be useful. Narrow your caught exception type.

Don't repeat yourself

get_cutoff needs to be rewritten as a loop over N values.

General

Once you've cleaned up your usage of lists, you should really consider moving to numpy. It'll execute more quickly.

Copy a list into a slice

This code:

for c in range(0, len(coeff)):
    if(c >= len(self.coeff)):
        print(f'EWMA Coefficients Length Mismatch! len(coeff) = {len(coeff)}, max is 6')
        break
    self.coeff[c] = coeff[c]

has a couple of problems. It's fine to validate the length of coeff, but it shouldn't be done like this. Also, don't do an element-by-element copy. Instead:

N = 6
self.coeff = [1]*N

if len(coeff) > N:
   raise ValueError(f'EWMA Coefficients Length Mismatch! {len(coeff)} > {N}')
self.coeff[:len(coeff)] = coeff

This:

    self.states = [0, 0, 0, 0, 0, 0, 0]
    self.states[0] = initialValue
    self.states[1] = initialValue
    self.states[2] = initialValue
    self.states[3] = initialValue
    self.states[4] = initialValue
    self.states[5] = initialValue
    self.states[6] = initialValue

should be

self.states = [initialValue] * (N+1)

And so on.

Docstrings

Move your function documentation into """triple quotes""" at the first line inside of your function.

Never try/except

This breaks Ctrl+C quit, and is too broad to be useful. Narrow your caught exception type.

Don't repeat yourself

get_cutoff needs to be rewritten as a loop over N values.

Don't cast unnecessarily

This:

def ewma(self, alpha: float, this: float, last: float) -> float:
    return (float(alpha)*float(this)) + ((1.0-float(alpha))*float(last))

already assumes that the inputs are floats - so don't call float again. Drop all of your casts.

General

Once you've cleaned up your usage of lists, you should really consider moving to numpy the array module. It'll execute more quickly.