Module `project.stats.pre_post_proc`

Expand source code

import csv
import itertools as it
import multiprocessing as mp
import os
import pickle as p
import random as r
from collections import ChainMap
from os import listdir
from os.path import isfile, join
from pprint import pp, pprint

import matplotlib.pyplot as pl
import numpy as np
import scipy.fft as f
import scipy.signal as s

import project.esn.transformer as t
import project.esn.utils as u
import project.stats.metrics as met

path_resources = "/home/vimmoos/NN/resources/"

path_max = path_resources + "esn/final_pso/"

path_csv = path_resources + "esn_csv/final_pso/"

path_mar = "/home/pasta/Desktop/uni/secondYear/block-2b/NN/NN/project/fitter/dumps/"

path_s = "/home/sneha-lodha/Desktop/Year2Block2B/NeuralNetworks/pso/"
path_sneha = "/home/sneha-lodha/Desktop/Year2Block2B/NeuralNetworks/pso_extract/"

def experiment(path):
    """given a path to a directory take all the file in it and returns a generator

    """
    return (join(path, f) for f in listdir(path) if isfile(join(path, f)))


def get_data (paths):
    """load each file given a list or a generator of path

    """
    for f in paths:
        yield [p.load(open(f, "rb"))]


get_experiment = u.comp(get_data,experiment)

squeeze_fs = [t._identity, t.my_sigm, np.tanh, t.sigmoid]


def apply_metrics(output, desired, raw_output):
    """apply all the possible metrics usign the output and the desired

    """
    return {
        x.name: x.value(output, desired)()
        if x.name != "teacher_loss_nd" else x.value(raw_output, desired)()
        for x in list(met.Metrics)
    }


def apply_transformers(dict_, data_len):
    """apply each tranformer to the output and then apply each metrics to
each output generated by the transformers

    """
    output = dict_["output"][:data_len]
    des = dict_["desired"][:data_len]
    return {
        trans.name: [{
            "param": (val := ((param * 2) / 10) + 0.2),
            "squeeze_f":
            squeeze_f.__name__,
            **apply_metrics(trans.value(val, squeeze_f)(output), des, output)
        } for param in range(5) for squeeze_f in squeeze_fs]
        for trans in list(t.Transformers)
    }


def remove_raw(data_gen, rkeys=["output", "desired", "input"]):
    """remove the raw data from the data generator

    """
    for x in data_gen:
        removed = [x[idx].pop(key) for idx in range(len(x)) for key in rkeys]
        yield x


@u.Comp(remove_raw)
def process_data(data, data_len, rkeys=None):
    """post process all the raw data given

    """
    return ([{
        **apply_transformers(y, data_len),
        **y
    } for y in x] for x in data)


def hget_metrics(metrics):
    """returns the headers for each metrics

    """
    return [x for x in metrics[0].keys() if x not in ["param", "squeeze_f"]]


def hget_params(params):
    """returns the headers for each parameter

    """
    return [
        x for x in params.keys()
        if x not in map(lambda x: x.name, list(t.Transformers))
    ]


def bget_metrics(metrics):
    """returns all the metrics

    """
    return [metrics[0][x] for x in hget_metrics(metrics)]


def bget_tparam(metrics):
    """return all the paramater for a transformer

    """
    return [metrics[x]["param"] for x in range(len(metrics))]


def bget_tsqueeze(metrics):
    """returns all the squeezing function for a transformer"""
    return [metrics[x]["squeeze_f"] for x in range(len(squeeze_fs))]


def bget_params(params):
    """returns all the parameter of the network

    """
    return [params[x] for x in hget_params(params)]


first_t = list(t.Transformers)[0].name


def get_header(single_run):
    """return all the headers

    """
    return [
        *hget_params(single_run[0]), "post_trans", "post_param",
        "post_squeeze", "metric", "metric_val"
    ]


def get_body(single_run):
    """return a nested body which need to be flattened

    """
    return [[[[[
        *bget_params(redun), trans, t_param, t_squeeze, x, redun[trans][0][x]
    ] for x in hget_metrics(redun[trans])]
              for t_param in bget_tparam(redun[first_t])
              for t_squeeze in bget_tsqueeze(redun[first_t])]
             for trans in map(lambda x: x.name, list(t.Transformers))]
            for redun in single_run]


