text:
import numpy as np
from gp_kernel import kernel
from fast_gp import sparse_w, post_cov_ys, post_mean
from scipy.linalg import sqrtm
import time
from pickle_io import pickle_load, pickle_save
from np_posterior_gp import PosteriorGP

def cov_mat(x1, x2, gp_params):
    a, b = gp_params
    return a * np.exp(-b * (x1[:, np.newaxis] - x2)**2)

def exact_proj_cholesky(x, y, x_test, gp_params, indep_noise, batch_size,
                        eps=None):
    Ktt = cov_mat(x_test, x_test, gp_params)
    Kxt = cov_mat(x, x_test, gp_params)
    Kxx = cov_mat(x, x, gp_params) + indep_noise * np.eye(len(x))
    KxtT_Kxxinv = np.linalg.solve(Kxx, Kxt).T
    K = Ktt - KxtT_Kxxinv.dot(Kxt) + 1e-12 * np.eye(len(x_test))
    R = np.linalg.cholesky(K)
    if eps is None:
        eps = np.random.normal(size=(batch_size, len(x_test)))
    return R.dot(eps.T).T

def exact_proj_sqrtm(x, y, x_test, gp_params, indep_noise, batch_size,
                        eps=None):
    Ktt = cov_mat(x_test, x_test, gp_params)
    Kxt = cov_mat(x, x_test, gp_params)
    Kxx = cov_mat(x, x, gp_params) + indep_noise * np.eye(len(x))
    KxtT_Kxxinv = np.linalg.solve(Kxx, Kxt).T
    K = Ktt - KxtT_Kxxinv.dot(Kxt) + 1e-12 * np.eye(len(x_test))
    R = sqrtm(K)
    if eps is None:
        eps = np.random.normal(size=(batch_size, len(x_test)))
    return R.dot(eps.T).T

def cov_sor(x, z, u, gp_params):
    Kxu = cov_mat(x, u, gp_params)
    Kuu = cov_mat(u, u, gp_params) #+ np.eye(len(u)) * 1e-12
    Kuz = cov_mat(u, z, gp_params)
    return Kxu.dot(np.linalg.solve(Kuu, Kuz))

def exact_gp_sample(x, y, x_test, gp_params, indep_noise, eps,
                    decomp='sqrtm', return_mean=False):
    Ktt = cov_mat(x_test, x_test, gp_params)
    Kxt = cov_mat(x, x_test, gp_params)
    Kxx = cov_mat(x, x, gp_params) + indep_noise * np.eye(len(x))
    #print 'condition number', np.linalg.cond(Kxx)
    KxtT_Kxxinv = np.linalg.solve(Kxx, Kxt).T
    K = Ktt - KxtT_Kxxinv.dot(Kxt) + 1e-12 * np.eye(len(x_test))

    if decomp == 'sqrtm':
        R = sqrtm(K).real
    elif decomp == 'cholesky':
        R = np.linalg.cholesky(K)
    else:
        raise NotImplementedError('Undefined decomposition method.')

    mu = KxtT_Kxxinv.dot(y)
    sample = mu + R.dot(eps.T).T
    if return_mean:
        return mu, sample
    return sample

def sor_gp_sample(x, y, x_test, u, gp_params, indep_noise, eps):
    Ktt = cov_sor(x_test, x_test, u, gp_params)
    Kxt = cov_sor(x, x_test, u, gp_params)
    Kxx = cov_sor(x, x, u, gp_params) + indep_noise * np.eye(len(x))
    KxtT_Kxxinv = np.linalg.solve(Kxx, Kxt).T
    K = Ktt - KxtT_Kxxinv.dot(Kxt) + 1e-12 * np.eye(len(x_test))
    R = sqrtm(K).real
    mu = KxtT_Kxxinv.dot(y)
    #print mu
    return mu + R.dot(eps.T).T

def approx_gp_sample(x, y, x_test, gp_params, indep_noise, eps,
                     len_u, lanczos_basis=10, return_mean=False):
    x_min, x_max = x.min(), x.max()
    extra_u = 2   # for better interpolation
    margin = (x_max - x_min) / (len_u - extra_u * 2) * 2
    u = np.linspace(x_min - margin, x_max + margin, len_u)

    idx_train, w_train = sparse_w(u, x)
    idx_test, w_test = sparse_w(u, x_test)
    post_gp = PosteriorGP(u, x_test, x, y, kernel, gp_params, indep_noise)
    cov_zs = post_gp.cov_rand_proj(n_lanczos_basis=lanczos_basis, eps=eps)
    mu = post_gp.mean()
    sample = mu + cov_zs
    if return_mean:
        return mu, sample
    return sample

def norm_diff():
    n_data = 1000
    x = np.random.uniform(size=n_data)
    #x = np.float32(x)
    x = np.sort(x)
    a = 1
    b = 1000.
    c = .01
    mu = np.zeros(n_data)
    cov = a * np.exp(-b * (x[:, np.newaxis] - x)**2) + c * np.eye(n_data)
    y = np.random.multivariate_normal(mu, cov)

    gp_params = np.array([a, b])
    indep_noise = c

    #data = 'ECG200/11_1000_60_dat.pkl'
    #gp_parms, ts_train, ts_test, l_train, l_test = pickle_load(data)[:5]
    #x_train = np.array([each_ts[0] for each_ts in ts_train])
    #y_train = np.array([each_ts[1] for each_ts in ts_train])
    #eg_id = 0
    #x = x_train[eg_id]
    #y = y_train[eg_id]

    #print x
    #print y
    x_min, x_max = x.min(), x.max()
    len_u = 4096
    #for t in xrange(3):

    ####################
    extra_u = 2
    margin = (x_max - x_min) / (len_u - extra_u * 2) * 2
    u = np.linspace(x_min - margin, x_max + margin, len_u)
    x_test = u[1:]

    idx_train, w_train = sparse_w(u, x)
    idx_test, w_test = sparse_w(u, x_test)
    post_gp = PosteriorGP(u, x_test, x, y, kernel, gp_params, indep_noise)

    batch_size = 10
    eps = np.random.normal(size=(batch_size, len(x_test)))

    '''
    exact_samples = exact_gp_sample(x, y, x_test,
                                    gp_params, indep_noise, eps)

    for n_basis in xrange(2, 20 + 1):

        print n_basis
        cov_zs = post_gp.cov_rand_proj(n_sample=batch_size,
                                       n_lanczos_basis=n_basis, eps=eps)
        mu = post_gp.mean()
        gp_samples = mu + cov_zs
        print mu

        #sor_samples = sor_gp_sample(x, y, x_test, u,
        #                            gp_params, indep_noise, eps)
        #print np.linalg.norm(gp_samples - sor_samples)
        #print np.linalg.norm(sor_samples - exact_samples)

        print np.linalg.norm(gp_samples - exact_samples)
        print

    print '-' * 30
    '''

    x_test = np.linspace(x_min, x_max, 512)
    #x_test = np.linspace(x_min, x_max, 1024)
    eps = np.random.normal(size=(batch_size, len(x_test)))
    exact_mu, exact_samples = exact_gp_sample(x, y, x_test,
                                    gp_params, indep_noise, eps,
                                    return_mean=True)

    import pylab as pl
    pl.figure()
    pl.ion()
    pl.plot(x, y, '.-')
    pl.plot(x_test, exact_mu, '-')
    pl.pause(0.001)

    for log_len_u in xrange(3, 12):
        len_u = 2**log_len_u
        print len_u

        mu, gp_samples = approx_gp_sample(x, y, x_test,
                                      gp_params, indep_noise, eps, len_u,
                                      lanczos_basis=10, return_mean=True)
        pl.plot(x_test, mu, '-')
        pl.pause(0.001)

        print np.linalg.norm(gp_samples - exact_samples)
        print
    pl.ioff()
    pl.show()

def gen_data(n_data=500, gp_params=(1, 10), indep_noise=.01):
    x = np.random.uniform(size=n_data)
    #x = np.float32(x)
    x = np.sort(x)
    a, b = gp_params
    c = indep_noise
    mu = np.zeros(n_data)
    cov = a * np.exp(-b * (x[:, np.newaxis] - x)**2) + c * np.eye(n_data)
    y = np.random.multivariate_normal(mu, cov)
    return x, y

