Code snippets for page BiMDP

Download bimdp.py. Browse the code snippet index.

# -*- coding: utf-8 -*-
# Generated by codesnippet sphinx extension on 2020-12-16

import mdp
import numpy as np
np.random.seed(0)
import bimdp

pca_node = bimdp.nodes.PCABiNode(node_id="pca")
biflow = bimdp.BiFlow([pca_node])
biflow["pca"]
# Expected:
## PCABiNode(input_dim=None, output_dim=None, dtype=None, node_id="pca")

samples = np.random.random((100,10))
labels = np.arange(100)
biflow = bimdp.BiFlow([mdp.nodes.PCANode(), bimdp.nodes.FDABiNode()])
biflow.train([[samples],[samples]], msg_iterables=[None,[{"labels": labels}]])

biflow = bimdp.nodes.PCABiNode(output_dim=10) + bimdp.nodes.SFABiNode()
x = np.random.random((100,20))
biflow.train(x)
y, msg = biflow.execute(x)
msg
# Expected:
## {}
y, msg = biflow.execute(x, msg_iterable={"test": 1})
msg
# Expected:
## {'test': 1}

class SimpleCoroutineNode(bimdp.nodes.IdentityBiNode):
   # the arg ["b"] means that that the signature will be (x, b)
   @bimdp.binode_coroutine(["b"])
   def _execute(self, x, n_iterations):
       """Gather all the incomming b and return them finally."""
       bs = []
       for _ in range(n_iterations):
           x, b = yield x
           bs.append(b)
       raise StopIteration(x, {"all the b": bs})
n_iterations = 3
x = np.random.random((1,1))
node = SimpleCoroutineNode()
x, msg = node.execute(x, {"n_iterations": n_iterations})
for i in range(n_iterations-1):
   x, msg = node.execute(x, {"b": i})
x, msg = node.execute(x, {"b": n_iterations-1})