Source code for priops.edge
from priops.node import EmptyNode
[docs]class Edge:
"""Edges describe connections between two :class:`~priops.Node` objects.
An edge has a direction; there is an input node and an output node.
An edge can be called, in that case the edge will transform an object
list of the input node specification into an object list of the output
node specification.
This class is abstract. Derived classes should provide:
* ``.process(self, input)``: Processes the non-internode input data to
non-internode output data. Non-internode data is different from
internode data format by the fact that inter-node data is always a
list, while intra-edge data might be scalar.
1. If the input node is scalar, *input* will be scalar (the first
element of the inter-node data stream). If the input node is not
scalar, *input* will be the direct inter-node data stream.
2. If the output node is scalar, ``.process()`` should return a
scalar, which will then be converted by the output node to
inter-node list format. If the output node is not scalar, the
output of ``.process()`` should already be a list."""
[docs] def __init__(self, input_node, output_node, weight=None, name=None):
"""*input_node* is the specification of the input, *output_node* is the
specification of the output of the ``Edge``. *weight* is the effort
to follow this edge (default 1). *name* is used for pretty
printing."""
if weight is None:
weight = 1
self.input_node = input_node
self.output_node = output_node
self.weight = weight
self.name = name
[docs] def __str__(self):
if self.name is None:
return '(' + str(self.input_node) + ' -> ' + \
str(self.output_node) + ')'
else:
return '(' + self.name + ' ' + str(self.input_node) + ' -> ' + \
str(self.output_node) + ')'
[docs] def __add__(self, other_edge):
"""Combines this edge *self* with another edge *other_edge* via
creating a :class:`CombinedEdge` instance."""
return CombinedEdge(constituents=[self, other_edge])
[docs] def __call__(self, inputs):
"""Processes the inter-edge data stream *inputs* into inter-edge
data stream *outputs*. Inter-edge data are lists.
Calls *self.process()* with appropriate arguments."""
intra_edge_input = self.input_node.frominter(inputs)
intra_edge_processed = self.process(input=intra_edge_input)
inter_edge_output = self.output_node.tointer(intra_edge_processed)
return inter_edge_output
[docs]class IdentityEdge(Edge):
"""An edge with identical input and output nodes, returning its
arguments on call. Has zero weight."""
[docs] def __init__(self, node):
"""*node* is the node to process by identity."""
Edge.__init__(self, input_node=node, output_node=node, weight=0,
name='Identity')
[docs]class CombinedEdge(Edge):
"""Combined edges combine a number of :class:`Edge` objects into one big
edge. The input node of the combined edge is the sum of all input nodes
of the constituets. The output node of a combined edge is similarly the
sum of all output nodes of the constituets.
When being called, the combined edge dispatches the inputs to the
constituents according to the length of their input nodes. Then the
constituents will be called with the dispatched inputs."""
[docs] def __init__(self, constituents):
"""Combines *constituents* into a combined egde."""
self.constituents = constituents
input_node = sum([constituent.input_node for constituent in \
constituents], EmptyNode())
output_node = sum([constituent.output_node for constituent in \
constituents], EmptyNode())
weight = sum([constituent.weight for constituent in constituents])
Edge.__init__(self, input_node=input_node, output_node=output_node,
weight=weight, name='Combined Edge')
[docs] def process(self, input):
"""Dispatches the input object list *input* to the constituents."""
result = []
input_left = input
for constituent in self.constituents:
len_constituent = len(constituent.input_node)
input_constituent = input_left[:len_constituent]
result.extend(constituent(input_constituent))
input_left = input_left[len_constituent:]
return result
