Module project.esn.core

Classes

class Data (data: numpy.ndarray, tempo: Tempo, init_len: int, train_len: int, test_len: int)

Wraps the data used during the training and the testing

Expand source code

class Data:
    """Wraps the data used during the training and the testing

    """
    data: np.ndarray
    tempo: Tempo
    init_len: int
    train_len: int
    test_len: int

    def desired(self):
        return self.data[self.init_len + 1:self.train_len + 1]

    def training_data(self):
        return self.data[:self.train_len]

    def test_data(self):
        return self.data[self.train_len + 1:]

    def start_input(self):
        return self.data[self.train_len]

Methods

def desired(self)

Expand source code

def desired(self):
    return self.data[self.init_len + 1:self.train_len + 1]

def start_input(self)

Expand source code

def start_input(self):
    return self.data[self.train_len]

def test_data(self)

Expand source code

def test_data(self):
    return self.data[self.train_len + 1:]

def training_data(self)

Expand source code

def training_data(self):
    return self.data[:self.train_len]

class ESN (_runner: Runner, trainer: Trainer, transformer: Transformer, init_len: int = 100)

Echo State Network class, this wraps a runner,a trainer and a transformer it performs both phase of the network

Expand source code

class ESN:
    """Echo State Network class, this wraps a runner,a trainer and a
    transformer it performs both phase of the network

    """
    _runner: r.Runner
    trainer: t.Trainer
    transformer: tr.Transformer
    init_len: int = 100

    def __lshift__(self, other):
        """ Training mode
        """
        (input, desired) = other
        ex_state = r.run_extended(self._runner, self.init_len, (input, None))
        self._runner.updator.weights.W_out = self.trainer((ex_state, desired))
        return ex_state

    def __rshift__(self, other):
        """Generation mode

        """
        return r.run_gen_mode(self._runner, self.transformer, other)

Class variables

var init_len : int

class Run (data: Data, in_out: int = 1, reservoir: int = 100, leaking_rate: float = 0.3, spectral_radius: float = 0.9, density: float = 0.5, reg: float = 1e-08, transformer: Transformers = Transformers.identity, t_param: float = 0.0, t_squeeze:  = <ufunc 'tanh'>, squeeze_o:  = <function Run.<lambda>>, noise: float = 0.0)

Exposed class, this simply patch everything, allowing to run the network easily

Expand source code

class Run:
    """Exposed class, this simply patch everything, allowing to run the
network easily

    """
    data: Data
    in_out: int = 1
    reservoir: int = 100
    leaking_rate: float = 0.3
    spectral_radius: float = 0.9
    density: float = 0.5
    reg: float = 1e-8
    transformer: tr.Transformers = tr.Transformers.identity
    t_param: float = 0.0
    t_squeeze: callable = np.tanh
    squeeze_o: callable = lambda x: x
    noise: float = 0.0

    def to_dict(self, kwargs={}):
        return {
            **{
                k: _get_val(getattr(self, k))
                for k, _ in self.__class__.__dict__["__dataclass_fields__"].items(
                ) if k not in ["data", "in_out"]
            },
            **kwargs
        }

    def __call__(self, input=None, desired=None):
        input = self.data.start_input() if input is None else input
        desired = self.data.test_data() if desired is None else desired
        output = self.esn >> input
        return self.to_dict({
            "input": input,
            "output": output,
            "desired": desired,
            "tempo": self.data.tempo
        })

    def load(self, path, idx):
        def matrixs_gen(self):
            dic = m.load_smatrix(path, idx)
            self.spectral_radius = dic["spectral_radius"]
            self.density = dic["density"]
            self.reservoir = dic["W_res"].shape[0]
            return m.Esn_matrixs(m.generate_rmatrix(self.reservoir,
                                                    self.in_out),
                                 dic["W_res"],
                                 spectral_radius=self.spectral_radius,
                                 density=self.density,
                                 scaled=True)

        setattr(self, "matrixs_gen", lambda: matrixs_gen(self))
        return self

    def matrixs_gen(self):
        return m.esn_matrixs(
            m.generate_rmatrix(self.reservoir, self.in_out),
            spectral_radius=self.spectral_radius,
            density=self.density,
        )

    def __enter__(self):
        self.init_len = self.data.init_len
        matrixs = self.matrixs_gen()

        updator = up.vanilla_updator(matrixs,
                                     np.zeros((self.reservoir, 1)),
                                     leaking_rate=self.leaking_rate,
                                     squeeze_o=self.squeeze_o)

        trainer = t.ridge_reg(param=self.reg)
        transformer = self.transformer.value(self.t_param, self.t_squeeze)

        self.esn = ESN(r.runner(updator, self.data.test_len), trainer,
                       transformer, self.init_len)
        self.activations = self.esn << (self.data.training_data(),
                                        self.data.desired())
        return self

    def __exit__(self, err_t, err_v, traceback):
        pass

Class variables

var density : float

var in_out : int

var leaking_rate : float

var noise : float

var reg : float

var reservoir : int

var spectral_radius : float

var t_param : float

var t_squeeze : 

var transformer : Transformers

Methods

def load(self, path, idx)

Expand source code

def load(self, path, idx):
    def matrixs_gen(self):
        dic = m.load_smatrix(path, idx)
        self.spectral_radius = dic["spectral_radius"]
        self.density = dic["density"]
        self.reservoir = dic["W_res"].shape[0]
        return m.Esn_matrixs(m.generate_rmatrix(self.reservoir,
                                                self.in_out),
                             dic["W_res"],
                             spectral_radius=self.spectral_radius,
                             density=self.density,
                             scaled=True)

    setattr(self, "matrixs_gen", lambda: matrixs_gen(self))
    return self

def matrixs_gen(self)

Expand source code

def matrixs_gen(self):
    return m.esn_matrixs(
        m.generate_rmatrix(self.reservoir, self.in_out),
        spectral_radius=self.spectral_radius,
        density=self.density,
    )

def squeeze_o(x) ‑> 

Expand source code

squeeze_o: callable = lambda x: x

def to_dict(self, kwargs={})

Expand source code

def to_dict(self, kwargs={}):
    return {
        **{
            k: _get_val(getattr(self, k))
            for k, _ in self.__class__.__dict__["__dataclass_fields__"].items(
            ) if k not in ["data", "in_out"]
        },
        **kwargs
    }