'''Functions for tracing Curry evaluation.'''
import collections, contextlib
# The context manager used when tracing is off. The bookkeeping of the trace
# has no effect then, so the evaluator skips it.
NULL_CONTEXT = contextlib.nullcontext()
[docs]
def show_queue(rts, qid=None):
qid = rts.qid if qid is None else qid
return 'queue %s: %s' % (qid, list(e.fingerprint for e in rts.qtable[qid]))
def _showexpr(expr):
from .... import show
return show.show(expr, stringifier=show.PlainFreeVarsStringifier())
[docs]
def enter_rewrite(rts, indent, expr):
if rts.tracing:
print('S <<< %s%s' % (' ' * indent, _showexpr(expr)))
[docs]
def exit_rewrite(rts, indent, expr):
if rts.tracing:
from .... import show
print('S >>> %s%s' % (' ' * indent, _showexpr(expr)))
[docs]
def failed(rts, qid=None):
if rts.tracing:
print('Q ::: failed config dropped from %s' % show_queue(rts, qid))
[docs]
def yield_(rts, value):
if rts.tracing:
print('Y ::: %s' % value)
[docs]
def position(rts, indent, expr, path):
if rts.tracing:
print('I ::: %sat path=%s of %s' % (
' ' * (indent + 1)
, '[' + ','.join(map(str, path)) + ']'
, expr
))
[docs]
def activate_queue(rts, qid):
if rts.tracing:
print('Q ::: switching to %s' % show_queue(rts, qid))
[docs]
def fork(rts, qid=None):
if rts.tracing:
return _fork(rts, qid)
else:
return NULL_CONTEXT
@contextlib.contextmanager
def _fork(rts, qid):
qid = rts.qid if qid is None else qid
cid = rts.grp_id()
head_fp = str(rts.qtable[qid][0].fingerprint)
try:
yield
finally:
print('Q ::: fork %s on cid=%s appending to %s' % (
head_fp, cid, show_queue(rts, qid)
))
[docs]
def expr_id(expr):
from .... import inspect
return id(inspect.fwd_chain_target(expr))
[docs]
class Trace(object):
'''
This object calls the freestanding trace functions to print log output. It
keeps track of the indent level and filters out clutter. As an example of
the latter, when an expression being inspected was just created in the
previous step, it is not repeated.
'''
def __init__(self, rts):
self.rts = rts
self.indents = collections.defaultdict(int)
self.prevexprs = collections.defaultdict(lambda: None)
self.prevpaths = collections.defaultdict(lambda: None)
[docs]
def indent(self, qid=None):
qid = self.rts.qid if qid is None else qid
self.indent_value = self.indents[qid]
self.indents[qid] += 1
[docs]
def dedent(self, qid=None):
qid = self.rts.qid if qid is None else qid
self.indents[qid] -= 1
self.indent_value = self.indents[qid]
[docs]
def enter_rewrite(self, expr):
self.indent()
expr = getattr(expr, 'target', expr)
if self.prevexprs[self.rts.qid] != expr_id(expr):
enter_rewrite(self.rts, self.indent_value, expr)
[docs]
def exit_rewrite(self, expr):
self.dedent()
expr = getattr(expr, 'target', expr)
exit_rewrite(self.rts, self.indent_value, expr)
self.prevexprs[self.rts.qid] = expr_id(expr)
[docs]
def failed(self):
failed(self.rts)
[docs]
def yield_(self, value):
yield_(self.rts, value)
[docs]
def activate_queue(self, qid):
activate_queue(self.rts, qid)
[docs]
def fork(self, qid=None):
return fork(self.rts, qid)
[docs]
def position(self, expr, path):
if self.rts.tracing:
return self._position(expr, path)
else:
return NULL_CONTEXT
@contextlib.contextmanager
def _position(self, expr, path):
qid = self.rts.qid
key = expr_id(expr), tuple(path)
self.indent(qid)
try:
if self.prevpaths[qid] != key:
position(self.rts, self.indent_value, expr, path)
yield
self.prevpaths[qid] = expr_id(expr), tuple(path)
finally:
self.dedent(qid)
[docs]
def trace_values(f):
'''Wraps the D procedure to trace value creation.'''
def value_tracer(rts):
for value in f(rts):
rts.trace.yield_(value)
yield value
return value_tracer
[docs]
def trace_steps(f):
'''Wraps the S procedure to trace the application of steps.'''
def step_tracer(rts, node, *args, **kwds):
if not rts.tracing:
return f(rts, node, *args, **kwds)
rts.trace.enter_rewrite(node)
try:
return f(rts, node, *args, **kwds)
finally:
rts.trace.exit_rewrite(node)
return step_tracer