Franck Pommereau / snakes

Go to a project
Toggle navigation Toggle navigation pinning
Franck Pommereau
baa7af9f 1 parent 1157b819

doc update + synchro fixes

Expand all
Inline Side-by-side
Showing 9 changed files with 330 additions and 61 deletions
......@@ -895,6 +895,25 @@ class Substitution (object) :
for var in other :
res._dict[var] = self(other(var))
return res
def restrict (self, domain) :
"""Restrict the substitution to `domain`, ie remove all
elements that are not in `domain`. Note that `domain` may
include names that are not in the substitution, they are
simply ignored.
>>> s = Substitution(a=1, b=2, c=3, d=4).restrict(['a', 'b', 'z'])
>>> list(sorted(s.domain()))
['a', 'b']
@param domain: the new domain as a set/list/... of names
@type domain: `iterable`
@return: the restricted substitution
@rtype: `Substitution`
"""
result = self.copy()
for name in result.domain() - set(domain) :
result._dict.pop(name, None)
return result
class Symbol (object) :
"""A symbol that may be used as a constant
......
This diff is collapsed. Click to expand it.
"""Draw Petri nets using PyGraphViz
"""Adds methods to draw `PetriNet` and `StateGraph` instances using
GraphViz.
* adds a method `draw` to `PetriNet` and `StateGraph` that creates
a drawing of the object in a file.
For example, let's first define a Petri net:
>>> import snakes.plugins
>>> snakes.plugins.load('gv', 'snakes.nets', 'nets')
......@@ -16,18 +16,35 @@
>>> n.add_output('p11', 't10', Expression('x+1'))
>>> n.add_input('p11', 't01', Variable('y'))
>>> n.add_output('p00', 't01', Expression('y-1'))
Thanks to plugin `gv`, we can draw it using the various engines of
GraphViz; we can also draw the state graph:
>>> for engine in ('neato', 'dot', 'circo', 'twopi', 'fdp') :
... n.draw(',test-gv-%s.png' % engine, engine=engine)
>>> s = StateGraph(n)
>>> s.build()
>>> s.draw(',test-gv-graph.png')
>>> for node in sorted(n.node(), key=str) :
The plugin also allows to layout the nodes without drawing the net
(this is only available for `PetriNet`, not for `StateGraph`). We
first move every node to position `(-100, -100)`; then, we layout the
net; finally, we check that every node has indeed been moved away from
where we had put it:
>>> for node in n.node() :
... node.pos.moveto(-100, -100)
>>> all(node.pos.x == node.pos.y == -100 for node in n.node())
True
>>> n.layout()
>>> any(node.pos == (-100, -100) for node in sorted(n.node(), key=str))
>>> any(node.pos.x == node.pos.y == -100 for node in n.node())
False
@todo: revise documentation
@note: setting nodes position has no influence on how a net is drawn:
GraphViz will redo the layout in any case. Method `layout` is here
just in the case you would need a layout of your nets.
@note: this plugin depens on plugins `pos` and `clusters` (that are
automatically loaded)
"""
import os, os.path, subprocess, collections
......@@ -35,6 +52,7 @@ import snakes.plugins
from snakes.plugins.clusters import Cluster
from snakes.compat import *
# apidoc skip
class Graph (Cluster) :
def __init__ (self, attr) :
Cluster.__init__(self)
......@@ -150,37 +168,56 @@ class Graph (Cluster) :
"snakes.plugins.pos"])
def extend (module) :
class PetriNet (module.PetriNet) :
"An extension with a method `draw`"
def draw (self, filename=None, engine="dot", debug=False,
"An extension with methods `draw` and `layout`"
def draw (self, filename, engine="dot", debug=False,
graph_attr=None, cluster_attr=None,
place_attr=None, trans_attr=None, arc_attr=None) :
"""
@param filename: the name of the image file to create or
`None` if only the computed graph is needed
@type filename: `None` or `str`
@param engine: the layout engine to use: 'dot' (default),
'neato', 'circo', 'twopi' or 'fdp'
"""Draw the Petri net to a picture file. How the net is
rendered can be controlled using the arguments `..._attr`.
For instance, to draw place in red with names in
uppercase, and hide True guards, we can proceed as
follows:
>>> import snakes.plugins
>>> snakes.plugins.load('gv', 'snakes.nets', 'nets')
<module ...>
>>> from nets import *
>>> n = PetriNet('N')
>>> n.add_place(Place('p'))
>>> n.add_transition(Transition('t'))
>>> n.add_input('p', 't', Value(dot))
>>> def draw_place (place, attr) :
... attr['label'] = place.name.upper()
... attr['color'] = '#FF0000'
>>> def draw_transition (trans, attr) :
... if str(trans.guard) == 'True' :
... attr['label'] = trans.name
... else :
... attr['label'] = '%s\\n%s' % (trans.name, trans.guard)
>>> n.draw(',net-with-colors.png',
... place_attr=draw_place, trans_attr=draw_transition)
@param filename: the name of the image file to create
@type filename: `str`
@param engine: the layout engine to use: `'dot'` (default),
`'neato'`, `'circo'`, `'twopi'` or `'fdp'`
@type engine: `str`
@param place_attr: a function to format places, it will be
called with the place and its attributes dict as
parameters
@type place_attr: `function(Place,dict)->None`
@type place_attr: `callable`
@param trans_attr: a function to format transitions, it
will be called with the transition and its attributes
dict as parameters
@type trans_attr: `function(Transition,dict)->None`
@type trans_attr: `callable`
@param arc_attr: a function to format arcs, it will be
called with the label and its attributes dict as
parameters
@type arc_attr: `function(ArcAnnotation,dict)->None`
@type arc_attr: `callable`
@param cluster_attr: a function to format clusters of
nodes, it will be called with the cluster and its
attributes dict as parameters
@type cluster_attr:
`function(snakes.plugins.clusters.Cluster,dict)->None`
@return: `None` if `filename` is not `None`, the computed
graph otherwise
@rtype: `None` or `pygraphviz.AGraph`
@type cluster_attr: `callable`
"""
nodemap = dict((node.name, "node_%s" % num)
for num, node in enumerate(self.node()))
......@@ -236,17 +273,51 @@ def extend (module) :
def layout (self, xscale=1.0, yscale=1.0, engine="dot",
debug=False, graph_attr=None, cluster_attr=None,
place_attr=None, trans_attr=None, arc_attr=None) :
"""Layout the nodes of the Petri net by calling GraphViz
and reading back the picture it creates. The effect is to
change attributes `pos` (see plugin `pos`) for every node
according to the positions calculated by GraphViz.
@param xscale: how much the image is scaled in the
horizontal axis after GraphViz has done the layout
@type xscale: `float`
@param yscale: how much the image is scaled in the
vertical axis after GraphViz has done the layout
@type yscale: `float`
@param filename: the name of the image file to create
@type filename: `str`
@param engine: the layout engine to use: `'dot'` (default),
`'neato'`, `'circo'`, `'twopi'` or `'fdp'`
@type engine: `str`
@param place_attr: a function to format places, it will be
called with the place and its attributes dict as
parameters
@type place_attr: `callable`
@param trans_attr: a function to format transitions, it
will be called with the transition and its attributes
dict as parameters
@type trans_attr: `callable`
@param arc_attr: a function to format arcs, it will be
called with the label and its attributes dict as
parameters
@type arc_attr: `callable`
@param cluster_attr: a function to format clusters of
nodes, it will be called with the cluster and its
attributes dict as parameters
@type cluster_attr: `callable`
"""
g = self.draw(None, engine, debug, graph_attr, cluster_attr,
place_attr, trans_attr, arc_attr)
node = dict((v, k) for k, v in g.nodemap.items())
for n, x, y in g.layout(engine, debug) :
self.node(node[n]).pos.moveto(x*xscale, y*yscale)
class StateGraph (module.StateGraph) :
"An extension with a method `draw`"
def draw (self, filename=None, engine="dot", debug=False,
def draw (self, filename, engine="dot", debug=False,
node_attr=None, edge_attr=None, graph_attr=None) :
"""@param filename: the name of the image file to create or
"""Draw the state graph to a picture file.
@param filename: the name of the image file to create or
`None` if only the computed graph is needed
@type filename: `None` or `str`
@param engine: the layout engine to use: 'dot' (default),
......@@ -255,19 +326,16 @@ def extend (module) :
@param node_attr: a function to format nodes, it will be
called with the state number, the `StateGraph` object
and attributes dict as parameters
@type node_attr: `function(int,StateGraph,dict)->None`
@type node_attr: `callable`
@param edge_attr: a function to format edges, it will be
called with the transition, its mode and attributes
dict as parameters
@type trans_attr:
`function(Transition,Substitution,dict)->None`
`callable`
@param graph_attr: a function to format grapg, it will be
called with the state graphe and attributes dict as
parameters
@type graph_attr: `function(StateGraph,dict)->None`
@return: `None` if `filename` is not `None`, the computed
graph otherwise
@rtype: `None` or `pygraphviz.AGraph`
@type graph_attr: `callable`
"""
attr = dict(style="invis",
splines="true")
......
"""A plugin to add labels to nodes and nets.
"""A plugin to add labels to nodes and nets. Labels are names (valid
Python identifiers) associated to arbitrary objects.
@todo: revise (actually make) documentation
>>> import snakes.plugins
>>> snakes.plugins.load('labels', 'snakes.nets', 'nets')
<module ...>
>>> from nets import *
>>> t = Transition('t')
>>> t.label(foo='bar', spam=42)
>>> t.label('foo')
'bar'
>>> t.label('spam')
42
Note that when nodes in a Petri net are merged, their labels are
merged too, but in an arbitrary order. So, for example:
>>> n = PetriNet('N')
>>> n.add_place(Place('p1'))
>>> n.place('p1').label(foo='bar', spam='ham')
>>> n.add_place(Place('p2'))
>>> n.place('p2').label(hello='world', spam='egg')
>>> n.merge_places('p', ['p1', 'p2'])
>>> n.place('p').label('hello')
'world'
>>> n.place('p').label('foo')
'bar'
>>> n.place('p').label('spam') in ['ham', 'egg']
True
In the latter statement, we cannot know whether the label will be one
or the other value because merging has been done in an arbitrary
order.
"""
from snakes.plugins import plugin, new_instance
......@@ -10,6 +40,22 @@ from snakes.pnml import Tree
def extend (module) :
class Transition (module.Transition) :
def label (self, *get, **set) :
"""Get and set labels for the transition. The labels given
in `get` will be returned as a `tuple` and the labels
assigned in `set` will be changed accordingly. If a label
is given both in `get`and `set`, the returned value is
that it had at the beginning of the call, ie, before it is
set by the call.
@param get: labels which values have to be returned
@type get: `str`
@param set: labels which values have to be changed
@type set: `object`
@return: the tuples of values corresponding to `get`
@rtype: `tuple`
@raise KeyError: when a label given in `get` has not been
assigned previouly
"""
if not hasattr(self, "_labels") :
self._labels = {}
result = tuple(self._labels[g] for g in get)
......@@ -21,10 +67,22 @@ def extend (module) :
elif len(set) == 0 :
return self._labels.copy()
def has_label (self, name, *names) :
"""Check is a label has been assigned to the transition.
@param name: the label to check
@type name: `str`
@param names: additional labels to check, if used, the
return value is a `tuple` of `bool` instead of a
single `bool`
@return: a Boolean indicating of the checked labels are
present or not in the transitions
@rtype: `bool`
"""
if len(names) == 0 :
return name in self._labels
else :
return tuple(n in self._labels for n in (name,) + names)
# apidoc stop
def copy (self, name=None, **options) :
if not hasattr(self, "_labels") :
self._labels = {}
......@@ -79,6 +137,7 @@ def extend (module) :
return t
class Place (module.Place) :
def label (self, *get, **set) :
"See documentation for `Transition.label` above"
if not hasattr(self, "_labels") :
self._labels = {}
result = tuple(self._labels[g] for g in get)
......@@ -90,10 +149,12 @@ def extend (module) :
elif len(set) == 0 :
return self._labels.copy()
def has_label (self, name, *names) :
"See documentation for `Transition.has_label` above"
if len(names) == 0 :
return name in self._labels
else :
return tuple(n in self._labels for n in (name,) + names)
# apidoc stop
def copy (self, name=None, **options) :
if not hasattr(self, "_labels") :
self._labels = {}
......@@ -152,6 +213,7 @@ def extend (module) :
return p
class PetriNet (module.PetriNet) :
def label (self, *get, **set) :
"See documentation for `Transition.label` above"
if not hasattr(self, "_labels") :
self._labels = {}
result = tuple(self._labels[g] for g in get)
......@@ -163,10 +225,12 @@ def extend (module) :
elif len(set) == 0 :
return self._labels.copy()
def has_label (self, name, *names) :
"See documentation for `Transition.has_label` above"
if len(names) == 0 :
return name in self._labels
else :
return tuple(n in self._labels for n in (name,) + names)
# apidoc stop
def copy (self, name=None, **options) :
if not hasattr(self, "_labels") :
self._labels = {}
......
"""A plugin to compose nets.
# -*- encoding: latin-1
"""A plugin to compose nets _à la_ Petri Box Calculus.
The compositions are based on place status and automatically merge
some nodes (buffers and variables, tick transitions).
@note: this plugin depends on plugins `clusters` and `status` that are
automatically loaded
"""
"""
## Control-flow and buffers places ##
When this module is used, places are equipped with statuses (see also
plugin `status` that provides this service). We distinguish in
particular:
* _entry_ places marked at the initial state of the net
* _exit_ places marked at a final state of the net
* _internal_ places marked during the execution
* all together, these places form the _control-flow places_ and can
be marked only by black tokens (ie, they are typed `tBlackToken`
and thus can hold only `dot` values)
* _buffer_ places that may hold data of any type, each buffer place
is given a name that is the name of the buffer modelled by the
place
Plugin `ops` exports these statuses as Python objects:[^1] `entry`,
`internal` and `exit` are instances of class `Status` while `buffer`
is a function that returns an instance of `Buffer` (a subclass of
`Status`) when called with a name as its parameter.
[^1]: All these objects are actually defined in plugin `status`
Let's define a net with one entry place, one exit place and two buffer
places. Note how we use keyword argument `status` to the place
constructor to define the status of the created place:
>>> import snakes.plugins
>>> snakes.plugins.load('ops', 'snakes.nets', 'nets')
<module ...>
>>> from nets import *
>>> from snakes.plugins.status import entry, internal, exit, buffer
>>> basic = PetriNet('basic')
>>> basic.add_place(Place('e', status=entry))
>>> basic.add_place(Place('x', status=exit))
>>> basic.add_transition(Transition('t'))
>>> basic.add_input('e', 't', Value(1))
>>> basic.add_output('x', 't', Value(2))
>>> basic.add_place(Place('b', [1], status=buffer('buf')))
>>> basic.add_input('e', 't', Value(dot))
>>> basic.add_output('x', 't', Value(dot))
>>> basic.add_place(Place('b1', [1], status=buffer('egg')))
>>> basic.add_place(Place('b2', [2], status=buffer('spam')))
>>> basic.add_input('b1', 't', Variable('x'))
>>> basic.add_output('b2', 't', Expression('x+1'))
The simplest operation is to change buffer names, let's do it on a
copy of our net. This operation is called `hide` because it is
basically used to hide a buffer:
>>> n = basic.copy()
>>> n.hide(entry)
>>> n.node('e').status
>>> n.node('b1').status
Buffer('buffer','egg')
>>> n.hide(buffer('egg'))
>>> n.node('b1').status
Status(None)
>>> n.hide(buffer('buf'), buffer(None))
>>> n.node('b').status
Buffer('buffer')
>>> n = basic / 'buf'
>>> n.node('[b/buf]').status
As we can see, place `b1` now has a dummy status. But `hide` can
accept a second argument and allows to rename a buffer:
>>> n.node('b2').status
Buffer('buffer','spam')
>>> n.hide(buffer('spam'), buffer('ham'))
>>> n.node('b2').status
Buffer('buffer','ham')
A slighlty different way for hiding buffers is to use operator `/`,
which actually constructs a new net, changing the names of every node
in the original net to `'[.../egg]'`:
>>> n = basic / 'egg'
>>> n.node('[b1/egg]').status
Buffer('buffer')
As we can see, a buffer hidden using `/` still has a buffer status but
with no name associated. Such an anonymous buffer is treated in a
special way as we'll see later on.
The systematic renaming of nodes is something we should get used to
before to continue. When one or two nets are constructed through an
operation, the nodes of the operand nets are combined in one way or
another. For an arbitrary operation `A % B`, the resulting net will be
called `'[A.name%B.name]'` (if `A` and `B` are nets). Whenever a node
`a` from `A` is combined with a node `b` from `B`, the resulting node
will be called `'[a%b]'`. If only one node is copied in the resulting
net, it will be called `'[a%]'` or `'[%b]'` depending on wher it comes
from. This systematic renaming allows to ensure that even if `A` and
`B` have nodes with the same names, there will be no name clash in the
result, while, at the same time allowing users to predict the new name
of a node.
## Control-flow compositions ##
Using control-flow places, it becomes possible to build nets by
composing smaller nets through control flow operations. Let's start
with the _sequential composition_ `A & B`, basically, its combines the
exit places of net `A` with the entry places of net `B`, ensuring thus
that the resulting net behaves like if we execute `A` followed by `B`.
In the example below, we use method `status` of a net to get the
places with a given status:
>>> n = basic & basic
>>> n.status(internal)
('[x&e]',)
>>> n.place('[x&e]').pre
{'[t&]': Value(2)}
{'[t&]': Value(dot)}
>>> n.place('[x&e]').post
{'[&t]': Value(1)}
>>> n.status(buffer('buf'))
('[b&b]',)
{'[&t]': Value(dot)}
>>> n.status(buffer('egg'))
('[b1&b1]',)
>>> n.status(buffer('spam'))
('[b2&b2]',)
We can see that `n` now has one internal place that is the combination
of the exit place of the left copy of `basic` with the entry place of
the right copy of `basic`. (Hence its name: `'[x&e]'`.) We can see
also how it is connected to the left/right copy of `t` as its
input/output. Last, we can see that buffer places from the two copies
of `basic` have been merged when they have had the same name. This
merging won't occur for anonymous buffers. For example, hiding
`'spam'` in the right net of the sequential composition will result in
two buffer places, one still named `'spam'` that is the copy of `'b2'`
from the left operand of `&`, another that is anonymous and is the
copy of `'[b2/spam]'` (ie, `'b2'`after its status was hidden) from the
right operand of `&`.
>>> n = basic & (basic / 'spam')
>>> n.status(buffer('spam'))
('[b2&]',)
>>> n.status(buffer(None))
('[&[b2/spam]]',)
The next operation is the _choice `A + B` that behave either as `A` or
as `B`. This is obtained by comining the entry places of both nets one
the one hand, and by combining their exit places on the other hand.
>>> n = basic + basic
>>> n.status(entry)
('[e+e]',)
>>> list(sorted(n.place('[e+e]').post.items()))
[('[+t]', Value(1)), ('[t+]', Value(1))]
[('[+t]', Value(dot)), ('[t+]', Value(dot))]
>>> n.status(exit)
('[x+x]',)
>>> list(sorted(n.place('[x+x]').pre.items()))
[('[+t]', Value(2)), ('[t+]', Value(2))]
[('[+t]', Value(dot)), ('[t+]', Value(dot))]
Another operation is the _iteration_ `A * B` that behaves by executing
`A` repeatedly (including no repetition) followed by one execution of
`B`. This is obtained by combining the entry and exit places of `A`
with the entry place of `B`.
>>> n = basic * basic
>>> n.status(entry)
('[e,x*e]',)
>>> n.place('[e,x*e]').post
{'[t*]': Value(1), '[*t]': Value(1)}
{'[t*]': Value(dot), '[*t]': Value(dot)}
>>> n.place('[e,x*e]').pre
{'[t*]': Value(2)}
{'[t*]': Value(dot)}
Finally, there is the _parallel composition_ `A | B` that just
executes both nets in parallel. But because of the merging of buffer
places, they are able to communicate.
>>> n = basic | basic
>>> n.status(buffer('egg'))
('[b1|b1]',)
>>> n.status(buffer('spam'))
('[b2|b2]',)
>>> pass
>>> n1 = basic.copy()
>>> n1.declare('global x; x=1')
>>> n2 = basic.copy()
......@@ -65,8 +181,6 @@ Buffer('buffer')
(1, 2)
>>> n._declare
['global x; x=1']
@todo: revise documentation
"""
import snakes.plugins
......@@ -113,13 +227,14 @@ def _glue (op, one, two) :
depends=["snakes.plugins.clusters",
"snakes.plugins.status"])
def extend (module) :
"Build the extended module"
"""Essentially, class `PetriNet` is extended to support the binary
operations discussed above."""
class PetriNet (module.PetriNet) :
def __or__ (self, other) :
"Parallel"
"Parallel composition"
return _glue("|", self, other)
def __and__ (self, other) :
"Sequence"
"Sequential composition"
result = _glue("&", self, other)
remove = set()
for x, e in cross((self.status(exit), other.status(entry))) :
......@@ -161,11 +276,13 @@ def extend (module) :
result.remove_place(p)
return result
def hide (self, old, new=None) :
"Status hiding and renaming"
if new is None :
new = Status(None)
for node in self.status(old) :
self.set_status(node, new)
def __div__ (self, name) :
"Buffer hiding"
result = self.copy()
for node in result.node() :
result.rename_node(node.name, "[%s/%s]" % (node, name))
......@@ -173,6 +290,7 @@ def extend (module) :
if status._value == name :
result.hide(status, status.__class__(status._name, None))
return result
# apidoc skip
def __truediv__ (self, other) :
return self.__div__(other)
return PetriNet
......
This diff is collapsed. Click to expand it.
This diff is collapsed. Click to expand it.
This diff is collapsed. Click to expand it.
......@@ -382,7 +382,7 @@ class DocExtract (object) :
% (info["type"].get(arg, "object").strip("`"),
arg))
for kw, text in sorted(info["keyword"].items()) :
self.writelist("`%s`: %s" % (kw, text))
self.writelist("keyword `%s`: %s" % (kw, text))
if any(k in info for k in ("return", "rtype")) :
if "return" in info :
self.writelist("`return %s`: %s"
......