STATEMENT = CONDITION | OPERATION | SUBTREE
OPERATION = WS VARNAME WS ( '=' | '+=' ) WS EXPRESSION
-SUBTREE = WS VARNAME WS '{' BLOCK '}'
+SUBTREE = WS VARNAME WS '{' BLOCK WS '}'
CONDITION = WS 'if' FORMULA WS '{' BLOCK WS '}'
FORMULA = WS (( EXPRESSION WS ( '!=' | '==' ) WS EXPRESSION ) | '(' AND WS ')' | '(' OR WS ')' | NOT )
import re, types, itertools, logging as log
import traceback
-import moe.conf
+import moe.conf as conf
+
class ConfigSyntaxError(conf.ConfigError):
+
def __init__(self, msg, fname='<unknown>', line=None, column=None):
self.msg = msg
self.fname = fname
self.line = line
self.column = column
+
def __str__(self):
return('ConfigSyntaxError %s:%d:%d: %s'%(self.fname, self.line, self.column, self.msg))
+
class ConfigParser(object):
c_varname_sep = u'.'
c_comment = u'#'
c_neq = u'!='
c_set = u'='
c_append = u'+='
+
def __init__(self, s, tree, fname='<unknown>', level=0):
"""Create a config file parser.
`s` is either a string, unicode or an open file. File is assumed to be utf-8, string is converted to unicode.
`fname` is an optional name of the file, for debugging and syntax errors.
`level` indicates the precedence the operations should have in the ConfigTree
"""
- self.s = s # Unicode, string or an open file
+ self.s = s # Unicode, ascii string or an open file
self.buf = u"" # Read-buffer for s file, whole unicode string for s string/unicode
if isinstance(self.s, types.StringTypes):
self.buf = unicode(self.s)
self.prefix = '' # Prefix of variable name, may begin with '.'
self.conditions = [] # Stack of nested conditions, these are chained, so only the last is necessary
self.read_ops = [] # List of parsed operations (varname, `Operation`), returned by `self.parse()`
+
def preread(self, l):
"Make sure buf contains at least `l` next characters, return True on succes and False on hitting EOF."
if isinstance(self.s, file):
self.buf = self.buf[self.bufpos:] + self.s.read(max(l, 1024)).decode('utf8')
self.bufpos = 0
return len(self.buf) >= self.bufpos + l
+
def peek(self, l = 1):
"Peek and return next `l` unicode characters or everything until EOF."
self.preread(l)
return self.buf[self.bufpos:self.bufpos+l]
+
def peeks(self, s):
"Peek and compare next `len(s)` characters to `s`. Converts `s` to unicode. False on hitting EOF."
s = unicode(s)
return self.peek(len(s)) == s
+
def next(self, l = 1):
"Eat and return next `l` unicode characters. Raise exception on EOF."
if not self.preread(l):
self.line += s.count('\n')
self.column = l - rnl - 1
return s
+
def nexts(self, s):
"""Compare next `len(s)` characters to `s`. On match, eat them and return True. Otherwise just return False.
Converts `s` to unicode. False on hitting EOF."""
self.next(len(s))
return True
return False
+
def eof(self):
"Check for end-of-stream."
return not self.preread(1)
+
def expect(self, s, msg=None):
"Eat and compare next `len(s)` characters to `s`. If not equal, raise an error with `msg`. Unicode."
s = unicode(s)
if not self.nexts(s):
self.syntax_error(msg or u"%r expected."%(s,))
+
def syntax_error(self, msg, *args):
"Raise a syntax error with file/line/column info"
raise ConfigSyntaxError(fname=self.fname, line=self.line, column=self.column, msg=(msg%args))
+
def dbg(self):
n = None; s = ''
for i in traceback.extract_stack():
s += ' '
n = i[2]
if n: log.debug(s + n + ' ' + repr(self.peek(15)) + '...')
+
def parse(self):
self.read_ops = []
self.p_BLOCK()
return self.read_ops
+
def p_BLOCK(self):
self.dbg() # Debug
self.p_WS()
else:
self.nexts(';') # NOTE: this is weird - can ';' occur anywhere? Or at most once, but only after any p_WS debris?
self.p_WS()
+
def p_WS(self):
self.dbg() # Debug
while not self.eof():
self.p_COMMENT()
else:
break
+
def p_COMMENT(self):
self.dbg() # Debug
self.expect(self.c_comment, "'#' expected at the beginning of a comment.")
while (not self.eof()) and (not self.nexts(self.c_nl)):
self.next(1)
+
def p_STATEMENT(self):
self.dbg() # Debug
self.p_WS()
self.p_SUBTREE(varname)
else:
self.p_OPERATION(varname)
+
def p_SUBTREE(self, varname=None):
self.dbg() # Debug
if not varname:
# close block and
self.p_WS()
self.expect(self.c_close)
+
def p_OPERATION(self, varname=None):
self.dbg() # Debug
if not varname:
# NOTE/WARNING: The last character of operation will be reported in case of error.
v.add_operation(op)
self.read_ops.append( (vname, op) )
+
def p_CONDITION(self):
self.dbg() # Debug
self.p_WS()
self.expect(self.c_close)
# Cleanup
self.conditions.pop()
+
def p_VARNAME(self):
self.dbg() # Debug
vnl = []
if not conf.re_VARNAME.match(vn):
self.syntax_error('Invalid variable name %r', vn)
return vn
+
def p_EXPRESSION(self):
self.dbg() # Debug
op = self.next()
else:
expr2.append(i)
return conf.ConfigExpression(tuple(expr2), exs)
+
def p_FORMULA(self):
self.dbg() # Debug
self.p_WS()
projects / eval.git / blobdiff