"""
This contains utilities for logging the program trace and prepending version to the file logs.
"""
import asyncio
import inspect
import logging
import os
import sys
import threading
def _abspath(path):
return os.path.abspath(os.path.join(os.curdir, path))
class _NoSource(RuntimeError):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.paths = []
def add_path(self, path):
self.paths.append(path)
def get_paths(self):
return self.paths
def _parse_trace_files(tfile_list, err):
trace_files_paths = tfile_list.split(',')
for trace_file_path in trace_files_paths:
trace_file_abspath = _abspath(trace_file_path)
if not os.path.exists(trace_file_abspath):
err.add_path(trace_file_abspath)
else:
yield trace_file_abspath
[docs]def get_trace_files():
"""
This function returns list of files to trace.
A list of files to show trace output from can be passed to the trace() executor within two ways:
1. An environment variable named TRACE_LIST, containing comma-separated list of curdir-relative paths to the
source files to trace.
Usage:
.. code::
TRACE_LIST="some_dir/some_file.py,some_dir/some_file1.py" python ailocals_capable_app.py
2. An environment variable named TRACE_FILE, which is a path to a file,comma-separated list of curdir-relative
paths to the source files to trace.
Usage:
.. code::
cat "some_dir/some_file.py,some_dir/some_file1.py" > some_dir/trace.csv
TRACE_FILE="some_dir/trace.csv" python ailocals_capable_app.py
If there are both TRACE_FILE and TRACE_LIST passed to os.environ, trace files will be combined from each output.
The trace files can also be added with add_trace_file(filename)
:return: List of files to trace
:rtype: list
"""
tfiles = []
err = _NoSource("Tracing files output missed: %s")
trace_list = os.environ.get('TRACE_LIST', None)
trace_file = os.environ.get('TRACE_FILE', None)
if trace_list:
tfiles += list(_parse_trace_files(trace_list, err))
if trace_file:
trace_file_abspath = _abspath(trace_file)
if not os.path.exists(trace_file_abspath):
err.add_path(trace_file_abspath)
else:
try:
with open(trace_file_abspath, "r") as fl:
trace_file_payload = fl.read()
except OSError:
err.add_path(trace_file_abspath)
else:
tfiles += list(_parse_trace_files(trace_file_payload, err))
if err.get_paths():
raise err
return tfiles
try:
trace_files = get_trace_files()
except _NoSource as e:
logging.error("%s: %s" % (e[0], ",".join(e.get_paths())))
trace_files = []
def _in_tracefiles(path):
for trace_file in trace_files:
if trace_file.endswith(path):
result = True
break
else:
result = False
return result
[docs]def trace(message=None):
"""
This outputs debug message into logs, which contains the next prefix:
TRACE: ``{filename}:{lineno} in {scope}({locals})``
where
- filename is the name of the file where this tracing shortcut was called
- lineno is the line of the source file where this tracing call was issued
- scope is the string representation of scope where this tracing call was issued
- locals is the representation of variables, which are local to the given scope
:param message: Optional message to send alongwith tracing prefix.
:type message: str, None
"""
if not trace_files:
return
stack = inspect.stack()
frame, filename, lineno, scope, code, x = stack[1]
# only show trace calls for watched files
if not _in_tracefiles(filename):
return
# add args if there is no message
flat_args = ""
if not message:
_, _, _, args = inspect.getargvalues(frame)
if 'self' in args:
del args['self']
flat_args = ', '.join(["%s=%s" % (k, v) for k, v in args.items()])
text = 'TRACE: %s:%s in %s(%s)' % (filename, lineno, scope, flat_args)
if message:
# convert unicode messages to ASCII
message = message.encode('ascii', 'backslashreplace')
text = text + " - " + str(message)
logging.info(text)
[docs]def add_trace_file(filename):
"""
Adds path which is relative to os.curdir to the list of tracing files
:param filename:
:return:
"""
trace_file_abspath = _abspath(filename)
if not os.path.exists(trace_file_abspath):
raise _NoSource(trace_file_abspath)
else:
trace_files.append(trace_file_abspath)
def qual(clazz):
return clazz.__module__ + '.' + clazz.__name__
def _get_task_ident():
try:
loop = asyncio.get_event_loop()
except AssertionError:
return 0
if loop.is_running():
task = asyncio.Task.current_task()
task_id = id(task)
return task_id
else:
return 0
[docs]def spewer(frame, s, ignored):
"""
A trace function for sys.settrace that prints every function or method call alongside with its task ID.
"""
ident_prefix = "%s: %s "
task_id = _get_task_ident()
if task_id:
ident_prefix = ident_prefix % ("task ID", task_id)
else:
ident_prefix = ident_prefix % ("thread ID", threading.get_ident())
if 'self' in frame.f_locals:
se = frame.f_locals['self']
if hasattr(se, '__class__'):
k = qual(se.__class__)
else:
k = qual(type(se))
print('%s: method %s of %s at %s' % (
ident_prefix, frame.f_code.co_name, k, id(se)))
else:
print('%s function %s in %s, line %s' % (
ident_prefix,
frame.f_code.co_name,
frame.f_code.co_filename,
frame.f_lineno))
[docs]def spew():
"""
Start print program tracing information for every statement.
"""
sys.settrace(spewer)