<commit_before><commit_msg>First testable ZI kernels<commit_after>"""
Kernel functions designed to be compatible with real-time digital filters
(IIR and FIR)

This implements three basic filters and a rounding function
- multipath_bias_tee
- multipath_filter
- multipath_filter2

"""
import numpy as np
from scipy import signal


def bias_tee_correction(ysig, tau: float, sampling_rate: float=1):
    """
    Corrects for a bias tee correction using a linear IIR filter with time
    constant tau.
    """
    k = tau*sampling_rate
    b = [1, -1]
    a = [(k+1)/k, -(k-1)/k]

    # hint: to get the inverse just use the filter with (b, a, ysig)
    filtered_signal = signal.lfilter(a, b, ysig)
    return filtered_signal


def exponential_decay_correction(ysig, tau: float,
                                 amp: float, offset: float,
                                 sampling_rate: float=1):
    """
    Corrects for an exponential decay using a linear IIR filter.
    """
    alpha = offset + amp
    k = tau*sampling_rate

    beta = 1-alpha
    b = [(1-k*alpha)/(1-k), -beta/(1-k)]
    a = [1, -beta]

    # hint: to get the inverse just use the filter with (b, a, ysig)
    filtered_signal = signal.lfilter(a, b, ysig)
    return filtered_signal


#################################################################
#              hardware friendly functions                      #
#################################################################


def coef_round(value):
    """
    hardware friendly
    rounds coefficients to hardware-compatible pseudo-float
    """
    if type(value) == dict:
        new_value = {}
        for key in value:
            new_value[key] = coef_round(value[key])
    elif type(value) == list:
        new_value = []
        for val in value:
            new_value.append(coef_round(val))
    else:
        # scalar value
        new_value = 0.
        # to simplify the cores we allow 4 discrete shifts by 4 bits: 0, -4, -8
        # and -12
        bshift = np.max(
            [np.min([np.floor(-np.log2(np.abs(value))/4.)*4., 12.]), 0.])
        # 18 bits within the multiplication itself
        scaling = np.power(2, 18+bshift)
        new_value = np.round(value*scaling) / scaling

    return new_value


def multipath_bias_tee(sig, k, paths):
    """
    hardware friendly
    hardware friendly bias-tee (or any other AC coupling) compensation filter
    """
    tpl = np.ones((paths, ))
    cs = 0
    acc = []
    for i in np.arange(0, sig.size, paths):
        cs = cs + np.sum(sig[i:(i+paths)])
        acc = np.append(acc, tpl*cs)
    return sig + 1./k * (2*acc - sig)


def multipath_filter(sig, alpha, k, paths):
    """
    hardware friendly
    exponential moving average correction filter
    """
    tpl = np.ones((paths, ))
    hw_alpha = alpha*float(paths)

    duf = tpl * 0.
    acc = 0

    for i in np.arange(0, sig.size, paths):
        acc = acc + hw_alpha*(np.mean(sig[i:(i+paths)]) - acc)
        duf = np.append(duf, tpl * acc)
    duf = duf[0:sig.size]
    return sig + k * (duf - sig)


def multipath_filter2(sig, alpha, k, paths):
    """
    hardware friendly
    exponential moving average correction filter with pipeline simulation
    """
    ppl = 4
    tpl = np.ones((paths, ))
    hw_alpha = alpha*float(paths*ppl)

    duf = tpl * 0.
    # due to the pipelining, there are actually ppl interleaved filters
    acc = np.zeros((ppl, ))

    # first create an array of averaged path values
    du = []
    for i in np.arange(0, sig.size, paths):
        du = np.append(du, np.mean(sig[i:(i+paths)]))

    # loop through the average bases
    for i in np.arange(0, du.size, ppl):
        # loop through the individual sub-filters
        for j in np.arange(0, ppl):
            # first calculate the filter input as an average of the previous
            # path averages
            ss = 0
            for l in np.arange(0, ppl):
                if i+j-l >= 0:
                    ss = ss + du[i+j-l]
            ss = ss / float(ppl)
            # then process it by the filter
            acc[j] = acc[j] + hw_alpha*(ss - acc[j])
            # and add the filter output to the signal correction vector
            duf = np.append(duf, tpl * acc[j])
    duf = duf[0:sig.size]
    return sig + k * (duf - sig)

{'fraction_non_alphanumeric': 0.1411901983663944, 'fraction_numerical': 0.011435239206534423, 'mean_word_length': 2.212893553223388, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4320441', 'n_tokens_mistral': 1407, 'n_tokens_neox': 1335, 'n_words': 424}