src.hunter.ColorStreamAction.stream() - Code Metrics - Inspection of "Update docstrings." - ionelmc/python-hunter - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 44d803...8bc4ee )

by Ionel Cristian

created 2016-01-24 17:50 UTC

src.hunter.ColorStreamAction.stream() A

↳ Parent: src.hunter.ColorStreamAction

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	19
rs	9.4285
cc	1

from __future__ import absolute_import

import ast
import os
import pdb
import sys

from colorama import AnsiToWin32
from colorama import Back
from colorama import Fore
from colorama import Style
from six import string_types

from .util import Fields

NO_COLORS = {
    'reset': '',
    'filename': '',
    'colon': '',
    'lineno': '',
    'kind': '',
    'continuation': '',
    'return': '',
    'exception': '',
    'detail': '',
    'vars': '',
    'vars-name': '',
    'call': '',
    'line': '',
    'internal-failure': '',
    'internal-detail': '',
    'source-failure': '',
    'source-detail': '',
}
EVENT_COLORS = {
    'reset': Style.RESET_ALL,
    'normal': Style.NORMAL,
    'filename': '',
    'colon': Style.BRIGHT + Fore.BLACK,
    'lineno': Style.RESET_ALL,
    'kind': Fore.CYAN,
    'continuation': Style.BRIGHT + Fore.BLUE,
    'call': Style.BRIGHT + Fore.BLUE,
    'return': Style.BRIGHT + Fore.GREEN,
    'exception': Style.BRIGHT + Fore.RED,
    'detail': Style.NORMAL,
    'vars': Style.RESET_ALL + Fore.MAGENTA,
    'vars-name': Style.BRIGHT,
    'internal-failure': Style.BRIGHT + Back.RED + Fore.RED,
    'internal-detail': Fore.WHITE,
    'source-failure': Style.BRIGHT + Back.YELLOW + Fore.YELLOW,
    'source-detail': Fore.WHITE,
}
CODE_COLORS = {
    'call': Fore.RESET + Style.BRIGHT,
    'line': Fore.RESET,
    'return': Fore.YELLOW,
    'exception': Fore.RED,
}
MISSING = type('MISSING', (), {'__repr__': lambda _: '?'})()


class Action(object):
    def __call__(self, event):
        raise NotImplementedError()


class Debugger(Fields.klass.kwargs, Action):
    """
    An action that starts ``pdb``.
    """

    def __init__(self, klass=pdb.Pdb, **kwargs):
        self.klass = klass
        self.kwargs = kwargs

    def __call__(self, event):
        """
        Runs a ``pdb.set_trace`` at the matching frame.
        """
        self.klass(**self.kwargs).set_trace(event.frame)


class ColorStreamAction(Fields.stream.force_colors.filename_alignment.repr_limit, Action):
    _stream_cache = {}
    _stream = None
    _tty = None

    def __init__(self,
                 stream=sys.stderr,
                 force_colors=False,
                 filename_alignment=40,
                 repr_limit=1024):
        self.force_colors = force_colors
        self.stream = stream
        self.filename_alignment = max(5, filename_alignment)
        self.repr_limit = repr_limit

    @property
    def stream(self):
        return self._stream

    @stream.setter
    def stream(self, value):
        if isinstance(value, string_types):
            if value in self._stream_cache:
                value = self._stream_cache[value]
            else:
                value = self._stream_cache[value] = open(value, 'a', buffering=0)

        isatty = getattr(value, 'isatty', None)
        if self.force_colors or (isatty and isatty() and os.name != 'java'):
            self._stream = AnsiToWin32(value, strip=False)
            self._tty = True
            self.event_colors = EVENT_COLORS
            self.code_colors = CODE_COLORS
        else:
            self._tty = False
            self._stream = value
            self.event_colors = NO_COLORS
            self.code_colors = NO_COLORS

    def _safe_repr(self, obj):
        limit = self.repr_limit

        try:
            s = repr(obj)
            s = s.replace('\n', r'\n')
            if len(s) > limit:
                cutoff = limit // 2
                return "{} {continuation}[...]{reset} {}".format(s[:cutoff], s[-cutoff:], **self.event_colors)
            else:
                return s
        except Exception as exc:
            return "{internal-failure}!!! FAILED REPR: {internal-detail}{!r}{reset}".format(exc, **self.event_colors)