def my_flatten(lol):
    """flatten function which return a lol given a nested list

    """
    if not isinstance(lol[0][0], list):
        return lol
    acc = []
    for x in lol:
        acc += my_flatten(x)
    return acc


def to_csv(single_run, file):
    """write a csv file for the single run

    """
    with open(file, "w") as f:
        writer = csv.writer(f)
        writer.writerows(
            [get_header(single_run), *my_flatten(get_body(single_run))])


def peek(iterable):
    """utility function used to see if the generator has still something

    """
    try:
        first = next(iterable)
    except StopIteration:
        return None
    return first, it.chain([first], iterable)


def multiproc(func, args, workers, slicing=20):
    """spawn a pool of procees and apply the function to the arguments in
a current manner

    """
    pool = mp.Pool(processes=workers)
    while True:
        res = pool.map(func, it.islice(args, slicing))
        if peek(args) is None:
            return res


def partition_data(graw_data, n):
    """partition a generator in batch (len(batch)==n)

    """
    while (True):
        acc = []
        for _ in range(n):
            try:
                acc += [next(graw_data)]
            except StopIteration as e:
                return
        yield acc

    # raw_data = get_experiment(path)


def pre_post_proc(enumeration, cpath, data_len):
    """post process the data and dump a csv file sequentially

    """
    (n, graw_data) = enumeration
    data = process_data(graw_data, data_len)
    for i, run in enumerate(data):
        to_csv(
            run, cpath + str(hash(str(os.getpid()))) + str(hash(str(i))) +
            str(hash(str(n))))
        print(f"dumped run {n}")


# def write_output(raw_data, opath, data_len):
#     data = process_data(raw_data, data_len)
#     for i, run in enumerate(data):

#         print(f"dumped output {n}")


def partial_fun(enum):
    return pre_post_proc(enum, path_sneha, 6000)


def cpre_post_proc(raw_data, size_group=4, workers=20):
    """perform the post process data with multiple process

    """
    multiproc(partial_fun, enumerate(partition_data(raw_data, size_group)),
              workers)

if __name__ == "__main__":
    cpre_post_proc(get_experiment(path_s), workers = 8)

Functions

def apply_metrics(output, desired, raw_output)

apply all the possible metrics usign the output and the desired

Expand source code

def apply_metrics(output, desired, raw_output):
    """apply all the possible metrics usign the output and the desired

    """
    return {
        x.name: x.value(output, desired)()
        if x.name != "teacher_loss_nd" else x.value(raw_output, desired)()
        for x in list(met.Metrics)
    }

def apply_transformers(dict_, data_len)

apply each tranformer to the output and then apply each metrics to each output generated by the transformers

Expand source code

def apply_transformers(dict_, data_len):
    """apply each tranformer to the output and then apply each metrics to
each output generated by the transformers

    """
    output = dict_["output"][:data_len]
    des = dict_["desired"][:data_len]
    return {
        trans.name: [{
            "param": (val := ((param * 2) / 10) + 0.2),
            "squeeze_f":
            squeeze_f.__name__,
            **apply_metrics(trans.value(val, squeeze_f)(output), des, output)
        } for param in range(5) for squeeze_f in squeeze_fs]
        for trans in list(t.Transformers)
    }

def bget_metrics(metrics)

returns all the metrics

Expand source code

def bget_metrics(metrics):
    """returns all the metrics

    """
    return [metrics[0][x] for x in hget_metrics(metrics)]

def bget_params(params)

returns all the parameter of the network

Expand source code

def bget_params(params):
    """returns all the parameter of the network

    """
    return [params[x] for x in hget_params(params)]

def bget_tparam(metrics)

return all the paramater for a transformer

Expand source code

def bget_tparam(metrics):
    """return all the paramater for a transformer

    """
    return [metrics[x]["param"] for x in range(len(metrics))]

def bget_tsqueeze(metrics)

returns all the squeezing function for a transformer

Expand source code

def bget_tsqueeze(metrics):
    """returns all the squeezing function for a transformer"""
    return [metrics[x]["squeeze_f"] for x in range(len(squeeze_fs))]

def cpre_post_proc(raw_data, size_group=4, workers=20)

perform the post process data with multiple process

Expand source code

def cpre_post_proc(raw_data, size_group=4, workers=20):
    """perform the post process data with multiple process

    """
    multiproc(partial_fun, enumerate(partition_data(raw_data, size_group)),
              workers)

