/usr/lib/python3/dist-packages/bimdp/parallel/parallelbihinet.py is in python3-mdp 3.5-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 | """
Parallel version of bihinet.
"""
from builtins import range
import mdp
from bimdp.hinet import CloneBiLayer
class ParallelCloneBiLayer(CloneBiLayer, mdp.parallel.ParallelExtensionNode):
"""Parallel version of CloneBiLayer.
This class also adds support for calling switch_to_instance during training,
using the join method of the internal nodes.
"""
def _set_use_copies(self, use_copies):
"""Switch internally between using a single node instance or copies.
In a normal CloneLayer a single node instance is used to represent all
the horizontally aligned nodes. But in a BiMDP where the nodes store
temporary data this may not work.
Via this method one can therefore create copies of the single node
instance.
This method can also be triggered by the use_copies msg key.
"""
if use_copies and not self.use_copies:
# switch to node copies
self.nodes = [self.node.copy() for _ in range(len(self.nodes))]
self.node = None # disable single node while copies are used
self._uses_copies = True
elif not use_copies and self.use_copies:
# switch to a single node instance
if self.is_training():
for forked_node in self.nodes[1:]:
self.nodes[0].join(forked_node)
elif self.is_bi_training():
for forked_node in self.nodes[1:]:
self.nodes[0].bi_join(forked_node)
self.node = self.nodes[0]
self.nodes = (self.node,) * len(self.nodes)
def _fork(self):
"""Fork the nodes in the layer to fork the layer."""
forked_node = ParallelCloneBiLayer(
node=self.nodes[0].fork(),
n_nodes=len(self.nodes),
use_copies=False,
node_id=self._node_id,
dtype=self.get_dtype())
if self.use_copies:
# simulate switch_to_copies
forked_node.nodes = [node.fork() for node in self.nodes]
forked_node.node = None
return forked_node
else:
return forked_node
def _join(self, forked_node):
"""Join the trained nodes from the forked layer."""
if self.use_copies:
for i_node, layer_node in enumerate(self.nodes):
layer_node.join(forked_node.nodes[i_node])
else:
self.node.join(forked_node.node)
def use_execute_fork(self):
if self.use_copies:
return any(node.use_execute_fork() for node in self.nodes)
else:
return self.node.use_execute_fork()
|