def read_data(data='ECG200/11_1000_60_dat.pkl'):
    #data = 'ECG200/11_1000_60_dat.pkl'
    gp_parms, ts_train, ts_test, l_train, l_test = pickle_load(data)[:5]
    x_train = np.array([each_ts[0] for each_ts in ts_train])
    y_train = np.array([each_ts[1] for each_ts in ts_train])
    eg_id = 0
    x = x_train[eg_id]
    y = y_train[eg_id]
    return x, y

def cmp_lanczos_basis(n_data=500):
    #n_data = 500
    print '# data:', n_data
    gp_params = np.array([.1, 100])
    indep_noise = .001
    x, y = gen_data(n_data, gp_params, indep_noise)

    #import pylab as pl
    #pl.plot(x, y, '.-')
    #pl.show()
    #return

    n_test = n_data

    x_test = np.linspace(x.min(), x.max(), n_test)
    batch_size = 1
    eps = np.random.normal(size=(batch_size, len(x_test)))

    exact_samples = exact_gp_sample(x, y, x_test,
                                    gp_params, indep_noise, eps)

    len_u = 256
    errors = []
    #for n_basis in xrange(1, 51):
    for n_basis in xrange(1, 21):
        print '# basis:', n_basis
        approx_samples = approx_gp_sample(x, y, x_test, gp_params,
                                          indep_noise, eps, len_u,
                                          lanczos_basis=n_basis)
        error = np.linalg.norm(approx_samples - exact_samples)
        print error
        errors.append(error)
        #results.append(error)
    return np.array(errors)

    #results = []
    import pylab as pl
    pl.figure()
    pl.ion()
    #for log_len_u in xrange(3, 13):
    for log_len_u in [8]:
        len_u = 2**log_len_u
        print log_len_u, len_u

        errors = []
        #for n_basis in xrange(1, 51):
        for n_basis in xrange(1, 21):
            print '# basis:', n_basis
            approx_samples = approx_gp_sample(x, y, x_test, gp_params,
                                              indep_noise, eps, len_u,
                                              lanczos_basis=n_basis)
            error = np.linalg.norm(approx_samples - exact_samples)
            print error
            errors.append(error)
            #results.append(error)
        pl.semilogy(errors, '.-')
        pl.pause(.001)
        print
    pl.ioff()
    pl.show()

    #return np.array(results)

def cmp_n_inducing_points(n_data=500):
    #n_data = 500
    print '# data:', n_data
    gp_params = np.array([.1, 100])
    indep_noise = .001
    x, y = gen_data(n_data, gp_params, indep_noise)

    #import pylab as pl
    #pl.plot(x, y, '.-')
    #pl.show()
    #return

    n_test = n_data

    x_test = np.linspace(x.min(), x.max(), n_test)
    batch_size = 1
    eps = np.random.normal(size=(batch_size, len(x_test)))

    exact_samples = exact_gp_sample(x, y, x_test,
                                    gp_params, indep_noise, eps)

    results = []
    for log_len_u in xrange(3, 13):
        len_u = 2**log_len_u
        print log_len_u, len_u

        approx_samples = approx_gp_sample(x, y, x_test,
                                          gp_params, indep_noise, eps, len_u)

        error = np.linalg.norm(approx_samples - exact_samples)
        print error
        results.append(error)
        print

    return np.array(results)

def cmp_time():
    n_data = 500
    gp_params = np.array([1., 10.])
    indep_noise = .1

    x, y = gen_data(n_data, gp_params, indep_noise)
    #x, y = read_data('ECG200/11_1000_60_dat.pkl')

    #print x
    #print y
    for log_u in xrange(4, 13):
        len_u = 2**log_u + 1
        print log_u, 2**log_u

        x_test = np.linspace(x.min(), x.max(), len_u)
        batch_size = 1
        eps = np.random.normal(size=(batch_size, len(x_test)))

        t1 = time.time()
        approx_gp_sample(x, y, x_test, gp_params, indep_noise, eps, len_u)
        t2 = time.time()
        print t2 - t1

        t1 = time.time()
        exact_gp_sample(x, y, x_test, gp_params, indep_noise,
                        eps, decomp='cholesky')
        t2 = time.time()
        print t2 - t1

        t1 = time.time()
        exact_gp_sample(x, y, x_test, gp_params, indep_noise,
                        eps, decomp='sqrtm')
        t2 = time.time()
        print t2 - t1

        print
        #print '-' * 40
        #print (cov_zs**2).sum(axis=0)
        #print (cov_zs_exact**2).sum(axis=0)

if __name__ == '__main__':
    np.random.seed(0)
    #cmp_time()
    #errors = cmp_lanczos_basis(500)

    for i in xrange(10):
        results = []
        for n_data in xrange(500, 3000 + 1, 500):
            errors = cmp_lanczos_basis(n_data)
            results.append(errors)
        pickle_save('results/lanczos-%02d.pkl' % i, np.vstack(results))

    '''
    for i in xrange(10):
        results = []
        for n_data in xrange(200, 3000 + 1, 200):
            errors = cmp_n_inducing_points(n_data)
            results.append(errors)
        pickle_save('results/error-%02d.pkl' % i, np.vstack(results))
    '''

    #norm_diff()

max_stars_repo_name: steveli/gp-adapter

max_issues_repo_name: steveli/gp-adapter

max_forks_repo_name: steveli/gp-adapter


text:
# Author: Sebastian Raschka

import tensorflow as tf
import numpy as np

from IPython.display import clear_output, Image, display, HTML

def strip_consts(graph_def, max_const_size=32):
    """Strip large constant values from graph_def."""
    strip_def = tf.GraphDef()
    for n0 in graph_def.node:
        n = strip_def.node.add()
        n.MergeFrom(n0)
        if n.op == 'Const':
            tensor = n.attr['value'].tensor
            size = len(tensor.tensor_content)
            if size > max_const_size:
                tensor.tensor_content = b"<stripped %d bytes>"%size
    return strip_def

def show_graph(graph_def, max_const_size=32):
    """Visualize TensorFlow graph."""
    if hasattr(graph_def, 'as_graph_def'):
        graph_def = graph_def.as_graph_def()
    strip_def = strip_consts(graph_def, max_const_size=max_const_size)
    code = """

        <link rel="import" href="https://tensorboard.appspot.com/tf-graph-basic.build.html" onload=load()>

          <tf-graph-basic id="{id}"></tf-graph-basic>

    """.format(data=repr(str(strip_def)), id='graph'+str(np.random.rand()))

    iframe = """

    """.format(code.replace('"', '"'))
    display(HTML(iframe))

ext: py

max_stars_repo_name: losandywy/MyDataX

max_issues_repo_name: losandywy/MyDataX

max_forks_repo_name: losandywy/MyDataX

reason: import numpy


text:
import pandas as pd
import numpy as np

def update_covar_and_loci(covar, loci, pos_to_delete_below, partition_file):

    covar = covar[covar.i >= pos_to_delete_below]
    loci = loci[loci >= pos_to_delete_below]

    to_add = read_partition(partition_file)

    covar = covar.append(to_add).reset_index(drop=True)
    loci = loci.append(to_add.i.drop_duplicates()).reset_index(drop=True)

    g = covar.i.reset_index().groupby("i").index
    covar_starts = g.first().values
    covar_ends = g.last().values + 1

    return covar, loci, covar_starts, covar_ends

def find_start_locus(curr_locus, loci, cutoff):

    if curr_locus < 0:
        for i, locus in enumerate(loci):
            if locus >= cutoff:
                return i, locus
    else:
        return loci.index(curr_locus), curr_locus

def find_end_locus(p_num, partitions, loci, cutoff):

    if p_num + 1 < len(partitions):
        end_locus = partitions.iloc[p_num + 1, 0]
        end_locus_index = -1
        return end_locus_index, end_locus
    else:
        for i in reversed(range(0, len(loci))):
            if loci[i] <= cutoff:
                return i, loci[i]

def read_partitions(pfiles):

    if pfiles:
        df = pd.concat([read_partition(f) for f in pfiles]).reset_index(drop=True)
        return df, df.i.drop_duplicates().reset_index(drop=True)
    else:
        df = pd.DataFrame({"i": np.array([], dtype=int), "j": np.array([], dtype=int), "val": np.array([], dtype=float)})
        return df, df.i

