/usr/share/pyshared/bimdp/hinet/biswitchboard.py is in python-mdp 3.3-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | import sys
import mdp
import mdp.hinet as hinet
n = mdp.numx
from bimdp import BiNode
class BiSwitchboard(BiNode, hinet.Switchboard):
"""BiMDP version of the normal Switchboard.
It adds support for stop_message and also tries to apply the switchboard
mapping to arrays in the message. The mapping is only applied if the
array is at least two dimensional and the second dimension matches the
switchboard dimension.
"""
def __init__(self, **kwargs):
"""Initialize BiSwitchboard.
args and kwargs are forwarded via super to the next __init__ method
in the MRO.
"""
super(BiSwitchboard, self).__init__(**kwargs)
if self.inverse_connections is not None:
self.down_connections = self.inverse_connections
else:
# a stable (order preserving where possible) sort is
# necessary here, therefore use mergesort
# otherwise channels (e.g. for a Rectangular2d...) are mixed up
self.down_connections = n.argsort(self.connections,
kind="mergesort")
def _execute(self, x, msg=None):
"""Return the routed input data."""
if x is not None:
y = super(BiSwitchboard, self)._execute(x)
else:
y = None
msg = self._execute_msg(msg)
if not msg:
return y
else:
return y, msg
def _inverse(self, x, msg=None):
"""Return the routed input data."""
if x is not None:
y = super(BiSwitchboard, self)._inverse(x)
else:
y = None
msg = self._inverse_msg(msg)
if not msg:
return y
else:
return y, msg
def is_bi_training(self):
return False
## Helper methods ##
def _inverse_msg(self, msg):
"""Inverse routing for msg."""
if not msg:
return None
out_msg = {}
for (key, value) in msg.items():
if (type(value) is n.ndarray and
len(value.shape) >= 2 and value.shape[1] == self.output_dim):
out_msg[key] = super(BiSwitchboard, self)._inverse(value)
else:
out_msg[key] = value
return out_msg
def _execute_msg(self, msg):
"""Feed-forward routing for msg."""
if not msg:
return None
out_msg = {}
for (key, value) in msg.items():
if (type(value) is n.ndarray and
len(value.shape) >= 2 and value.shape[1] == self.input_dim):
out_msg[key] = super(BiSwitchboard, self)._execute(value)
else:
out_msg[key] = value
return out_msg
## create BiSwitchboard versions of the standard MDP switchboards ##
# corresponding methods for the switchboard_factory extension are
# created as well
@classmethod
def _binode_create_switchboard(cls, free_params, prev_switchboard,
prev_output_dim, node_id):
"""Modified version of create_switchboard to support node_id.
This method can be used as a substitute when using the switchboard_factory
extension.
"""
compatible = False
for base_class in cls.compatible_pre_switchboards:
if isinstance(prev_switchboard, base_class):
compatible = True
if not compatible:
err = ("The prev_switchboard class '%s'" %
prev_switchboard.__class__.__name__ +
" is not compatible with this switchboard class" +
" '%s'." % cls.__name__)
raise mdp.hinet.SwitchboardException(err)
for key, value in free_params.items():
if key.endswith('_xy') and isinstance(value, int):
free_params[key] = (value, value)
kwargs = cls._get_switchboard_kwargs(free_params, prev_switchboard,
prev_output_dim)
return cls(node_id=node_id, **kwargs)
# TODO: Use same technique as for binodes?
# But have to take care of the switchboard_factory extension.
# use a function to avoid poluting the namespace
def _create_bi_switchboards():
switchboard_classes = [
mdp.hinet.ChannelSwitchboard,
mdp.hinet.Rectangular2dSwitchboard,
mdp.hinet.DoubleRect2dSwitchboard,
mdp.hinet.DoubleRhomb2dSwitchboard,
]
current_module = sys.modules[__name__]
for switchboard_class in switchboard_classes:
node_name = switchboard_class.__name__
binode_name = node_name[:-len("Switchboard")] + "BiSwitchboard"
docstring = ("Automatically created BiSwitchboard version of %s." %
node_name)
docstring = "Automatically created BiNode version of %s." % node_name
exec ('class %s(BiSwitchboard, mdp.hinet.%s): "%s"' %
(binode_name, node_name, docstring)) in current_module.__dict__
# create appropriate FactoryExtension nodes
mdp.extension_method("switchboard_factory",
current_module.__dict__[binode_name],
"create_switchboard")(_binode_create_switchboard)
_create_bi_switchboards()
|