Source code for priops.node
[docs]class Node:
"""Nodes define nodes in the graph, by giving a specification of a list of
objects. A node contains a list, each element describes the category of an
operand. (For technical reason if cannot *be* a list: Overloading
``__getitem__()`` in that case doesn't work.)
This class is abstract. Derived classes should provide:
* ``.fulfils_specification(self, other_node)``: Gives ``True`` if
objects of specification *self* fulfil also the specification
*other_node*. Should accept ``None`` entries in *other_node*, where
``None`` is a wildcard for "any item"."""
[docs] def __init__(self, item=None, items=None, node=None):
"""Initialises the Node from either one *item* or many *items*. One
of *item* or *items* must be given, but not both. If both *item* and
*items* are ``None``, it is assumed that *item* is given as ``None``,
and that *items* is not specified.
Alternatively, the Node can be initialised from another ``Node``
instance *node*. This is needed by derived classes in some cases
to provide e.g. slicing operations based on this class'es slicing
operation. If *node* is given, it overrides all other parameters."""
if node is not None:
self.scalar = node.scalar
self.elements = node.elements
else:
if item is None and items is not None:
self.elements = list(items)
self.scalar = False
elif items is None:
# The use might want to specify *item* as precisely ``None``.
self.elements = [item]
self.scalar = True
else:
# This can happen only if 1) *items* is not None and 2) *item*
# is not None (at the same time).
raise ValueError('Exactly one of *item* and *items* must be '
'given, not both')
[docs] def frominter(self, data):
"""Processes given data to match the scalar/vector specification
of the node. This is defined at initialisation time.
*data* is always assumed to be a list. If *self* is scalar, the
length must be unity, and the first element is returned. Else,
the list is returned unchanged."""
if self.scalar:
if len(data) != 1:
raise ValueError('Scalar nodes only accept scalar-like data')
return data[0]
else:
return data
[docs] def tointer(self, data):
"""Processes the output of an edge according to the scalar/vector
specification of the node to match the inter-node data flow
specification (which is lists).
If *self* is scalar, the *data* is wrapped into a list, else, the
*data* is returned unchanged."""
if self.scalar:
return [data]
else:
return data
[docs] def __str__(self):
if self.scalar is True:
return '(Scalar Node ' + str(self.elements[0]) + ')'
elif len(self.elements) == 0:
return '(Empty Node)'
else:
return '(Vector Node ' + ', '.join(map(str, self.elements)) + ')'
[docs] def __add__(self, other_node):
"""Combines two nodes by combining them into a
:class:`CombinedNode`."""
return CombinedNode(constituents=[self, other_node])
[docs] def __getitem__(self, key):
"""If a slice is requested, returns a portion of this Node as a Node.
If an element is requested (technically, everything that's not a
slice), then the element of *self.elements* is returned."""
if isinstance(key, slice):
return Node(items=self.elements[key])
else:
return self.elements[key]
def __len__(self):
return len(self.elements)
[docs]class EmptyNode(Node):
"""A node without elements. Needed as initial element for summing up
nodes via ``sum()``."""
[docs] def __init__(self):
"""Initialises the emptiness of the node."""
Node.__init__(self, items=[])
[docs] def __add__(self, other_node):
"""Returns the other node *other_node*, since this node is empty and
doesn't matter."""
return other_node
[docs] def fulfils_specification(self, other_node):
"""If the other node *other_node* is empty too, returns ``True``,
else ``False``."""
if len(other_node) == 0:
# Both zero length:
return True
return False
[docs] def __getitem__(self, key):
"""Under slicing always returns itself, else (if non-slicing
getitem operation occurs), raises ``IndexError``."""
if isinstance(key, slice):
return self
else:
raise IndexError('Empty Node has no elements')
[docs]class CombinedNode(Node):
"""Combined nodes combine a number of :class:`Node` objects into one big
node. The elements of the combined node are the sum of all constituents,
seeing the constituents as ``list`` objects.
If a combined node fulfils another specification depends on the
constituents."""
[docs] def __init__(self, constituents):
Node.__init__(self, items=sum(map(list, constituents), []))
self.constituents = constituents
[docs] def fulfils_specification(self, other_node):
"""Checks if *self*, defined by its constituents, fulfils the
specification *other_node*. If the lengthes mismatch, ``False`` is
returned."""
if len(self) != len(other_node):
return False
other_left = other_node
for constituent in self.constituents:
other_fulfil = other_left[:len(constituent)]
if not constituent.fulfils_specification(other_fulfil):
return False
other_left = other_left[len(constituent):]
return True
[docs] def __getitem__(self, key):
"""If *key* is a slice, the constituents are sliced accordingly.
Else, the element is returned. Steps in slices are ignored."""
if isinstance(key, slice):
# Slicing requested.
# The number of items of *self* iterated over so far:
already_iterated = 0
# The list of sliced constituents:
new_constituents = []
for constituent in self.constituents:
# The indices into the *constituent*; negative indices are
# not to be wrapped around:
desired_start_idx = key.start - already_iterated
desired_stop_idx = key.stop - already_iterated
# Clip the start and stop indices to non-negative indices:
nonneg_start_idx = max(0, desired_start_idx)
nonneg_stop_idx = max(0, desired_stop_idx)
# Clip the start and stop indices to the end of the
# *constituent*:
start_idx = min(len(constituent), nonneg_start_idx)
stop_idx = min(len(constituent), nonneg_stop_idx)
# Check if the constituent is in at all:
# It is not in if the slice would be empty.
if stop_idx > start_idx:
# It is in. So slice it:
new_constituents.append(constituent[start_idx:stop_idx])
# Irrespective of whether the *constituent* was in or not
# we have processed it, so advance *already_iterated*:
already_iterated += len(constituent)
# Concatenate the new constituents to a new CombinedNode:
return CombinedNode(constituents=new_constituents)
else:
# No slicing requested, return element:
return Node.__getitem__(self, key)
