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

Completed

Push — master ( 4e3b27...2dc96c )

by Ionel Cristian

created 2015-12-20 14:53 UTC

src.hunter.ColorStreamAction.stream() B

↳ Parent: src.hunter.ColorStreamAction

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	7
dl	0
loc	19
rs	7.3333

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 not lines:
                raise RuntimeError("Source code string is empty.")
            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 not lines:
139			raise RuntimeError("Source code string is empty.")
140			return lines
141			except Exception as exc:
142			return "{source-failure}??? NO SOURCE: {source-detail}{!r}".format(exc, **self.event_colors),
143
144			def __call__(self, event, sep=os.path.sep, join=os.path.join):
145			"""
146			Handle event and print filename, line number and source code. If event.kind is a `return` or `exception` also
147			prints values.
148			"""
149			filename = event.filename or "<???>"
150			if len(filename) > self.filename_alignment:
151			filename = '[...]{}'.format(filename[5 - self.filename_alignment:])
152
153			# context = event.tracer
154			# alignment = context.filename_alignment = max(
155			# getattr(context, 'filename_alignment', 5),
156			# len(filename)
157			# )
158			lines = self._safe_source(event)
159			self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {code}{}{reset}\n".format(
160			filename,
161			event.lineno,
162			event.kind,
163			lines[0],
164			align=self.filename_alignment,
165			code=self.code_colors[event.kind],
166			**self.event_colors
167			))
168			for line in lines[1:]:
169			self.stream.write("{:>{align}} {kind}{:9} {code}{}{reset}\n".format(
170			"",
171			r" \|",
172			line,
173			align=self.filename_alignment,
174			code=self.code_colors[event.kind],
175			**self.event_colors
176			))
177
178			if event.kind in ('return', 'exception'):
179			self.stream.write("{:>{align}} {continuation}{:9} {color}{} value: {detail}{}{reset}\n".format(
180			"",
181			"...",
182			event.kind,
183			self._safe_repr(event.arg),
184			align=self.filename_alignment,
185			color=self.event_colors[event.kind],
186			**self.event_colors
187			))
188
189
190			class VarsPrinter(Fields.names.globals.stream.filename_alignment, ColorStreamAction):
191			"""
192			An action that prints local variables and optionally global variables visible from the current executing frame.
193
194			Args:
195			*names (strings):
196			Names to evaluate. Expressions can be used (will only try to evaluate if all the variables are present on
197			the frame.
198			stream (file-like): Stream to write to. Default: ``sys.stderr``.
199			filename_alignment (int): Default size for the filaneme column (files are right-aligned). Default: ``40``.
200			globals (bool): Allow access to globals. Default: ``False`` (only looks at locals).
201			"""
202
203			def __init__(self, names, *options):
204			if not names:
205			raise TypeError("Must give at least one name/expression.")
206			self.stream = options.pop('stream', self.default_stream)
207			self.force_colors = options.pop('force_colors', False)
208			self.filename_alignment = max(5, options.pop('filename_alignment', DEFAULT_MIN_FILENAME_ALIGNMENT))
209			self.names = {
210			name: set(self._iter_symbols(name))
211			for name in names
212			}
213			self.globals = options.pop('globals', False)
214
215			@staticmethod
216			def _iter_symbols(code):
217			"""
218			Iterate all the variable names in the given expression.
219
220			Example:
221
222			* ``self.foobar`` yields ``self``
223			* ``self[foobar]`` yields `self`` and ``foobar``
224			"""
225			for node in ast.walk(ast.parse(code)):
226			if isinstance(node, ast.Name):
227			yield node.id
228
229			def _safe_eval(self, code, event):
230			"""
231			Try to evaluate the given code on the given frame. If failure occurs, returns some ugly string with exception.
232			"""
233			try:
234			return eval(code, event.globals if self.globals else {}, event.locals)
235			except Exception as exc:
236			return "{internal-failure}FAILED EVAL: {internal-detail}{!r}".format(exc, **self.event_colors)
237
238			def __call__(self, event):
239			"""
240			Handle event and print the specified variables.
241			"""
242			first = True
243			frame_symbols = set(event.locals)
244			if self.globals:
245			frame_symbols \|= set(event.globals)
246
247			for code, symbols in self.names.items():
248			try:
249			obj = eval(code, event.globals if self.globals else {}, event.locals)
250			except AttributeError:
251			continue
252			except Exception as exc:
253			printout = "{internal-failure}FAILED EVAL: {internal-detail}{!r}".format(exc, **self.event_colors)
254			else:
255			printout = self._safe_repr(obj)
256
257			if frame_symbols >= symbols:
258			self.stream.write("{:>{align}} {vars}{:9} {vars-name}{} {vars}=> {reset}{}{reset}\n".format(
259			"",
260			"vars" if first else "...",
261			code,
262			printout,
263			align=self.filename_alignment,
264			**self.event_colors
265			))
266			first = False
267

ionelmc / python-hunter

Push — master ( 4e3b27...2dc96c )

src.hunter.ColorStreamAction.stream() B

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like