def read_partition(pfile):

    if pfile.endswith(".pq"):
        return pd.read_parquet(pfile)
    else:
        return pd.read_table(pfile, sep=" ", usecols=[2, 3, 7], names="i j val".split())

def covar_files_map(covariance_files, partitions):

    """Map of interval to covariance file."""

    covar_dict = {}
    for f, (start, end) in zip(covariance_files, partitions.itertuples(index=False)):
        covar_dict[start, end] = f

    return covar_dict

def files_to_preread(partitions, covar_dict, larger_than):

    i = 0
    to_preread = []
    while i < len(partitions) - 1 and larger_than >= partitions[i + 1][0]:
        pstart, pend = partitions[i]
        to_preread.append(covar_dict[pstart, pend])
        i += 1

    return to_preread, i

def preread_files(partitions, covar_dict, larger_than):

    to_preread, i = files_to_preread(partitions, covar_dict, larger_than)
    covar, loci = read_partitions(to_preread)

    return covar, loci, i


text:
import json
import matplotlib.pyplot as plt
import numpy as np
from scipy.optimize import curve_fit

noapproxX = []
exactDecompX = []
LsampleX = []

noapproxY = []
exactDecompY = []
LsampleY = []

noapproxYerr = []
exactDecompYerr = []
LsampleYerr = []

L = 1000

f = open("performance.txt")
for line in f.readlines():
    try:
        data = json.loads(line)
    except:
        continue

    if data["mode"] == "noapprox":
        noapproxX.append(data["k"])
        noapproxY.append(data["mutime"])
        noapproxYerr.append(data["sigmatime"]/(data["ntests"]*data["ntests"]))
    if data["mode"] == "exactDecomp":
        exactDecompX.append(data["k"])
        exactDecompY.append(L*data["mutime"]/data["L"])
        # exactDecompYerr.append(data["sigmatime"]/(data["L"]*data["ntests"]*data["ntests"]))
        exactDecompYerr.append(L*data["sigmatime"]/(data["L"]))
    if data["mode"] == "Lsample":
        LsampleX.append(data["k"])
        LsampleY.append(L*data["mutime"]/data["L"])
        # LsampleYerr.append(data["sigmatime"]/(data["L"]*data["ntests"]*data["ntests"]))
        LsampleYerr.append(L*data["sigmatime"]/(data["L"]))

f.close()

def func(x,a,b,c):
    return a * 2**(b*x) + c

# plt.errorbar(noapproxX, noapproxY, yerr=noapproxYerr)

popt, _ = curve_fit(func, exactDecompX, exactDecompY)
plt.plot(exactDecompX, func(np.array(exactDecompX), *popt), label="exactDecomp fit")
plt.errorbar(exactDecompX, exactDecompY, yerr=exactDecompYerr, fmt="o", label="exactDecomp")

print("exactDecomp time = ", popt[0],"* 2^(", popt[1], "* k) +", popt[2])

plt.errorbar(LsampleX, LsampleY, yerr=LsampleYerr, fmt="o", label="Lsample")

plt.xlabel("k = 2t")
plt.ylabel("Time per %d samples in seconds" % L)
plt.title("Simulation time for 16 cores over two nodes")
plt.yscale("log")

plt.legend()

plt.show()

max_stars_repo_name: patrickrall/CircuitSimulator

max_issues_repo_name: patrickrall/CircuitSimulator

max_forks_repo_name: patrickrall/CircuitSimulator

reason: import numpy,from scipy


text:
import numpy as np
import os
import glob

classes=["EW","EA","RRab","RRc","RRd","RSCVn","LPV"]

y=np.load("y_test.npy")
pr=np.load("preds.npy")

for clss in classes:
    #l=glob.glob("trainingdata\\saliency\\"+clss+"\\*.png")
    l=glob.glob("keras-vis\\grad_CAM\\gradcam\\"+clss+"\\*.png")
    #l=glob.glob("keras-vis\\saliency\\"+clss+"\\*.png")
    for loc in range(len(l)):
        i=int(l[loc].split("\\")[-1][:-4])
        if y[i]!=pr[i]:
            os.remove(l[loc])


text:
import tensorflow as tf
import cv2
import numpy as np
import matplotlib.pyplot as plt
from tensorflow import keras
from keras import backend as K
from keras.models import Model

#funzione pubblica per visualizzare le heatmap calcolate.
def visualize(model, layer_name, image):
    target_size = (model.input.shape[1], model.input.shape[2])
    prediction = model.predict(image)
    classIdx = np.argmax(prediction[0])
    cam, cam3 = __compute_heatmap(model, layer_name, image, classIdx=classIdx, upsample_size=target_size)
    heatmap = __overlay_gradCAM(image,cam3)
    heatmap = heatmap[..., ::-1] # BGR to RGB
    __vis_heatmap(cam, cam3, heatmap)

#funzione pubblica per visualizzare le heatmap guided calcolate.
def visualize_guided(model, layer_name, image):
    target_size = (model.input.shape[1], model.input.shape[2])
    prediction = model.predict(image)
    classIdx = np.argmax(prediction[0])
    cam, cam3 = __compute_heatmap(model, layer_name, image, classIdx=classIdx, upsample_size=target_size)
    gh_cam = __guided_backprop(__build_guided_model(model, layer_name), image, target_size)
    guided_gradcam = __deprocess_image(gh_cam*cam3)
    gb_cam = __guided_backprop(__build_guided_model(model, layer_name), image, target_size)
    gb_im = __deprocess_image(gb_cam)
    gb_im = gb_im[..., ::-1] # BGR to RGB
    __vis_guided(guided_gradcam, gb_im)

#funzione per il calcolo delle heatmap di uno specifico livello di uno specifico modello.
def __compute_heatmap(model, layer_name, image, upsample_size, classIdx=None, eps=1e-5):
        grad_model = Model(
            inputs=[model.inputs],
            outputs=[model.get_layer(layer_name).output, model.output]
        )
        # registro le operazioni di differenziazione automatica

        with tf.GradientTape() as tape:
            inputs = tf.cast(image, tf.float32)
            (conv_outs, preds) = grad_model(inputs)  # predizioni dopo la softmax
            if classIdx is None:
                classIdx = np.argmax(preds)
            loss = preds[:, classIdx]

        # calcolo dei gradienti con la differenziazione automatica
        grads = tape.gradient(loss, conv_outs)
        # scartare la batch
        conv_outs = conv_outs[0]
        grads = grads[0]
        norm_grads = tf.divide(grads, tf.reduce_mean(tf.square(grads)) + tf.constant(eps))

        # calcolo dei pesi
        weights = tf.reduce_mean(norm_grads, axis=(0, 1))
        cam = tf.reduce_sum(tf.multiply(weights, conv_outs), axis=-1)

        # applicazione della reLU
        camR = np.maximum(cam, 0)
        camR = camR / np.max(camR)
        camR = cv2.resize(camR, upsample_size,cv2.INTER_LINEAR)

        # conversione in 3 dimensioni
        cam3 = np.expand_dims(camR, axis=2)
        cam3 = np.tile(cam3, [1, 1, 3])

        return cam, cam3

#funzione per l'applicazione della grad-cam sull'immagine originale.
def __overlay_gradCAM(img, cam3):
    cam3 = np.uint8(255 * cam3)
    cam3 = cv2.applyColorMap(cam3, cv2.COLORMAP_JET)

    new_img = 0.4 * cam3 / 255 + img

    return new_img

#funzione per plottare heatmap, heatmap con la relu e l'immagini con l'heatmap.
def __vis_heatmap(cam, cam3, heatmap):
    fig, ax = plt.subplots(1, 3, figsize=(15,15))
    ax[0].imshow(cam)
    ax[0].set_title("Cam")
    ax[1].imshow(cam3)
    ax[1].set_title("Cam Relu")
    ax[2].imshow(heatmap[0])
    ax[2].set_title("Image with HeatMap")
    plt.tight_layout()
    plt.show()

#funzione per plottare la grad-cam guidata e la back-propagation guidata.
def __vis_guided(guided_gradcam, guided_backprop):
    fig, ax = plt.subplots(1, 2, figsize=(10,10))
    ax[0].imshow(guided_gradcam)
    ax[0].set_title("Guided GRAD-Cam")
    ax[1].imshow(guided_backprop)
    ax[1].set_title("Guided Back-Propagation")
    plt.tight_layout()
    plt.show()

