Franck Pommereau / zinc

Go to a project
Toggle navigation Toggle navigation pinning
Franck Pommereau
d5c75f46 1 parent f3b1b000

added plugin system

Inline Side-by-side
Showing 11 changed files with 311 additions and 46 deletions
......@@ -16,3 +16,6 @@ class ParseError (ZINCError) :
if loc :
msg = "[%s] %s" % (loc, msg)
ZINCError.__init__(self, msg)
class CompileError (ParseError) :
pass
......
......@@ -40,10 +40,20 @@ _errre = re.compile(r"^.*?error at line ([0-9]+), col ([0-9]+):[ \t]*"
"((.|\n)*)$", re.I|re.A)
class BaseParser (object) :
@classmethod
def make_parser (cls) :
def parse (source) :
if isinstance(source, str) :
return cls().parse(source, "<string>")
else :
return cls().parse(source.read(), getattr(source, "name", "<string>"))
return parse
def init (self, parser) :
pass
def parse (self, source, path) :
parser = self.__parser__(indedent(source))
self.source = source
self.path = path
self.parser = parser = self.__parser__(indedent(source))
self.init(parser)
try :
do_parse = getattr(parser, parser.__default__, "INPUT")
......
......@@ -10,8 +10,4 @@ class Parser (BaseParser) :
["negate"],
["task"]]
def parse (source) :
if isinstance(source, str) :
return Parser().parse(source, "<string>")
else :
return Parser().parse(source.read(), getattr(source, "name", "<string>"))
parse = Parser.make_parser()
......
import pathlib, logging, inspect
from .. import CompileError, nets
from . import metaparse, BaseParser
from .meta import Compiler as BaseCompiler, node as decl
class Parser (BaseParser) :
class __parser__ (metaparse.MetaParser) :
__compound__ = [["if", "*elif", "?else"],
["for", "?else"],
["while", "?else"],
["try", "except"],
["def"],
["task"]]
parse = Parser.make_parser()
class Context (dict) :
def __call__ (self, **args) :
d = self.copy()
d.update(args)
return d
def __getattr__ (self, key) :
return self[key]
def __setattr__ (self, key, val) :
self[key] = val
class Compiler (BaseCompiler) :
def __init__ (self, module=nets) :
self.n = module
def __call__ (self, source) :
self.p = Parser()
tree = self.parser.parse(source)
self._tasks = {}
return self.visit(tree, Context(_path=pathlib.Path("/")))
def _get_pos (self, node) :
if "_pos" in node :
return (self.p.parser._p_get_line(node._pos),
self.p.parser._p_get_col(node._pos) - 1)
else :
return None, None
def warn (self, node, message) :
lno, cno = self._get_pos(node)
if lno is not None :
message = "[%s] %s" % (":".join([self.p.path, str(lno)]), message)
logging.warn(message)
def _check (self, node, cond, error) :
lno, cno = self._get_pos(node)
raise CompileError(error, lno=lno, cno=cno, path=self.p.path)
def build_seq (self, seq, ctx) :
net = self.visit(seq[0], ctx)
for other in seq[1:] :
n = self.visit(other, ctx)
if n is not None :
net &= n
return net
def visit_model (self, node, ctx) :
return self.visit_seq(node.body, ctx)
def visit_task (self, node, ctx) :
pass
def visit_def (self, node, ctx) :
self._check(node, node.deco is None, "decorators are not supported")
self._check(node, node.largs, "missing parameters NAME")
self._check(node, node.largs[0].type == "name",
"invalid function name %r (%s)" % (node.largs[0].value,
node.largs[0].type))
for a in node.largs[1:] :
self._check(node, a.kind == "name", "invalid parameter %s (%s)"
% (a.value, a.kind))
for a, v in node.kargs.items() :
self._check(node, v.kind in ("int", "code", "str"),
"invalid defaut value for parameter %s: %r (%s)"
% (a, v.value, v.kind))
name = node.largs[0].value
if name in ctx :
self.warn(node, "previous declaration of %r is masked" % name)
path = ctx._path / name
flag = inspect.Parameter.POSITIONAL_OR_KEYWORD
sig = inspect.Signature([inspect.Parameter(a.value, flag)
for a in node.largs[1:]]
+ [inspect.Parameter(a, flag, default=v.value)
for a, v in node.kargs.items()])
ctx[name] = decl(kind="function", args=sig)
body = self.build_seq(node.body, ctx(_path=path,
**{a : decl(kind="variable")
for a in sig.parameters}))
# TODO: add initial transition that gets all the arguments and store them into
# buffer places + terminal transition that flush these places and return
# the value
call = ret = None
net = (call & body & ret) / tuple(sig.parameters)
self._tasks[path] = net.task(name)
def visit_assign (self, node, ctx) :
tgt = node.target
val = node.value
self._check(node, tgt not in ctx or ctx[tgt].kind == "variable",
"cannot assign to previously declared %s %r"
% (ctx[tgt].kind, tgt))
if val.tag == "call" :
net = self.visit(val)
if tgt not in net :
net.add_place(self.n.Place(tgt, status=self.n.buffer(tgt)))
if tgt in ctx :
net.add_input(tgt, "_w", self.n.Variable("_"))
net.add_output(tgt, "_w", self.n.Variable("_r"))
else :
net = self.n.PetriNet("[%s:%s] %s" % (self._get_pos(node) + (node._src,)))
net.add_transition(self.n.Transition("_t"))
net.add_place(self.n.Place(tgt, status=self.n.buffer(tgt)))
net.add_place(self.n.Place("_e", status=self.n.entry))
net.add_place(self.n.Place("_x", status=self.n.exit))
net.add_input("_e", "_t", self.n.Value(self.n.dot))
net.add_output("_x", "_t", self.n.Value(self.n.dot))
if tgt in ctx :
net.add_input(tgt, "_t", self.n.Variable("_"))
if val.tag == "name" :
net.add_place(self.n.Place(tgt.value))
net.add_input(tgt.value, "_t", self.n.Test(self.n.Variable("_v")))
net.add_output(tgt, "_t", self.n.Variable("_v"))
elif val.tag in ("int", "str") :
net.add_output(tgt, "_t", self.n.Value(repr(val.value)))
elif val.tag == "code" :
net.add_output(tgt, "_t", self.n.Expression(val.value))
else :
self._check(node, False, "unsupported assignment %r" % node._src)
ctx[tgt] = decl(kind="variable")
return net
def visit_call (self, node, ctx) :
self._check(node, node.name in ctx, "undeclared function %r" % node.name)
self._check(node, ctx[node.name].kind == "function",
"%s %s is not a function" % (ctx[node.name].kind, node.name))
try :
argmap = ctx[node.name].args.bind(*node.largs, **node.kwargs)
argmap.apply_defaults()
except Exception as err :
self._check(node, False, str(err))
net = self.n.PetriNet("[%s:%s] %s" % (self._get_pos(node) + (node._src,)))
net.add_place(self.n.Place("_e", status=self.n.entry))
net.add_place(self.n.Place("_i", status=self.n.internal))
net.add_place(self.n.Place("_x", status=self.n.exit))
net.add_transition(self.n.Transition("_c"))
net.add_transition(self.n.Transition("_w"))
net.add_input("_e", "_t", self.n.Value(self.n.dot))
net.add_output("_i", "_t", self.n.Value(self.n.dot))
net.add_input("_i", "_w", self.n.Value(self.n.dot))
net.add_output("_x", "_w", self.n.Variable("_s"))
label = []
for arg, val in argmap.arguments :
if val.kind == "name" :
if not net.has_place(val.value) :
net.add_place(self.n.Place(val.value, status=self.n.entry))
net.add_input(val.value, "_c", self.n.Test(self.n.Variable(val.value)))
label.append(self.n.Variable(val.value))
elif val.kind in ("int", "str") :
label.append(self.n.Value(repr(val.value)))
elif val.kind == "code" :
label.append(self.n.Expression(val.value))
net.add_place(self.n.Place("call_" + node.name))
net.add_output("call_" + node.name, "_c", self.n.Tuple(self.n.dot,
self.n.Tuple(*label)))
net.add_place(self.n.Place("ret_" + node.name))
net.add_input("ret_" + node.name, "_w", self.n.Tuple(self.n.Variable("_s"),
self.n.Variable("_r")))
return net
def visit_if (self, node, ctx) :
pass
def visit_for (self, node, ctx) :
pass
def visit_while (self, node, ctx) :
pass
def visit_try (self, node, ctx) :
pass
......@@ -7,10 +7,6 @@ class Parser (BaseParser) :
__compound__ = spec
self.__parser__ = MP
def parse (source, spec) :
if isinstance(source, str) :
return Parser(spec).parse(source, "<string>")
else :
return Parser(spec).parse(source.read(), getattr(source, "name", "<string>"))
parse = Parser.make_parser()
Compiler = metaparse.Compiler
node = metaparse.node
......
# coding: utf-8
import ast, sys
import ast, sys, collections
import fastidious
class node (object) :
......@@ -114,26 +114,26 @@ class Nest (object) :
return n
class Compiler (object) :
def visit (self, tree) :
def visit (self, tree, *l, **k) :
if isinstance(tree, node) :
method = getattr(self, "visit_" + tree.tag, self.generic_visit)
return method(tree)
return method(tree, *l, **k)
else :
return tree
def generic_visit (self, tree) :
def generic_visit (self, tree, *l, **k) :
sub = {}
ret = {tree.tag : sub}
for name, child in tree :
if isinstance(child, list) :
sub[name] = []
for c in child :
sub[name].append(self.visit(c))
sub[name].append(self.visit(c, *l, **k))
elif isinstance(child, dict) :
sub[name] = {}
for k, v in child.items() :
sub[name][k] = self.visit(v)
sub[name][k] = self.visit(v, *l, **k)
else :
sub[name] = self.visit(child)
sub[name] = self.visit(child, *l, **k)
return ret
class MetaParser (fastidious.Parser) :
......@@ -157,7 +157,6 @@ class MetaParser (fastidious.Parser) :
DEDENT <- _ "↤" NL? _ {_drop}
NUMBER <- _ ~"[+-]?[0-9]+" {_number}
NAME <- _ ~"[a-z][a-z0-9_]*"i _ {p_flatten}
atom <- :name / num:NUMBER / :code / :string {_first}
name <- name:NAME {_name}
code <- codec / codeb {_code}
codec <- LCB (~"([^{}\\\\]|[{}])+" / codec)* RCB {p_flatten}
......@@ -170,14 +169,17 @@ class MetaParser (fastidious.Parser) :
def _drop (self, match) :
return ""
def _number (self, match) :
return node("const", value=int(self.p_flatten(match)), type="int")
return node("const", value=int(self.p_flatten(match)), type="int",
_pos=self.pos, _src=self.p_flatten(match))
def _name (self, match, name) :
return node("const", value=name, type="name")
return node("const", value=name, type="name",
_pos=self.pos, _src=self.p_flatten(match))
def _code (self, match) :
return node("const", value=self.p_flatten(match).strip()[1:-1], type="code")
return node("const", value=self.p_flatten(match).strip()[1:-1], type="code",
_pos=self.pos, _src=self.p_flatten(match))
def _string (self, match) :
return node("const", value=ast.literal_eval(self.p_flatten(match).strip()),
type="str")
type="str", _pos=self.pos, _src=self.p_flatten(match))
def _tuple (self, match) :
lst = []
for m in match :
......@@ -192,11 +194,14 @@ class MetaParser (fastidious.Parser) :
if isinstance(v, node) :
return v
else :
return node(k, value=v)
return node(k, value=v, _pos=self.pos, _src=self.p_flatten(match))
return self.NoMatch
__grammar__ += r"""
model <- stmt:stmt+
stmt <- :call / :block {_first}
stmt <- :assign / :call / :block {_first}
assign <- name:NAME EQ :expr
expr <- :atom / :call {_first}
atom <- :name / num:NUMBER / :code / :string {_first}
call <- name:NAME :args NL
args <- LP ( :arglist )? RP
arglist <- arg (COMMA arg)* {_tuple}
......@@ -241,13 +246,17 @@ class MetaParser (fastidious.Parser) :
return s
def on_model (self, match, stmt) :
return node("model", body=self._glue(stmt), _pos=self.pos)
def on_assign (self, match, name, expr) :
return node("assign", target=name, value=expr,
_pos=self.pos, _src=self.p_flatten(match))
def on_call (self, match, name, args) :
return node("call", name=name, largs=args.l, kargs=args.k, _pos=self.pos)
return node("call", name=name, largs=args.l, kargs=args.k,
_pos=self.pos, _src=self.p_flatten(match))
def on_args (self, match, arglist=None) :
if arglist is self.NoMatch or not arglist :
arglist = []
l = []
k = {}
k = collections.OrderedDict()
pos = True
for arg in arglist :
if arg.kw :
......@@ -258,15 +267,17 @@ class MetaParser (fastidious.Parser) :
else :
self.p_parse_error("forbidden positional arg after a keyword arg",
arg._pos)
return node("args", l=l, k=k)
return node("args", l=l, k=k, _pos=self.pos, _src=self.p_flatten(match))
def on_arg (self, match, left, right=None) :
if right is None :
return node("arg", kw=False, value=left, _pos=self.pos)
return node("arg", kw=False, value=left,
_pos=self.pos, _src=self.p_flatten(match))
elif left.type != "name" :
self.p_parse_error("invalid parameter %r (of type %s)"
% (left.value, left.type))
else :
return node("arg", kw=True, name=left.value, value=right, _pos=self.pos)
return node("arg", kw=True, name=left.value, value=right,
_pos=self.pos, _src=self.p_flatten(match))
def on_block (self, match, name, stmt, args=None, deco=None) :
if args is self.NoMatch or not args :
args = None
......@@ -275,6 +286,7 @@ class MetaParser (fastidious.Parser) :
return self.nest(name, body=self._glue(stmt), args=args, deco=deco,
_pos=self.pos)
def on_deco (self, match, name, args=[]) :
return node("deco", name=name, args=args)
return node("deco", name=name, args=args,
_pos=self.pos, _src=self.p_flatten(match))
def __INIT__ (self) :
self.nest = Nest(self.__compound__)
......
This diff could not be displayed because it is too large.
......@@ -57,6 +57,12 @@ class PetriNet (object) :
return self._place[name]
else :
raise ConstraintError("place %r not found" % name)
def has_place (self, name) :
return name in self._place
def has_transition (self, name) :
return name in self._trans
def __contains__ (self, name) :
return name in self._node
def transition (self, name=None) :
if name is None :
return self._trans.values()
......
import importlib, types, sys, inspect
from .. import ZINCError
class PluginError (ZINCError) :
pass
class Plugin (object) :
modname = "zn"
plugins = []
modules = []
def __init__ (self, *extends, conflicts=[], depends=[]) :
# check conflicts
for p in conflicts :
if p in self.plugins :
raise PluginError("plugin conflicts with %r" % p)
# load dependencies
for p in depends :
try :
return importlib.import_module(p)
except :
try :
return importlib.import_module("zinc.plugins." + p)
except :
raise PluginError("could not load %r from '.' or 'zinc.plugins'" % p)
# create result module
if self.modname not in sys.modules :
self._mod = sys.modules[self.modname] = types.ModuleType(self.modname)
# load extended modules
for base in extends :
if base not in self.modules :
mod = importlib.import_module(base)
self._mod.__dict__.update(mod.__dict__)
self.modules.append(base)
# record loaded plugin
stack = inspect.stack()
caller = inspect.getmodule(stack[1][0])
self.plugins.append(caller.__name__)
def __getattr__ (self, name) :
return getattr(self._mod, name)
def __call__ (self, obj) :
setattr(self._mod, obj.__name__, obj)
return obj
@classmethod
def reset_plugins (cls, name="zn") :
for p in cls.plugins + cls.modules + [cls.modname] :
del sys.modules[p]
cls.modname = name
cls.plugins = []
cls.modules = []
"""An example plugin that allows instances of `PetriNet` to say hello.
The source code can be used as a starting example:
1. Import `zinc.plugins.Plugin`
2. Create an instance of it, passing:
* the modules that are extended
* the conflicts (if any)
* the dependencies (if any)
3. This instance will serve as:
* a decorator for each class or function that should be included
in the extended module
* classes or functions to extend are retreived as its attributes
"""
from . import Plugin
plug = Plugin("zinc.nets")
@plug
class PetriNet (plug.PetriNet) :
def hello (self, name="world") :
print("hello %s from %s" % (name, self.name))
import enum
from zinc.data import hashable, mset
class Token (object) :
pass
class CodeToken (Token) :
def __init__ (self, code) :
self.code = code
def __ast__ (self, place, ctx) :
......@@ -37,21 +41,17 @@ class Token (object) :
def __lt__ (self, other) :
return not self.__ge__(other)
class BlackToken (Token) :
def __init__ (self) :
pass
def __str__ (self) :
return "dot"
def __repr__ (self) :
return "dot"
def __new__ (cls) :
if not hasattr(cls, "dot") :
cls.dot = object.__new__(cls)
return cls.dot
def __hash__ (self) :
return hash(self.__class__.__name__)
class BlackToken (Token, enum.IntEnum) :
DOT = enum.auto()
dot = BlackToken.DOT
class BlackWhiteToken (Token, enum.IntEnum) :
BLACK = enum.auto()
WHITE = enum.auto()
dot = BlackToken()
black = BlackWhiteToken.BLACK
white = BlackWhiteToken.WHITE
@hashable
class Marking (dict) :
......