class CodePrinter(ColorStreamAction):
    """
    An action that just prints the code being executed.

    Args:
        stream (file-like): Stream to write to. Default: ``sys.stderr``.
        filename_alignment (int): Default size for the filename column (files are right-aligned). Default: ``40``.
        force_colors (bool): Force coloring. Default: ``False``.
        repr_limit (bool): Limit length of ``repr()`` output. Default: ``512``.
    """
    def _safe_source(self, event):
        try:
            lines = event._raw_fullsource.rstrip().splitlines()
            if lines:
                return lines
            else:
                return "{source-failure}??? NO SOURCE: {source-detail}" \
                       "Source code string for module {!r} is empty.".format(event.module, **self.event_colors),
            return lines
        except Exception as exc:
            return "{source-failure}??? NO SOURCE: {source-detail}{!r}".format(exc, **self.event_colors),

    def _format_filename(self, event):
        filename = event.filename or "<???>"
        if len(filename) > self.filename_alignment:
            filename = '[...]{}'.format(filename[5 - self.filename_alignment:])
        return filename

    def __call__(self, event, sep=os.path.sep, join=os.path.join):
        """
        Handle event and print filename, line number and source code. If event.kind is a `return` or `exception` also
        prints values.
        """

        # context = event.tracer
        # alignment = context.filename_alignment = max(
        #     getattr(context, 'filename_alignment', 5),
        #     len(filename)
        # )
        lines = self._safe_source(event)
        self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {code}{}{reset}\n".format(
            self._format_filename(event),
            event.lineno,
            event.kind,
            lines[0],
            align=self.filename_alignment,
            code=self.code_colors[event.kind],
            **self.event_colors
        ))
        for line in lines[1:]:
            self.stream.write("{:>{align}}       {kind}{:9} {code}{}{reset}\n".format(
                "",
                r"   |",
                line,
                align=self.filename_alignment,
                code=self.code_colors[event.kind],
                **self.event_colors
            ))

        if event.kind in ('return', 'exception'):
            self.stream.write("{:>{align}}       {continuation}{:9} {color}{} value: {detail}{}{reset}\n".format(
                "",
                "...",
                event.kind,
                self._safe_repr(event.arg),
                align=self.filename_alignment,
                color=self.event_colors[event.kind],
                **self.event_colors
            ))


class CallPrinter(CodePrinter):
    """
    An action that just prints the code being executed, but unlike :obj:`hunter.CodePrinter` it indents based on
    callstack depth and it also shows ``repr()`` of function arguments.

    Args:
        stream (file-like): Stream to write to. Default: ``sys.stderr``.
        filename_alignment (int): Default size for the filename column (files are right-aligned). Default: ``40``.
        force_colors (bool): Force coloring. Default: ``False``.
        repr_limit (bool): Limit length of ``repr()`` output. Default: ``512``.

    .. versionadded:: 1.2.0

    .. note::

        This will be the default action in `hunter 2.0`.
    """

    def __init__(self, **options):
        super(CallPrinter, self).__init__(**options)
        self.stack = []

    def __call__(self, event, sep=os.path.sep, join=os.path.join):
        """
        Handle event and print filename, line number and source code. If event.kind is a `return` or `exception` also
        prints values.
        """
        filename = self._format_filename(event)
        ident = event.module, event.function

        if event.kind == 'call':
            code = event.code
            self.stack.append(ident)
            self.stream.write(
                "{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {}{call}=>{normal} "
                "{}({}{call}{normal}){reset}\n".format(
                    filename,
                    event.lineno,
                    event.kind,
                    '   ' * (len(self.stack) - 1),
                    event.function,
                    ', '.join('{vars}{vars-name}{0}{vars}={reset}{1}'.format(
                        var,
                        self._safe_repr(event.locals.get(var, MISSING)),
                        **self.event_colors
                    ) for var in code.co_varnames[:code.co_argcount]),
                    align=self.filename_alignment,
                    **self.event_colors
                ))
        elif event.kind in ('return', 'exception'):
            self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {code}{}{}{normal} {}: {reset}{}\n".format(
                filename,
                event.lineno,
                event.kind,
                '   ' * (len(self.stack) - 1),
                {'return': '<=', 'exception': '<!'}[event.kind],
                event.function,
                self._safe_repr(event.arg),
                align=self.filename_alignment,
                code=self.event_colors[event.kind],
                **self.event_colors
            ))
            if self.stack and self.stack[-1] == ident:
                self.stack.pop()
        else:
            self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {reset}{}{}\n".format(
                filename,
                event.lineno,
                event.kind,
                '   ' * len(self.stack),
                event.source.strip(),
                align=self.filename_alignment,
                code=self.code_colors[event.kind],
                **self.event_colors
            ))