@tf.custom_gradient
def __guidedRelu(x):
    def grad(dy):
        return tf.cast(dy>0,"float32") * tf.cast(x>0, "float32") * dy
    return tf.nn.relu(x), grad

def __build_guided_model(model, layerName):
        gbModel = Model(
            inputs = [model.inputs],
            outputs = [model.get_layer(layerName).output]
        )
        layer_dict = [layer for layer in gbModel.layers[1:] if hasattr(layer,"activation")]
        for layer in layer_dict:
            if layer.activation == tf.keras.activations.relu:
                layer.activation = __guidedRelu

        return gbModel

#funzione per  la guided backpropagation per visualizzare gli input salienti.
def __guided_backprop(gbModel, images, upsample_size):
        with tf.GradientTape() as tape:
            inputs = tf.cast(images, tf.float32)
            tape.watch(inputs)
            outputs = gbModel(inputs)

        grads = tape.gradient(outputs, inputs)[0]

        saliency = cv2.resize(np.asarray(grads), upsample_size)

        return saliency

def __deprocess_image(x):
    # normalizzazione del tensore: centro su 0., assicura che std sia 0.25
    x = x.copy()
    x -= x.mean()
    x /= (x.std() + K.epsilon())
    x *= 0.25

    # clippa il valore tra [0, 1]
    x += 0.5
    x = np.clip(x, 0, 1)

    # conversione in array RGB
    x *= 255
    if K.image_data_format() == 'channels_first':
        x = x.transpose((1, 2, 0))
    x = np.clip(x, 0, 255).astype('uint8')
    return x

max_stars_repo_name: enrignagna/CovidNN

max_issues_repo_name: enrignagna/CovidNN

max_forks_repo_name: enrignagna/CovidNN


text:
import chainer
from chainer import serializers
from chainer import functions
from chainer import optimizers
from chainer import training
from chainer import iterators
from chainer.dataset import to_device
from chainer.datasets import TransformDataset
from chainer.training import extensions as E
from chainer.dataset.convert import concat_examples
from chainer.datasets.concatenated_dataset import ConcatenatedDataset

import numpy as np
import os
import argparse
from distutils.util import strtobool

# self made
import models.dgcnn as dgcnn
import ply_dataset as pd

def main():
    parser = argparse.ArgumentParser(
        description='DGCNN')
    # parser.add_argument('--conv-layers', '-c', type=int, default=4)
    parser.add_argument('--gpu', '-g', type=int, default=-1)
    parser.add_argument('--num_point', '-n', type=int, default=1024)
    parser.add_argument('--epoch', '-e', type=int, default=250)
    parser.add_argument('--batchsize', '-b', type=int, default=32)
    parser.add_argument('--dropout_ratio', type=float, default=0)
    parser.add_argument('--out', '-o', type=str, default='result')
    parser.add_argument('--model_filename','-m', type=str, default='model.npz')
    parser.add_argument('--resume','-r', type=str, default='')
    parser.add_argument('--use_bn', type=strtobool, default='true')
    parser.add_argument('--residual', type=strtobool, default='false')
    parser.add_argument('--k','-k', type=int, default='20')
    args = parser.parse_args()

    batch_size = args.batchsize
    dropout_ratio = args.dropout_ratio
    num_point = args.num_point
    device = args.gpu
    out_dir = args.out
    epoch = args.epoch
    model_filename = args.model_filename
    resume = args.resume
    use_bn = args.use_bn
    residual = args.residual
    k = args.k

    trans_lam1 = 0.001
    out_dim = 40
    in_dim = 3

    try:
        os.makedirs(out_dir, exist_ok=True)
        import chainerex.utils as cl
        fp = os.path.join(out_dir, 'args.json')
        cl.save_json(fp, vars(args))
        print('save args to', fp)
    except ImportError:
        pass

    gpu = False
    if device >= 0:
        gpu = True

    # Network
    print('Train DGCNN model... use_bn={} dropout={}'
          .format(use_bn, dropout_ratio))
    model = dgcnn.DGCNN(out_dim=out_dim, in_dim=in_dim, dropout_ratio=dropout_ratio, use_bn=use_bn,
                          trans_lam1=trans_lam1, residual=residual, k=k, gpu=gpu)

    print("Dataset setting... num_point={}".format(num_point))
    # Dataset preparation

    train = pd.get_train_dataset(num_point=num_point)
    val = pd.get_test_dataset(num_point=num_point)
    train_iter = iterators.SerialIterator(train, args.batchsize)
    val_iter = iterators.SerialIterator(val, args.batchsize, repeat=False, shuffle=False)

    print("GPU setting...")
    # gpu setting
    if(device >= 0):
        print('using gpu {}'.format(device))
        chainer.backends.cuda.get_device_from_id(device).use()
        model.to_gpu()

    # Optimizer
    optimizer = optimizers.Adam()
    optimizer.setup(model)

    # traning
    converter = concat_examples
    updater = training.StandardUpdater(
        train_iter, optimizer, device=device, converter=converter)
    trainer = training.Trainer(updater, (epoch, 'epoch'), out=out_dir)

    from chainerex.training.extensions import schedule_optimizer_value
    from chainer.training.extensions import observe_value
    # trainer.extend(observe_lr)
    observation_key = 'lr'
    trainer.extend(observe_value(
        observation_key,
        lambda trainer: trainer.updater.get_optimizer('main').alpha))
    trainer.extend(schedule_optimizer_value(
        [10, 20, 100, 150, 200, 230],
        [0.003, 0.001, 0.0003, 0.0001, 0.00003, 0.00001]))

    trainer.extend(E.Evaluator(val_iter, model,
                                converter=converter, device=device))
    trainer.extend(E.PrintReport(
        ['epoch', 'main/loss','main/accuracy','main/cls_loss', 'main/trans_loss',
            'validation/main/loss','validation/main/accuracy','validation/main/cls_loss',
            'validation/main/trans_loss','lr', 'elapsed_time']))
    trainer.extend(E.snapshot(), trigger=(epoch, 'epoch'))
    trainer.extend(E.LogReport())
    trainer.extend(E.ProgressBar(update_interval=10))

    resume = ''
    if resume:
        serializers.load_npz(resume, trainer)
    print("Traning start.")
    trainer.run()

    # --- save classifier ---
    # protocol = args.protocol
    # classifier.save_pickle(
    #     os.path.join(out_dir, args.model_filename), protocol=protocol)
    serializers.save_npz(
        os.path.join(out_dir, model_filename), model)

if __name__ == '__main__':
    main()

max_stars_repo_name: Obarads/DGCNN-Chainer

max_issues_repo_name: Obarads/DGCNN-Chainer


text:
import numpy as np
import cv2

from algorithm.debug import *
from algorithm.Common import scanPointer, meterFinderByTemplate