def experiment(path)

given a path to a directory take all the file in it and returns a generator

Expand source code

def experiment(path):
    """given a path to a directory take all the file in it and returns a generator

    """
    return (join(path, f) for f in listdir(path) if isfile(join(path, f)))

def get_body(single_run)

return a nested body which need to be flattened

Expand source code

def get_body(single_run):
    """return a nested body which need to be flattened

    """
    return [[[[[
        *bget_params(redun), trans, t_param, t_squeeze, x, redun[trans][0][x]
    ] for x in hget_metrics(redun[trans])]
              for t_param in bget_tparam(redun[first_t])
              for t_squeeze in bget_tsqueeze(redun[first_t])]
             for trans in map(lambda x: x.name, list(t.Transformers))]
            for redun in single_run]

def get_data(paths)

load each file given a list or a generator of path

Expand source code

def get_data (paths):
    """load each file given a list or a generator of path

    """
    for f in paths:
        yield [p.load(open(f, "rb"))]

def get_header(single_run)

return all the headers

Expand source code

def get_header(single_run):
    """return all the headers

    """
    return [
        *hget_params(single_run[0]), "post_trans", "post_param",
        "post_squeeze", "metric", "metric_val"
    ]

def hget_metrics(metrics)

returns the headers for each metrics

Expand source code

def hget_metrics(metrics):
    """returns the headers for each metrics

    """
    return [x for x in metrics[0].keys() if x not in ["param", "squeeze_f"]]

def hget_params(params)

returns the headers for each parameter

Expand source code

def hget_params(params):
    """returns the headers for each parameter

    """
    return [
        x for x in params.keys()
        if x not in map(lambda x: x.name, list(t.Transformers))
    ]

def multiproc(func, args, workers, slicing=20)

spawn a pool of procees and apply the function to the arguments in a current manner

Expand source code

def multiproc(func, args, workers, slicing=20):
    """spawn a pool of procees and apply the function to the arguments in
a current manner

    """
    pool = mp.Pool(processes=workers)
    while True:
        res = pool.map(func, it.islice(args, slicing))
        if peek(args) is None:
            return res

def my_flatten(lol)

flatten function which return a lol given a nested list

Expand source code

def my_flatten(lol):
    """flatten function which return a lol given a nested list

    """
    if not isinstance(lol[0][0], list):
        return lol
    acc = []
    for x in lol:
        acc += my_flatten(x)
    return acc

def partial_fun(enum)

Expand source code

def partial_fun(enum):
    return pre_post_proc(enum, path_sneha, 6000)

def partition_data(graw_data, n)

partition a generator in batch (len(batch)==n)

Expand source code

def partition_data(graw_data, n):
    """partition a generator in batch (len(batch)==n)

    """
    while (True):
        acc = []
        for _ in range(n):
            try:
                acc += [next(graw_data)]
            except StopIteration as e:
                return
        yield acc

def peek(iterable)

utility function used to see if the generator has still something

Expand source code

def peek(iterable):
    """utility function used to see if the generator has still something

    """
    try:
        first = next(iterable)
    except StopIteration:
        return None
    return first, it.chain([first], iterable)

def pre_post_proc(enumeration, cpath, data_len)

post process the data and dump a csv file sequentially

Expand source code

def pre_post_proc(enumeration, cpath, data_len):
    """post process the data and dump a csv file sequentially

    """
    (n, graw_data) = enumeration
    data = process_data(graw_data, data_len)
    for i, run in enumerate(data):
        to_csv(
            run, cpath + str(hash(str(os.getpid()))) + str(hash(str(i))) +
            str(hash(str(n))))
        print(f"dumped run {n}")

def remove_raw(data_gen, rkeys=['output', 'desired', 'input'])

remove the raw data from the data generator

Expand source code

def remove_raw(data_gen, rkeys=["output", "desired", "input"]):
    """remove the raw data from the data generator

    """
    for x in data_gen:
        removed = [x[idx].pop(key) for idx in range(len(x)) for key in rkeys]
        yield x

def to_csv(single_run, file)

write a csv file for the single run

Expand source code

def to_csv(single_run, file):
    """write a csv file for the single run

    """
    with open(file, "w") as f:
        writer = csv.writer(f)
        writer.writerows(
            [get_header(single_run), *my_flatten(get_body(single_run))])