class VarsPrinter(Fields.names.globals.stream.filename_alignment, ColorStreamAction):
    """
    An action that prints local variables and optionally global variables visible from the current executing frame.

    Args:
        *names (strings): Names to evaluate. Expressions can be used (will only try to evaluate if all the variables are
            present on the frame.
        globals (bool): Allow access to globals. Default: ``False`` (only looks at locals).
        stream (file-like): Stream to write to. Default: ``sys.stderr``.
        filename_alignment (int): Default size for the filename column (files are right-aligned). Default: ``40``.
        force_colors (bool): Force coloring. Default: ``False``.
        repr_limit (bool): Limit length of ``repr()`` output. Default: ``512``.

    .. note::
        This is the default action.

    .. warning::

        In `hunter 2.0` the default action will be :obj:`hunter.CallPrinter`.
    """

    def __init__(self, *names, **options):
        if not names:
            raise TypeError("VarsPrinter requires at least one variable name/expression.")
        self.names = {
            name: set(self._iter_symbols(name))
            for name in names
        }
        self.globals = options.pop('globals', False)
        super(VarsPrinter, self).__init__(**options)

    @staticmethod
    def _iter_symbols(code):
        """
        Iterate all the variable names in the given expression.

        Example:

        * ``self.foobar`` yields ``self``
        * ``self[foobar]`` yields `self`` and ``foobar``
        """
        for node in ast.walk(ast.parse(code)):
            if isinstance(node, ast.Name):
                yield node.id

    def _safe_eval(self, code, event):
        """
        Try to evaluate the given code on the given frame. If failure occurs, returns some ugly string with exception.
        """
        try:
            return eval(code, event.globals if self.globals else {}, event.locals)
        except Exception as exc:
            return "{internal-failure}FAILED EVAL: {internal-detail}{!r}".format(exc, **self.event_colors)

    def __call__(self, event):
        """
        Handle event and print the specified variables.
        """
        first = True
        frame_symbols = set(event.locals)
        if self.globals:
            frame_symbols |= set(event.globals)

        for code, symbols in self.names.items():
            try:
                obj = eval(code, event.globals if self.globals else {}, event.locals)
            except AttributeError:
                continue
            except Exception as exc:
                printout = "{internal-failure}FAILED EVAL: {internal-detail}{!r}".format(exc, **self.event_colors)
            else:
                printout = self._safe_repr(obj)

            if frame_symbols >= symbols:
                self.stream.write("{:>{align}}       {vars}{:9} {vars-name}{} {vars}=> {reset}{}{reset}\n".format(
                    "",
                    "vars" if first else "...",
                    code,
                    printout,
                    align=self.filename_alignment,
                    **self.event_colors
                ))
                first = False