def doubleArrester(image, info):
    """
    :param image: ROI image
    :param info: information for this meter
    :return: value
    """
    meter = meterFinderByTemplate(image, info["template"])

    ovalCenter = np.array([info["centerPointUp"]["x"], info["centerPointUp"]["y"]])
    ovalStart = np.array([info["startPointUp"]["x"], info["startPointUp"]["y"]])
    ovalEnd = np.array([info["endPointUp"]["x"], info["endPointUp"]["y"]])
    rectCenter = np.array([info["centerPoint"]["x"], info["centerPoint"]["y"]])
    rectStart = np.array([info["startPoint"]["x"], info["startPoint"]["y"]])
    rectEnd = np.array([info["endPoint"]["x"], info["endPoint"]["y"]])
    h, w, _ = meter.shape
    result = {}

    # crop oval meter
    r = int(np.sqrt(np.sum(np.square(ovalCenter - ovalStart))) * 1.6)
    y = max(0, int(ovalCenter[1] - r))
    x = max(0, int(ovalCenter[0] - r))
    width = min(w, x + 2 * r)
    height = min(h, y + 2 * r)
    ovalCenter = np.array([ovalCenter[0] - x, ovalCenter[1] - y])
    ovalStart = np.array([ovalStart[0] - x, ovalStart[1] - y])
    ovalEnd = np.array([ovalEnd[0] - x, ovalEnd[1] - y])
    ovalMeter = meter[y:height, x:width]

    result["ovalMeter"] = scanPointer(ovalMeter, [ovalStart, ovalEnd, ovalCenter], info["startValueUp"],
                                      info["totalValueUp"])
    if ifShow:
        cv2.circle(ovalMeter, (ovalStart[0], ovalStart[1]), 5, (0, 0, 255), -1)
        cv2.circle(ovalMeter, (ovalCenter[0], ovalCenter[1]), 5, (0, 255, 0), -1)
        cv2.circle(ovalMeter, (ovalEnd[0], ovalEnd[1]), 5, (255, 0, 0), -1)
        cv2.imshow("rectMeter", ovalMeter)
        cv2.waitKey(0)
        print("ovalMeter", result["ovalMeter"])

    # crop rectangle meter
    r = int(np.sqrt(np.sum(np.square(rectCenter - rectStart))) * 1.6)
    y = max(0, int(rectCenter[1] - r))
    x = max(0, int(rectCenter[0] - r))
    width = min(w, x + 2 * r)
    height = min(h, y + 2 * r)
    rectCenter = np.array([rectCenter[0] - x, rectCenter[1] - y])
    rectStart = np.array([rectStart[0] - x, rectStart[1] - y])
    rectEnd = np.array([rectEnd[0] - x, rectEnd[1] - y])
    rectMeter = meter[y:height, x:width]
    result["rectMeter"] = scanPointer(rectMeter, [rectStart, rectEnd, rectCenter], info["startValue"],
                                      info["totalValue"])

    if ifShow:
        cv2.circle(rectMeter, (rectStart[0], rectStart[1]), 5, (0, 0, 255), -1)
        cv2.circle(rectMeter, (rectCenter[0], rectCenter[1]), 5, (0, 0, 255), -1)
        cv2.circle(rectMeter, (rectEnd[0], rectEnd[1]), 5, (0, 0, 255), -1)
        cv2.imshow("rectMeter", rectMeter)
        cv2.waitKey(0)
        print("rectMeter", result["rectMeter"])

    return result

max_stars_repo_name: maoyingxue/meterReader

max_issues_repo_name: maoyingxue/meterReader

max_forks_repo_name: maoyingxue/meterReader


text:
import torch
from torch.utils.data import Dataset
import numpy as np
import scipy.io as sio
import utils
import copy
import os
import cv2

class RoomTypeDataset(Dataset):
    def __init__(self, data_root, phase, max_room_per_type):
        self.data_split_root = os.path.join(data_root, phase)
        self.floorplans = os.listdir(self.data_split_root)
        self.max_room_per_type = max_room_per_type
        self.max_room_num = np.array(max_room_per_type).sum()
        self.len = self.__len__()

    def __len__(self):
        return len(self.floorplans)

    def __getitem__(self, index):
        assert index <= self.__len__(), 'index range error'
        floorplan_name = self.floorplans[index]
        floorplan_path = os.path.join(self.data_split_root, floorplan_name)
        floorplan = copy.deepcopy(sio.loadmat(floorplan_path, squeeze_me=True, struct_as_record=False))['data']

        boundary = floorplan.Boundary
        rTypes = floorplan.gt_rTypes

        input_img = self.init_input_img(boundary)
        input_vector = self.init_input_vector(rTypes)

        input_img = torch.FloatTensor(input_img).unsqueeze(0)
        input_img = self.normalize(input_img)
        input_vector = torch.FloatTensor(input_vector)

        return input_img, input_vector

    def normalize(self, data):
        data = torch.div(data, utils.total_num_category-1)
        data = torch.sub(data, 0.5)
        data = torch.div(data, 0.5)
        return data

    def init_input_vector(self, rTypes):
        input_vector = []
        for i in range(13):  # 13 is the total room types
            temp = np.zeros(self.max_room_per_type[i])
            num = (rTypes == i).sum()
            temp[:num] = 1
            input_vector.append(temp)

        input_vector = np.concatenate(input_vector, 0)
        return input_vector

    def init_input_img(self, boundary):
        boundary = boundary.astype(np.int)
        boundary = boundary[:, [1, 0, 2, 3]]

        image = np.ones((128, 128)) * 13
        image = cv2.polylines(image, boundary[:, :2].reshape(1, -1, 2), True, 14, 5)

        for w, h in boundary[:, :2]:
            image[h - 3:h + 4, w - 3:w + 4] = 14

        if boundary[0, 2] == 0:
            image[boundary[0, 1] - 3:boundary[1, 1], boundary[0, 0]: boundary[1, 0]] = 15
        elif boundary[0, 2] == 1:
            image[boundary[0, 1]:boundary[1, 1], boundary[0, 0] + 1: boundary[1, 0] + 4] = 15
        elif boundary[0, 2] == 2:
            image[boundary[0, 1] + 1:boundary[1, 1] + 4, boundary[1, 0]: boundary[0, 0]] = 15
        elif boundary[0, 2] == 3:
            image[boundary[1, 1]:boundary[0, 1], boundary[0, 0] - 3: boundary[1, 0]] = 15

        image = cv2.fillPoly(image, boundary[:, :2].reshape(1, -1, 2), 16)
        return image

max_stars_repo_name: realcrane/floorplan

max_issues_repo_name: realcrane/floorplan

max_forks_repo_name: realcrane/floorplan

reason: import numpy,import scipy


text:
from .pre_proc import Datafile, Dataset

from datetime import datetime, timedelta
import numpy as np
import math as math
import matplotlib.pyplot as plt
import matplotlib.cm as cm
from matplotlib.pyplot import rc_context
import os
import pandas as pd
from time import perf_counter

# specify the three colours to use in plots
OPT=['#347f3a','#36358b','#e47327']

# specify a rc_file which contains plotting parameters
plot_config_file = 'plotting_params.txt'
dir = os.path.abspath(os.path.dirname(__file__))
rc_fname=os.path.join(dir, plot_config_file)

class SV:

    """ Stern Volmer class for converting spectral information for Stern Volmer
            analysis.

	Attributes:
        abs representing the absorbance of the solution at the specified
            wavelength
        counts representing the non-normalised fluorescence counts (float)
        dataset_f representing a Dataset object specified in pre_proc.py for
            the sample fluorescence.
        dataset_a representing a Dataset object specified in pre_proc.py for
            the sample absorbance.
        ncounts representing the normalised fluorescence counts (float)
        normalise representing a Boolean on whether to normalise counts or not
        quencher representing the quencher concentration (float)
        state representing the status of the data (string)
        wavelength representing the peak fluorescence wavelength (float)

    """

    def __init__(self,dataset_f,dataset_a,wavelength,quencher,normalise=True):

        self.dataset_f = dataset_f
        self.dataset_a = dataset_a
        self.normalise = normalise
        self.quencher = quencher
        self.state = 'pre_proc'
        self.wavelength = wavelength

    def nearest_value(self,value,value_arr):

        """Function to find the nearest value in a value array

		Args:
			value representing the value you wish to find (float)
            value_array representing the array you wish to search through (array)

		Returns:
			float: nearest value to that specfied

		"""

        idx = np.searchsorted(value_arr, value, side="left")
        if idx > 0 and (idx == len(value_arr) or math.fabs(value- value_arr[idx-1]) < math.fabs(value - value_arr[idx])):
            return value_arr[idx-1]
        else:
            return value_arr[idx]

    def widen_range(self,value,value_arr,range):

        """Function to return the nearest values for the specified
                range defined by [Value-Range/2, Value + Range/2]

		Args:
			value representing the value you wish to find (float)
            value_array representing the array you wish to search through (array)
            range representing the range you wish to expand to (float)

		Returns:
			Two floats: lowest value and highest value for range

		"""

        middle = self.nearest_value(value, value_arr)
        return self.nearest_value(value - range/2 , value_arr), self.nearest_value(value + range/2 , value_arr)

    def calculate_counts(self,range):

        """Function to return the absorbance-normalised counts

		Args:
			range representing the range either side of the central wavelength to
                average counts over (float)

		Returns:
			float: averaged and absorbance-normalised counts

		"""

        counts_f_arr=self.dataset_f.pre_proc_data
        low_limit, high_limit = self.widen_range(self.wavelength,self.dataset_f.wavelengths,range)

        print('Calculating average counts using range {} nm to {} nm.'.format(low_limit, high_limit))

        # temp stores the count values in the wavelength range
        temp = counts_f_arr.loc[:,low_limit:high_limit]
        print('Counts will be averaged over {} data points'.format((np.array(temp)).shape[1]))

        # average counts over the wavelength range and over time
        counts_f = temp.mean()
        self.counts = counts_f

        # check whether counts should be normalised
        if self.normalise is True:

            # apply similar process to absorbance data
            abs_a_arr=self.dataset_a.abs_data
            low_limit, high_limit = self.widen_range(self.wavelength,self.dataset_a.wavelengths,range)
            temp = abs_a_arr.loc[:,low_limit:high_limit]
            abs_a = temp.mean()
            self.abs = abs_a

        else:
            self.abs = 1

        self.ncounts = self.counts*self.abs
        self.state = 'calculated'

        return self.ncounts

    def __repr__(self):

        """Function to output the characteristics of the Stern Volmer instance

		Args:
			None

		Returns:
			string: characteristics of the Stern Volmer instance

		"""

        return str(self.dataset.exp_label)

