src.hunter.ColorStreamAction.stream() - Code Metrics - Inspection of "Tweak a bit the new perf test." - ionelmc/python-hunter - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 67ac21...36843e )

by Ionel Cristian

created 2015-12-23 01:54 UTC

src.hunter.ColorStreamAction.stream() C

↳ Parent: src.hunter.ColorStreamAction

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	7
dl	0
loc	3
rs	5.5

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 fields import Fields
from six import string_types

DEFAULT_MIN_FILENAME_ALIGNMENT = 40
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,
    'filename': '',
    'colon': Fore.BLACK + Style.BRIGHT,
    'lineno': Style.RESET_ALL,
    'kind': Fore.CYAN,
    'continuation': 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': Back.RED + Style.BRIGHT + 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,
}


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(Action):
    _stream_cache = {}
    _stream = None
    _tty = None
    default_stream = sys.stderr
    force_colors = False

    @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)
            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):
        try:
            return repr(obj)
        except Exception as exc:
            return "{internal-failure}!!! FAILED REPR: {internal-detail}{!r}".format(exc, **self.event_colors)


class CodePrinter(Fields.stream.filename_alignment, 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``.
    """

    def __init__(self,
                 stream=ColorStreamAction.default_stream, force_colors=False,
                 filename_alignment=DEFAULT_MIN_FILENAME_ALIGNMENT):
        self.stream = stream
        self.force_colors = force_colors
        self.filename_alignment = max(5, filename_alignment)

    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 __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 = event.filename or "<???>"
        if len(filename) > self.filename_alignment:
            filename = '[...]{}'.format(filename[5 - self.filename_alignment:])

        # 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(
            filename,
            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 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.
        stream (file-like): Stream to write to. Default: ``sys.stderr``.
        filename_alignment (int): Default size for the filaneme column (files are right-aligned). Default: ``40``.
        globals (bool): Allow access to globals. Default: ``False`` (only looks at locals).
    """

    def __init__(self, *names, **options):
        if not names:
            raise TypeError("Must give at least one name/expression.")
        self.stream = options.pop('stream', self.default_stream)
        self.force_colors = options.pop('force_colors', False)
        self.filename_alignment = max(5, options.pop('filename_alignment', DEFAULT_MIN_FILENAME_ALIGNMENT))
        self.names = {
            name: set(self._iter_symbols(name))
            for name in names
            }
        self.globals = options.pop('globals', False)

    @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 fields import Fields
13			from six import string_types
14
15			DEFAULT_MIN_FILENAME_ALIGNMENT = 40
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			'filename': '',
38			'colon': Fore.BLACK + Style.BRIGHT,
39			'lineno': Style.RESET_ALL,
40			'kind': Fore.CYAN,
41			'continuation': Fore.BLUE,
42			'return': Style.BRIGHT + Fore.GREEN,
43			'exception': Style.BRIGHT + Fore.RED,
44			'detail': Style.NORMAL,
45			'vars': Style.RESET_ALL + Fore.MAGENTA,
46			'vars-name': Style.BRIGHT,
47			'internal-failure': Back.RED + Style.BRIGHT + Fore.RED,
48			'internal-detail': Fore.WHITE,
49			'source-failure': Style.BRIGHT + Back.YELLOW + Fore.YELLOW,
50			'source-detail': Fore.WHITE,
51			}
52			CODE_COLORS = {
53			'call': Fore.RESET + Style.BRIGHT,
54			'line': Fore.RESET,
55			'return': Fore.YELLOW,
56			'exception': Fore.RED,
57			}
58
59
60			class Action(object):
61			def __call__(self, event):
62			raise NotImplementedError()
63
64
65			class Debugger(Fields.klass.kwargs, Action):
66			"""
67			An action that starts ``pdb``.
68			"""
69
70			def __init__(self, klass=pdb.Pdb, **kwargs):
71			self.klass = klass
72			self.kwargs = kwargs
73
74			def __call__(self, event):
75			"""
76			Runs a ``pdb.set_trace`` at the matching frame.
77			"""
78			self.klass(**self.kwargs).set_trace(event.frame)
79
80
81			class ColorStreamAction(Action):
82			_stream_cache = {}
83			_stream = None
84			_tty = None
85			default_stream = sys.stderr
86			force_colors = False
87
88			@property
89			def stream(self):
90			return self._stream
91
92			@stream.setter
93			def stream(self, value):
94			if isinstance(value, string_types):
95			if value in self._stream_cache:
96			value = self._stream_cache[value]
97			else:
98			value = self._stream_cache[value] = open(value, 'a', buffering=0)
99
100			isatty = getattr(value, 'isatty', None)
101			if self.force_colors or (isatty and isatty() and os.name != 'java'):
102			self._stream = AnsiToWin32(value)
103			self._tty = True
104			self.event_colors = EVENT_COLORS
105			self.code_colors = CODE_COLORS
106			else:
107			self._tty = False
108			self._stream = value
109			self.event_colors = NO_COLORS
110			self.code_colors = NO_COLORS
111
112			def _safe_repr(self, obj):
113			try:
114			return repr(obj)
115			except Exception as exc:
116			return "{internal-failure}!!! FAILED REPR: {internal-detail}{!r}".format(exc, **self.event_colors)
117
118
119			class CodePrinter(Fields.stream.filename_alignment, ColorStreamAction):
120			"""
121			An action that just prints the code being executed.
122
123			Args:
124			stream (file-like): Stream to write to. Default: ``sys.stderr``.
125			filename_alignment (int): Default size for the filename column (files are right-aligned). Default: ``40``.
126			"""
127
128			def __init__(self,
129			stream=ColorStreamAction.default_stream, force_colors=False,
130			filename_alignment=DEFAULT_MIN_FILENAME_ALIGNMENT):
131			self.stream = stream
132			self.force_colors = force_colors
133			self.filename_alignment = max(5, filename_alignment)
134
135			def _safe_source(self, event):
136			try:
137			lines = event._raw_fullsource.rstrip().splitlines()
138			if lines:
139			return lines
140			else:
141			return "{source-failure}??? NO SOURCE: {source-detail}" \
142			"Source code string for module {!r} is empty.".format(event.module, **self.event_colors),
143			return lines
144			except Exception as exc:
145			return "{source-failure}??? NO SOURCE: {source-detail}{!r}".format(exc, **self.event_colors),
146
147			def __call__(self, event, sep=os.path.sep, join=os.path.join):
148			"""
149			Handle event and print filename, line number and source code. If event.kind is a `return` or `exception` also
150			prints values.
151			"""
152			filename = event.filename or "<???>"
153			if len(filename) > self.filename_alignment:
154			filename = '[...]{}'.format(filename[5 - self.filename_alignment:])
155
156			# context = event.tracer
157			# alignment = context.filename_alignment = max(
158			# getattr(context, 'filename_alignment', 5),
159			# len(filename)
160			# )
161			lines = self._safe_source(event)
162			self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {code}{}{reset}\n".format(
163			filename,
164			event.lineno,
165			event.kind,
166			lines[0],
167			align=self.filename_alignment,
168			code=self.code_colors[event.kind],
169			**self.event_colors
170			))
171			for line in lines[1:]:
172			self.stream.write("{:>{align}} {kind}{:9} {code}{}{reset}\n".format(
173			"",
174			r" \|",
175			line,
176			align=self.filename_alignment,
177			code=self.code_colors[event.kind],
178			**self.event_colors
179			))
180
181			if event.kind in ('return', 'exception'):
182			self.stream.write("{:>{align}} {continuation}{:9} {color}{} value: {detail}{}{reset}\n".format(
183			"",
184			"...",
185			event.kind,
186			self._safe_repr(event.arg),
187			align=self.filename_alignment,
188			color=self.event_colors[event.kind],
189			**self.event_colors
190			))
191
192
193			class VarsPrinter(Fields.names.globals.stream.filename_alignment, ColorStreamAction):
194			"""
195			An action that prints local variables and optionally global variables visible from the current executing frame.
196
197			Args:
198			*names (strings):
199			Names to evaluate. Expressions can be used (will only try to evaluate if all the variables are present on
200			the frame.
201			stream (file-like): Stream to write to. Default: ``sys.stderr``.
202			filename_alignment (int): Default size for the filaneme column (files are right-aligned). Default: ``40``.
203			globals (bool): Allow access to globals. Default: ``False`` (only looks at locals).
204			"""
205
206			def __init__(self, names, *options):
207			if not names:
208			raise TypeError("Must give at least one name/expression.")
209			self.stream = options.pop('stream', self.default_stream)
210			self.force_colors = options.pop('force_colors', False)
211			self.filename_alignment = max(5, options.pop('filename_alignment', DEFAULT_MIN_FILENAME_ALIGNMENT))
212			self.names = {
213			name: set(self._iter_symbols(name))
214			for name in names
215			}
216			self.globals = options.pop('globals', False)
217
218			@staticmethod
219			def _iter_symbols(code):
220			"""
221			Iterate all the variable names in the given expression.
222
223			Example:
224
225			* ``self.foobar`` yields ``self``
226			* ``self[foobar]`` yields `self`` and ``foobar``
227			"""
228			for node in ast.walk(ast.parse(code)):
229			if isinstance(node, ast.Name):
230			yield node.id
231
232			def _safe_eval(self, code, event):
233			"""
234			Try to evaluate the given code on the given frame. If failure occurs, returns some ugly string with exception.
235			"""
236			try:
237			return eval(code, event.globals if self.globals else {}, event.locals)
238			except Exception as exc:
239			return "{internal-failure}FAILED EVAL: {internal-detail}{!r}".format(exc, **self.event_colors)
240
241			def __call__(self, event):
242			"""
243			Handle event and print the specified variables.
244			"""
245			first = True
246			frame_symbols = set(event.locals)
247			if self.globals:
248			frame_symbols \|= set(event.globals)
249
250			for code, symbols in self.names.items():
251			try:
252			obj = eval(code, event.globals if self.globals else {}, event.locals)
253			except AttributeError:
254			continue
255			except Exception as exc:
256			printout = "{internal-failure}FAILED EVAL: {internal-detail}{!r}".format(exc, **self.event_colors)
257			else:
258			printout = self._safe_repr(obj)
259
260			if frame_symbols >= symbols:
261			self.stream.write("{:>{align}} {vars}{:9} {vars-name}{} {vars}=> {reset}{}{reset}\n".format(
262			"",
263			"vars" if first else "...",
264			code,
265			printout,
266			align=self.filename_alignment,
267			**self.event_colors
268			))
269			first = False
270

ionelmc / python-hunter

Push — master ( 67ac21...36843e )

src.hunter.ColorStreamAction.stream() C

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like