1			from __future__ import absolute_import
2
3			import ast
4			import os
5			import pdb
6			import sys
7
8			from colorama import AnsiToWin32
9			from colorama import Back
10			from colorama import Fore
11			from colorama import Style
12			from six import string_types
13
14			from .util import Fields
15
16			NO_COLORS = {
17			'reset': '',
18			'filename': '',
19			'colon': '',
20			'lineno': '',
21			'kind': '',
22			'continuation': '',
23			'return': '',
24			'exception': '',
25			'detail': '',
26			'vars': '',
27			'vars-name': '',
28			'call': '',
29			'line': '',
30			'internal-failure': '',
31			'internal-detail': '',
32			'source-failure': '',
33			'source-detail': '',
34			}
35			EVENT_COLORS = {
36			'reset': Style.RESET_ALL,
37			'normal': Style.NORMAL,
38			'filename': '',
39			'colon': Style.BRIGHT + Fore.BLACK,
40			'lineno': Style.RESET_ALL,
41			'kind': Fore.CYAN,
42			'continuation': Style.BRIGHT + Fore.BLUE,
43			'call': Style.BRIGHT + Fore.BLUE,
44			'return': Style.BRIGHT + Fore.GREEN,
45			'exception': Style.BRIGHT + Fore.RED,
46			'detail': Style.NORMAL,
47			'vars': Style.RESET_ALL + Fore.MAGENTA,
48			'vars-name': Style.BRIGHT,
49			'internal-failure': Style.BRIGHT + Back.RED + Fore.RED,
50			'internal-detail': Fore.WHITE,
51			'source-failure': Style.BRIGHT + Back.YELLOW + Fore.YELLOW,
52			'source-detail': Fore.WHITE,
53			}
54			CODE_COLORS = {
55			'call': Fore.RESET + Style.BRIGHT,
56			'line': Fore.RESET,
57			'return': Fore.YELLOW,
58			'exception': Fore.RED,
59			}
60			MISSING = type('MISSING', (), {'__repr__': lambda _: '?'})()
61
62
63			class Action(object):
64			def __call__(self, event):
65			raise NotImplementedError()
66
67
68			class Debugger(Fields.klass.kwargs, Action):
69			"""
70			An action that starts ``pdb``.
71			"""
72
73			def __init__(self, klass=pdb.Pdb, **kwargs):
74			self.klass = klass
75			self.kwargs = kwargs
76
77			def __call__(self, event):
78			"""
79			Runs a ``pdb.set_trace`` at the matching frame.
80			"""
81			self.klass(**self.kwargs).set_trace(event.frame)
82
83
84			class ColorStreamAction(Fields.stream.force_colors.filename_alignment.repr_limit, Action):
85			_stream_cache = {}
86			_stream = None
87			_tty = None
88
89			def __init__(self,
90			stream=sys.stderr,
91			force_colors=False,
92			filename_alignment=40,
93			repr_limit=1024):
94			self.force_colors = force_colors
95			self.stream = stream
96			self.filename_alignment = max(5, filename_alignment)
97			self.repr_limit = repr_limit
98
99			@property
100			def stream(self):
101			return self._stream
102
103			@stream.setter
104			def stream(self, value):
105			if isinstance(value, string_types):
106			if value in self._stream_cache:
107			value = self._stream_cache[value]
108			else:
109			value = self._stream_cache[value] = open(value, 'a', buffering=0)
110
111			isatty = getattr(value, 'isatty', None)
112			if self.force_colors or (isatty and isatty() and os.name != 'java'):
113			self._stream = AnsiToWin32(value, strip=False)
114			self._tty = True
115			self.event_colors = EVENT_COLORS
116			self.code_colors = CODE_COLORS
117			else:
118			self._tty = False
119			self._stream = value
120			self.event_colors = NO_COLORS
121			self.code_colors = NO_COLORS
122
123			def _safe_repr(self, obj):
124			limit = self.repr_limit
125
126			try:
127			s = repr(obj)
128			s = s.replace('\n', r'\n')
129			if len(s) > limit:
130			cutoff = limit // 2
131			return "{} {continuation}[...]{reset} {}".format(s[:cutoff], s[-cutoff:], **self.event_colors)
132			else:
133			return s
134			except Exception as exc:
135			return "{internal-failure}!!! FAILED REPR: {internal-detail}{!r}{reset}".format(exc, **self.event_colors)
136
137
138			class CodePrinter(ColorStreamAction):
139			"""
140			An action that just prints the code being executed.
141
142			Args:
143			stream (file-like): Stream to write to. Default: ``sys.stderr``.
144			filename_alignment (int): Default size for the filename column (files are right-aligned). Default: ``40``.
145			force_colors (bool): Force coloring. Default: ``False``.
146			repr_limit (bool): Limit length of ``repr()`` output. Default: ``512``.
147			"""
148			def _safe_source(self, event):
149			try:
150			lines = event._raw_fullsource.rstrip().splitlines()
151			if lines:
152			return lines
153			else:
154			return "{source-failure}??? NO SOURCE: {source-detail}" \
155			"Source code string for module {!r} is empty.".format(event.module, **self.event_colors),
156			return lines
157			except Exception as exc:
158			return "{source-failure}??? NO SOURCE: {source-detail}{!r}".format(exc, **self.event_colors),
159
160			def _format_filename(self, event):
161			filename = event.filename or "<???>"
162			if len(filename) > self.filename_alignment:
163			filename = '[...]{}'.format(filename[5 - self.filename_alignment:])
164			return filename
165
166			def __call__(self, event, sep=os.path.sep, join=os.path.join):
167			"""
168			Handle event and print filename, line number and source code. If event.kind is a `return` or `exception` also
169			prints values.
170			"""
171
172			# context = event.tracer
173			# alignment = context.filename_alignment = max(
174			# getattr(context, 'filename_alignment', 5),
175			# len(filename)
176			# )
177			lines = self._safe_source(event)
178			self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {code}{}{reset}\n".format(
179			self._format_filename(event),
180			event.lineno,
181			event.kind,
182			lines[0],
183			align=self.filename_alignment,
184			code=self.code_colors[event.kind],
185			**self.event_colors
186			))
187			for line in lines[1:]:
188			self.stream.write("{:>{align}} {kind}{:9} {code}{}{reset}\n".format(
189			"",
190			r" \|",
191			line,
192			align=self.filename_alignment,
193			code=self.code_colors[event.kind],
194			**self.event_colors
195			))
196
197			if event.kind in ('return', 'exception'):
198			self.stream.write("{:>{align}} {continuation}{:9} {color}{} value: {detail}{}{reset}\n".format(
199			"",
200			"...",
201			event.kind,
202			self._safe_repr(event.arg),
203			align=self.filename_alignment,
204			color=self.event_colors[event.kind],
205			**self.event_colors
206			))
207
208
209			class CallPrinter(CodePrinter):
210			"""
211			An action that just prints the code being executed, but unlike :obj:`hunter.CodePrinter` it indents based on
212			callstack depth and it also shows ``repr()`` of function arguments.
213
214			Args:
215			stream (file-like): Stream to write to. Default: ``sys.stderr``.
216			filename_alignment (int): Default size for the filename column (files are right-aligned). Default: ``40``.
217			force_colors (bool): Force coloring. Default: ``False``.
218			repr_limit (bool): Limit length of ``repr()`` output. Default: ``512``.
219
220			.. versionadded:: 1.2.0
221
222			.. note::
223
224			This will be the default action in `hunter 2.0`.
225			"""
226
227			def __init__(self, **options):
228			super(CallPrinter, self).__init__(**options)
229			self.stack = []
230
231			def __call__(self, event, sep=os.path.sep, join=os.path.join):
232			"""
233			Handle event and print filename, line number and source code. If event.kind is a `return` or `exception` also
234			prints values.
235			"""
236			filename = self._format_filename(event)
237			ident = event.module, event.function
238
239			if event.kind == 'call':
240			code = event.code
241			self.stack.append(ident)
242			self.stream.write(
243			"{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {}{call}=>{normal} "
244			"{}({}{call}{normal}){reset}\n".format(
245			filename,
246			event.lineno,
247			event.kind,
248			' ' * (len(self.stack) - 1),
249			event.function,
250			', '.join('{vars}{vars-name}{0}{vars}={reset}{1}'.format(
251			var,
252			self._safe_repr(event.locals.get(var, MISSING)),
253			**self.event_colors
254			) for var in code.co_varnames[:code.co_argcount]),
255			align=self.filename_alignment,
256			**self.event_colors
257			))
258			elif event.kind in ('return', 'exception'):
259			self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {code}{}{}{normal} {}: {reset}{}\n".format(
260			filename,
261			event.lineno,
262			event.kind,
263			' ' * (len(self.stack) - 1),
264			{'return': '<=', 'exception': '<!'}[event.kind],
265			event.function,
266			self._safe_repr(event.arg),
267			align=self.filename_alignment,
268			code=self.event_colors[event.kind],
269			**self.event_colors
270			))
271			if self.stack and self.stack[-1] == ident:
272			self.stack.pop()
273			else:
274			self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {reset}{}{}\n".format(
275			filename,
276			event.lineno,
277			event.kind,
278			' ' * len(self.stack),
279			event.source.strip(),
280			align=self.filename_alignment,
281			code=self.code_colors[event.kind],
282			**self.event_colors
283			))
284
285
286			class VarsPrinter(Fields.names.globals.stream.filename_alignment, ColorStreamAction):
287			"""
288			An action that prints local variables and optionally global variables visible from the current executing frame.
289
290			Args:
291			*names (strings): Names to evaluate. Expressions can be used (will only try to evaluate if all the variables are
292			present on the frame.
293			globals (bool): Allow access to globals. Default: ``False`` (only looks at locals).
294			stream (file-like): Stream to write to. Default: ``sys.stderr``.
295			filename_alignment (int): Default size for the filename column (files are right-aligned). Default: ``40``.
296			force_colors (bool): Force coloring. Default: ``False``.
297			repr_limit (bool): Limit length of ``repr()`` output. Default: ``512``.
298
299			.. note::
300			This is the default action.
301
302			.. warning::
303
304			In `hunter 2.0` the default action will be :obj:`hunter.CallPrinter`.
305			"""
306
307			def __init__(self, names, *options):
308			if not names:
309			raise TypeError("VarsPrinter requires at least one variable name/expression.")
310			self.names = {
311			name: set(self._iter_symbols(name))
312			for name in names
313			}
314			self.globals = options.pop('globals', False)
315			super(VarsPrinter, self).__init__(**options)
316
317			@staticmethod
318			def _iter_symbols(code):
319			"""
320			Iterate all the variable names in the given expression.
321
322			Example:
323
324			* ``self.foobar`` yields ``self``
325			* ``self[foobar]`` yields `self`` and ``foobar``
326			"""
327			for node in ast.walk(ast.parse(code)):
328			if isinstance(node, ast.Name):
329			yield node.id
330
331			def _safe_eval(self, code, event):
332			"""
333			Try to evaluate the given code on the given frame. If failure occurs, returns some ugly string with exception.
334			"""
335			try:
336			return eval(code, event.globals if self.globals else {}, event.locals)
337			except Exception as exc:
338			return "{internal-failure}FAILED EVAL: {internal-detail}{!r}".format(exc, **self.event_colors)
339
340			def __call__(self, event):
341			"""
342			Handle event and print the specified variables.
343			"""
344			first = True
345			frame_symbols = set(event.locals)
346			if self.globals:
347			frame_symbols \|= set(event.globals)
348
349			for code, symbols in self.names.items():
350			try:
351			obj = eval(code, event.globals if self.globals else {}, event.locals)
352			except AttributeError:
353			continue
354			except Exception as exc:
355			printout = "{internal-failure}FAILED EVAL: {internal-detail}{!r}".format(exc, **self.event_colors)
356			else:
357			printout = self._safe_repr(obj)
358
359			if frame_symbols >= symbols:
360			self.stream.write("{:>{align}} {vars}{:9} {vars-name}{} {vars}=> {reset}{}{reset}\n".format(
361			"",
362			"vars" if first else "...",
363			code,
364			printout,
365			align=self.filename_alignment,
366			**self.event_colors
367			))
368			first = False
369

ionelmc / python-hunter

Push — master ( 44d803...8bc4ee )

src.hunter.ColorStreamAction.stream() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like