max_stars_repo_name: takashilawson/optofluidics

max_issues_repo_name: takashilawson/optofluidics

max_forks_repo_name: takashilawson/optofluidics


text:
"""DVS128.

Author: Yuhuang Hu
Email : duguyue100@gmail.com
"""
from __future__ import print_function, absolute_import
from builtins import range
import numpy as np

from pyaer import libcaer
from pyaer.device import USBDevice
from pyaer.filters import DVSNoise
from pyaer import utils

class DVS128(USBDevice):
    """DVS128.

    # Arguments
        device_id: `int`

            a unique ID to identify the device from others.
            Will be used as the source for EventPackets being
            generate from its data.

            `default is 1`
        bus_number_restrict: `int`

            restrict the search for viable devices to only this USB
            bus number.

            `default is 0`
        dev_address_restrict: `int`

            restrict the search for viable devices to only this USB
            device address.

            `default is 0`
        serial_number: `str`

            restrict the search for viable devices to only devices which do
            possess the given Serial Number in their USB
            SerialNumber descriptor.

            `default is ""`
        noise_filter: `bool`

            if enable noise filter,

            `default is False`.
    """
    def __init__(self,
                 device_id=1,
                 bus_number_restrict=0,
                 dev_address_restrict=0,
                 serial_number="",
                 noise_filter=False):
        """DVS128."""
        super(DVS128, self).__init__()
        # open device
        self.open(device_id, bus_number_restrict,
                  dev_address_restrict, serial_number)
        # get camera information
        self.obtain_device_info(self.handle)

        # noise filter
        self.filter_noise = noise_filter
        if noise_filter is True:
            self.noise_filter = DVSNoise(self.dvs_size_X, self.dvs_size_Y)
        else:
            self.noise_filter = None

    def set_noise_filter(self, noise_filter):
        """Set noise filter.

        # Arguments
            noise_filter: `filters.DVSNoise`

                A valid `DVSNoise` object. This filter implements
                software-level background activity filter.
        """
        if noise_filter is not None:
            self.noise_filter = noise_filter

    def enable_noise_filter(self):
        """Enalbe DVS noise filter.

        This function enables the DVS noise filter.
        Note that this function will initialize a `DVSNoise` filter
        if there is None.
        """
        if self.filter_noise is False:
            self.filter_noise = True

        if self.noise_filter is None:
            self.noise_filter = DVSNoise(self.dvs_size_X, self.dvs_size_Y)

    def disable_noise_filter(self):
        """Disable noise filter.

        This method disable the noise filter. Note that this function
        doesn't destroy the existed noise filter. It simply switches off
        the function.
        """
        if self.filter_noise is True:
            self.filter_noise = False

    def obtain_device_info(self, handle):
        """Obtain DVS128 info.

        This function collects the following information from the device:

        - Deveice ID
        - If the device is a master camera
        - Device string
        - Device USB bus number
        - Device USB device address
        - Camera width
        - Camera height
        - Logic version

        # Arguments
            handle: `caerDeviceHandle`

                a valid device handle that can be used with the other
                `libcaer` functions, or `None` on error.
        """
        if handle is not None:
            info = libcaer.caerDVS128InfoGet(handle)

            # port all info data field out
            self.device_id = info.deviceID
            self.device_serial_number = info.deviceSerialNumber
            self.device_usb_bus_number = info.deviceUSBBusNumber
            self.device_usb_device_address = info.deviceUSBDeviceAddress
            self.device_string = info.deviceString
            self.firmware_version = info.firmwareVersion
            self.device_is_master = info.deviceIsMaster
            self.dvs_size_X = info.dvsSizeX
            self.dvs_size_Y = info.dvsSizeY

    def open(self,
             device_id=1,
             bus_number_restrict=0,
             dev_address_restrict=0,
             serial_number=""):
        """Open device.

        # Arguments
            device_id: `int`

                a unique ID to identify the device from others.
                Will be used as the source for EventPackets being
                generate from its data.

                `default is 1`.
            bus_number_restrict: `int`

                restrict the search for viable devices to only this USB
                bus number.

                `default is 0`.
            dev_address_restrict: `int`

                restrict the search for viable devices to only this USB
                device address.

                `default is 0`.
            serial_number: `str`

                restrict the search for viable devices to only devices which do
                possess the given Serial Number in their USB
                SerialNumber descriptor.

                `default is ""`
        """
        super(DVS128, self).open(
            libcaer.CAER_DEVICE_DVS128, device_id,
            bus_number_restrict, dev_address_restrict,
            serial_number)

    def set_bias_from_json(self, file_path, verbose=False):
        """Set bias from loading JSON configuration file.

        # Arguments
            file_path: `str`

                absolute path of the JSON bias file.
            verbose: `bool`

                optional debugging message.
        """
        bias_obj = utils.load_dvs_bias(file_path, verbose)
        self.set_bias(bias_obj)

    def set_bias(self, bias_obj):
        """Set bias from bias dictionary.

        # Arguments
            bias_obj: `dict`

                dictionary that contains DVS128 biases.

        # Returns
            flag: `bool`

                True if set successful, False otherwise.
        """
        self.set_config(libcaer.DVS128_CONFIG_BIAS,
                        libcaer.DVS128_CONFIG_BIAS_CAS,
                        bias_obj["cas"])
        self.set_config(libcaer.DVS128_CONFIG_BIAS,
                        libcaer.DVS128_CONFIG_BIAS_INJGND,
                        bias_obj["injGnd"])
        self.set_config(libcaer.DVS128_CONFIG_BIAS,
                        libcaer.DVS128_CONFIG_BIAS_REQPD,
                        bias_obj["reqPd"])
        self.set_config(libcaer.DVS128_CONFIG_BIAS,
                        libcaer.DVS128_CONFIG_BIAS_PUX,
                        bias_obj["puX"])
        self.set_config(libcaer.DVS128_CONFIG_BIAS,
                        libcaer.DVS128_CONFIG_BIAS_DIFFOFF,
                        bias_obj["diffOff"])
        self.set_config(libcaer.DVS128_CONFIG_BIAS,
                        libcaer.DVS128_CONFIG_BIAS_REQ,
                        bias_obj["req"])
        self.set_config(libcaer.DVS128_CONFIG_BIAS,
                        libcaer.DVS128_CONFIG_BIAS_REFR,
                        bias_obj["refr"])
        self.set_config(libcaer.DVS128_CONFIG_BIAS,
                        libcaer.DVS128_CONFIG_BIAS_PUY,
                        bias_obj["puY"])
        self.set_config(libcaer.DVS128_CONFIG_BIAS,
                        libcaer.DVS128_CONFIG_BIAS_DIFFON,
                        bias_obj["diffOn"])
        self.set_config(libcaer.DVS128_CONFIG_BIAS,
                        libcaer.DVS128_CONFIG_BIAS_DIFF,
                        bias_obj["diff"])
        self.set_config(libcaer.DVS128_CONFIG_BIAS,
                        libcaer.DVS128_CONFIG_BIAS_FOLL,
                        bias_obj["foll"])
        self.set_config(libcaer.DVS128_CONFIG_BIAS,
                        libcaer.DVS128_CONFIG_BIAS_PR,
                        bias_obj["Pr"])
        # setting for noise filter
        if self.filter_noise is True:
            self.noise_filter.set_bias(bias_obj["noise_filter_configs"])

    def get_bias(self):
        """Get bias settings.

        # Returns
            bias_obj: `dict`

                dictionary that contains DVS128 current bias settings.
        """
        bias_obj = {}
        bias_obj["cas"] = self.get_config(
            libcaer.DVS128_CONFIG_BIAS,
            libcaer.DVS128_CONFIG_BIAS_CAS)
        bias_obj["injGnd"] = self.get_config(
            libcaer.DVS128_CONFIG_BIAS,
            libcaer.DVS128_CONFIG_BIAS_INJGND)
        bias_obj["reqPd"] = self.get_config(
            libcaer.DVS128_CONFIG_BIAS,
            libcaer.DVS128_CONFIG_BIAS_REQPD)
        bias_obj["puX"] = self.get_config(
            libcaer.DVS128_CONFIG_BIAS,
            libcaer.DVS128_CONFIG_BIAS_PUX)
        bias_obj["diffOff"] = self.get_config(
            libcaer.DVS128_CONFIG_BIAS,
            libcaer.DVS128_CONFIG_BIAS_DIFFOFF)
        bias_obj["req"] = self.get_config(
            libcaer.DVS128_CONFIG_BIAS,
            libcaer.DVS128_CONFIG_BIAS_REQ)
        bias_obj["refr"] = self.get_config(
            libcaer.DVS128_CONFIG_BIAS,
            libcaer.DVS128_CONFIG_BIAS_REFR)
        bias_obj["puY"] = self.get_config(
            libcaer.DVS128_CONFIG_BIAS,
            libcaer.DVS128_CONFIG_BIAS_PUY)
        bias_obj["diffOn"] = self.get_config(
            libcaer.DVS128_CONFIG_BIAS,
            libcaer.DVS128_CONFIG_BIAS_DIFFON)
        bias_obj["diff"] = self.get_config(
            libcaer.DVS128_CONFIG_BIAS,
            libcaer.DVS128_CONFIG_BIAS_DIFF)
        bias_obj["foll"] = self.get_config(
            libcaer.DVS128_CONFIG_BIAS,
            libcaer.DVS128_CONFIG_BIAS_FOLL)
        bias_obj["Pr"] = self.get_config(
            libcaer.DVS128_CONFIG_BIAS,
            libcaer.DVS128_CONFIG_BIAS_PR)

        # get noise filter configs
        if self.noise_filter is not None:
            bias_obj["noise_filter_configs"] = self.noise_filter.get_bias()

        return bias_obj

    def save_bias_to_json(self, file_path):
        """Save bias to JSON.

        # Arguments
            file_path: `str`

                the absolute path to the destiation.

        # Returns
            flag: `bool`

                returns True if success in writing, False otherwise.
        """
        bias_obj = self.get_bias()
        return utils.write_json(file_path, bias_obj)

    def start_data_stream(self, send_default_config=True):
        """Start streaming data.

        # Arguments
            send_default_config: `bool`

                send default config to the device before starting
                the data streaming.

                `default is True`
        """
        if send_default_config is True:
            self.send_default_config()
        self.data_start()
        self.set_data_exchange_blocking()

    def get_polarity_event(self, packet_header, noise_filter=False):
        """Get a packet of polarity event.

        # Arguments
            packet_header: `caerEventPacketHeader`

                the header that represents a event packet
            noise_filter: `bool`

                the background activity filter is applied if True.

        # Returns
            events: `numpy.ndarray`

                a 2-D array that has the shape of (N, 4) where N
                is the number of events in the event packet.
                Each row in the array represents a single polarity event.
                The first number is the timestamp.
                The second number is the X position of the event.
                The third number is the Y position of the event.
                The fourth number represents the polarity of the event
                (positive or negative).

                If the `noise_filter` option is set to `True`,
                this array has an additional column at the end.
                The last column represents the validity of the corresponding
                event. Filtered events will be marked as 0.
            num_events: `int`

                number of the polarity events available in the packet.

        """
        num_events, polarity = self.get_event_packet(
            packet_header, libcaer.POLARITY_EVENT)

        if noise_filter is True:
            polarity = self.noise_filter.apply(polarity)

            events = libcaer.get_filtered_polarity_event(
                polarity, num_events*5).reshape(num_events, 5)
        else:
            events = libcaer.get_polarity_event(
                polarity, num_events*4).reshape(num_events, 4)

        return events, num_events

    def get_event(self, mode="events"):
        """Get event.

        # Returns
            pol_events: `numpy.ndarray`

                a 2-D array that has the shape of (N, 4) where N
                is the number of events in the event packet.
                Each row in the array represents a single polarity event.
                The first number is the timestamp.
                The second number is the X position of the event.
                The third number is the Y position of the event.
                The fourth number represents the polarity of the event
                (positive or negative).

                If the `noise_filter` option is set to `True`,
                this array has an additional column at the end.
                The last column represents the validity of the corresponding
                event. Filtered events will be marked as 0.
            num_pol_events: `int`

                number of the polarity events available in the packet.
            special_events: `numpy.ndarray`

                a 2-D array that has the shape of (N, 2) where N
                is the number of events in the event packet.
                Each row in the array represents a single special event.
                The first value is the timestamp of the event.
                The second value is the special event data.
            num_special_events: `int`

                number of the special events in the packet.
        """
        packet_container, packet_number = self.get_packet_container()
        if packet_container is not None:
            num_pol_event = 0
            num_special_event = 0
            pol_events = None
            special_events = None
            for packet_id in range(packet_number):
                packet_header, packet_type = self.get_packet_header(
                    packet_container, packet_id)
                if packet_type == libcaer.POLARITY_EVENT:
                    if mode == "events":
                        events, num_events = self.get_polarity_event(
                            packet_header, self.filter_noise)
                        pol_events = np.hstack((pol_events, events)) \
                            if pol_events is not None else events
                    elif mode == "events_hist":
                        hist, num_events = self.get_polarity_hist(
                            packet_header, device_type="DVS128")
                        pol_events = hist if pol_events is None \
                            else pol_events+hist

                    num_pol_event += num_events
                elif packet_type == libcaer.SPECIAL_EVENT:
                    events, num_events = self.get_special_event(
                        packet_header)
                    special_events = np.hstack((special_events, events)) \
                        if special_events is not None else events
                    num_special_event += num_events
            libcaer.caerEventPacketContainerFree(packet_container)

            return (pol_events, num_pol_event, special_events,
                    num_special_event)
        else:
            return None

max_stars_repo_name: franciumlee/pyaer

max_issues_repo_name: franciumlee/pyaer

max_forks_repo_name: franciumlee/pyaer


text:
"""Implementations for classes in q_learning_v3.py."""

from typing import Iterable

import numpy
from keras import layers
from keras import models

from lib.q_learning_v3 import *

class KerasModelQFunctionOneHotAction(QFunction):
    """A Q-Function implementation using a model built in Keras.

    This version uses one-hot vector for actions.
    """

    def __init__(
            self,
            env: Environment,
            num_nodes_in_layers: Iterable[int],
            learning_rate: float = None,
            discount_factor: float = None,
    ):
        """Constructor.

        Args:
            env: environment.
            num_nodes_in_layers: a list of how many nodes are used in each
                layer, starting from the input layter.
        """
        super().__init__(
            learning_rate=learning_rate,
            discount_factor=discount_factor)

        self._env = env
        self._state_array_size = env.GetStateArraySize()
        self._action_space_size = len(env.GetActionSpace())
        self._input_size = self._state_array_size + self._action_space_size

        self._model = _BuildModelOneHotAction(
            self._env.GetStateArraySize(),
            len(self._env.GetActionSpace()),
            num_nodes_in_layers)

        # For creating copy.
        self._num_nodes_in_layers = num_nodes_in_layers
        self._learning_rate = learning_rate
        self._discount_factor = discount_factor

    # @Override
    def MakeCopy(self) -> 'KerasModelQFunctionOneHotAction':
        new_q_func = KerasModelQFunctionOneHotAction(
            self._env, self._num_nodes_in_layers,
            learning_rate=self._learning_rate,
            discount_factor=self._discount_factor)
        self.UpdateCopy(new_q_func)
        return new_q_func

    # @Override
    def UpdateCopy(self, target: 'KerasModelQFunctionOneHotAction') -> None:
        target._model.set_weights(self._model.get_weights())
        target.debug_verbosity = self.debug_verbosity

    def GetValue(
            self,
            state: State,
            action: Action,
    ) -> float:
        value = self._model.predict(self._GetStateActionArray(state, action))
        if self.debug_verbosity >= 5:
            print('GET: (%s, %s) -> %s' % (state, action, value))
        return value

    # @Override
    def SetValue(
            self,
            state: State,
            action: Action,
            new_value: float,
    ) -> None:
        if self.debug_verbosity >= 5:
            print('SET: (%s, %s) <- %s' % (state, action, new_value))
        return self._model.fit(
            self._GetStateActionArray(state, action), new_value, verbose=0)

    def _GetStateActionArray(
            self,
            state: State,
            action: Action,
    ) -> numpy.ndarray:
        """Creates a (state, action) array."""
        input_array = numpy.zeros((1, self._input_size))
        input_array[0, :self._state_array_size] = state
        # Use hot-spot for action.
        input_array[0, self._state_array_size + action] = 1.0
        return input_array

def _BuildModelOneHotAction(
        state_array_size: int,
        action_space_size: int,
        num_nodes_in_layers: Iterable[int],
) -> models.Model:
    """Builds a model with the given info; use one-hot vector for action."""
    input_size = state_array_size + action_space_size
    model = models.Sequential()
    model.add(
        layers.Dense(input_size, activation='relu', input_dim=input_size))
    for num_nodes in num_nodes_in_layers:
        model.add(layers.Dense(num_nodes, activation='relu'))
    model.add(layers.Dense(1))

    model.compile(optimizer='sgd', loss='mse')

    return model

class KerasModelQFunctionIntAction(QFunction):
    """A Q-Function implementation using a model built in Keras.

    This version uses an int for actions.
    """

    def __init__(
            self,
            env: Environment,
            num_nodes_in_layers: Iterable[int],
            learning_rate: float = None,
            discount_factor: float = None,
    ):
        """Constructor.

        Args:
            env: environment.
            num_nodes_in_layers: a list of how many nodes are used in each
                layer, starting from the input layter.
        """
        super().__init__(
            learning_rate=learning_rate,
            discount_factor=discount_factor)

        self._env = env
        self._state_array_size = env.GetStateArraySize()
        self._action_space_size = len(env.GetActionSpace())
        self._input_size = self._state_array_size + 1

        self._model = _BuildModelIntAction(
            self._env.GetStateArraySize(),
            num_nodes_in_layers)

        # For faster input vector creation.
        self._input_vector = numpy.zeros((1, self._input_size))

        # For creating copy.
        self._num_nodes_in_layers = num_nodes_in_layers
        self._learning_rate = learning_rate
        self._discount_factor = discount_factor

    # @Override
    def MakeCopy(self) -> 'KerasModelQFunctionIntAction':
        new_q_func = KerasModelQFunctionIntAction(
            self._env, self._num_nodes_in_layers,
            learning_rate=self._learning_rate,
            discount_factor=self._discount_factor)
        self.UpdateCopy(new_q_func)
        return new_q_func

    # @Override
    def UpdateCopy(self, target: 'KerasModelQFunctionIntAction') -> None:
        target._model.set_weights(self._model.get_weights())
        target.debug_verbosity = self.debug_verbosity

    def GetValue(
            self,
            state: State,
            action: Action,
    ) -> float:
        value = self._model.predict(self._GetStateActionArray(state, action))
        if self.debug_verbosity >= 5:
            print('GET: (%s, %s) -> %s' % (state, action, value))
        return value

    # @Override
    def SetValue(
            self,
            state: State,
            action: Action,
            new_value: float,
    ) -> None:
        if self.debug_verbosity >= 5:
            print('SET: (%s, %s) <- %s' % (state, action, new_value))
        return self._model.fit(
            self._GetStateActionArray(state, action), new_value, verbose=0)

    def _GetStateActionArray(
            self,
            state: State,
            action: Action,
    ) -> numpy.ndarray:
        """Creates a (state, action) array."""
        self._input_vector[0, :self._state_array_size] = state
        self._input_vector[-1] = action
        return self._input_vector

def _BuildModelIntAction(
        state_array_size: int,
        num_nodes_in_layers: Iterable[int],
) -> models.Model:
    """Builds a model with the given info; using an int for action."""
    input_size = state_array_size + 1
    model = models.Sequential()
    model.add(
        layers.Dense(input_size, activation='relu', input_dim=input_size))
    for num_nodes in num_nodes_in_layers:
        model.add(layers.Dense(num_nodes, activation='relu'))
    model.add(layers.Dense(1))

    model.compile(optimizer='sgd', loss='mse')

    return model

class MultiModelQFunction(QFunction):
    """A Q-Function implementation using one model per action."""

    def __init__(
            self,
            env: Environment,
            num_nodes_in_shared_layers: Iterable[int],
            num_nodes_in_multi_head_layers: Iterable[int],
            activation: str = 'relu',
            learning_rate: float = None,
            discount_factor: float = None,
    ):
        """Constructor.

        Args:
            env: the environment.
            num_nodes_in_shared_layers: a list of how many nodes are used in
                each shared layer, starting from the input layter.
            num_nodes_in_multi_head_layers: a list of how many nodes are used
                in the rest of the model for each action, starting from the
                next layer after the last shared layer.
            activation: the activation function.
        """
        super().__init__(
            learning_rate=learning_rate,
            discount_factor=discount_factor)

        self._env = env
        self._models = _BuildMultiHeadModels(
            self._env.GetStateArraySize(),
            self._env.GetActionSpace(),
            num_nodes_in_shared_layers,
            num_nodes_in_multi_head_layers,
            activation)

        # For creating copy.
        self._num_nodes_in_shared_layers = num_nodes_in_shared_layers
        self._num_nodes_in_multi_head_layers = num_nodes_in_multi_head_layers
        self._activation = activation
        self._learning_rate = learning_rate
        self._discount_factor = discount_factor

    # @Override
    def MakeCopy(self) -> 'MultiModelQFunction':
        new_q_func = MultiModelQFunction(
            self._env, self._num_nodes_in_shared_layers,
            self._num_nodes_in_multi_head_layers, activation=self._activation,
            learning_rate=self._learning_rate,
            discount_factor=self._discount_factor)
        self.UpdateCopy(new_q_func)
        return new_q_func

    # @Override
    def UpdateCopy(self, target: 'MultiModelQFunction') -> None:
        target.debug_verbosity = self.debug_verbosity
        for idx, _ in enumerate(self._models):
            target._models[idx].set_weights(self._models[idx].get_weights())

    # @Override
    def GetValue(
            self,
            state: State,
            action: Action,
    ) -> float:
        value = self._models[action].predict(state.reshape(1, state.size))
        if self.debug_verbosity >= 5:
            print('GET: (%s, %s) -> %s' % (state, action, value))
        return value

    # @Override
    def SetValue(
            self,
            state: State,
            action: Action,
            new_value: float,
    ) -> None:
        if self.debug_verbosity >= 5:
            print('SET: (%s, %s) <- %s' % (state, action, new_value))
        self._models[action].fit(
            state.reshape(1, state.size), new_value, verbose=0)

def _BuildMultiHeadModels(
        state_array_size: int,
        action_space: ActionSpace,
        num_nodes_in_shared_layers: Iterable[int],
        num_nodes_in_multi_head_layers: Iterable[int],
        activation: str,
) -> Tuple[models.Model]:
    """Builds a model with the given info."""
    shared_layers = []
    for num_nodes in num_nodes_in_shared_layers:
        shared_layers.append(layers.Dense(num_nodes, activation=activation))

    input_size = state_array_size
    action_models = []
    for action in action_space:
        model = models.Sequential()
        model.add(layers.Dense(
            input_size, activation=activation, input_dim=input_size))
        # First shared layers.
        for layer in shared_layers:
            model.add(layer)
        # Then separated layers.
        for num_nodes in num_nodes_in_multi_head_layers:
            model.add(layers.Dense(num_nodes, activation=activation))
        # Output layer.
        model.add(layers.Dense(1))
        model.compile(optimizer='sgd', loss='mse')
        action_models.append(model)

    return tuple(action_models)