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

Completed

Push — master ( bb545f...077227 )

by Ionel Cristian

created 2016-01-08 04:28 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	3
rs	10
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 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 filename 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): Names to evaluate. Expressions can be used (will only try to evaluate if all the variables are
199			present on the frame.
200			stream (file-like): Stream to write to. Default: ``sys.stderr``.
201			filename_alignment (int): Default size for the filename column (files are right-aligned). Default: ``40``.
202			globals (bool): Allow access to globals. Default: ``False`` (only looks at locals).
203			"""
204
205			def __init__(self, names, *options):
206			if not names:
207			raise TypeError("Must give at least one name/expression.")
208			self.stream = options.pop('stream', self.default_stream)
209			self.force_colors = options.pop('force_colors', False)
210			self.filename_alignment = max(5, options.pop('filename_alignment', DEFAULT_MIN_FILENAME_ALIGNMENT))
211			self.names = {
212			name: set(self._iter_symbols(name))
213			for name in names
214			}
215			self.globals = options.pop('globals', False)
216
217			@staticmethod
218			def _iter_symbols(code):
219			"""
220			Iterate all the variable names in the given expression.
221
222			Example:
223
224			* ``self.foobar`` yields ``self``
225			* ``self[foobar]`` yields `self`` and ``foobar``
226			"""
227			for node in ast.walk(ast.parse(code)):
228			if isinstance(node, ast.Name):
229			yield node.id
230
231			def _safe_eval(self, code, event):
232			"""
233			Try to evaluate the given code on the given frame. If failure occurs, returns some ugly string with exception.
234			"""
235			try:
236			return eval(code, event.globals if self.globals else {}, event.locals)
237			except Exception as exc:
238			return "{internal-failure}FAILED EVAL: {internal-detail}{!r}".format(exc, **self.event_colors)
239
240			def __call__(self, event):
241			"""
242			Handle event and print the specified variables.
243			"""
244			first = True
245			frame_symbols = set(event.locals)
246			if self.globals:
247			frame_symbols \|= set(event.globals)
248
249			for code, symbols in self.names.items():
250			try:
251			obj = eval(code, event.globals if self.globals else {}, event.locals)
252			except AttributeError:
253			continue
254			except Exception as exc:
255			printout = "{internal-failure}FAILED EVAL: {internal-detail}{!r}".format(exc, **self.event_colors)
256			else:
257			printout = self._safe_repr(obj)
258
259			if frame_symbols >= symbols:
260			self.stream.write("{:>{align}} {vars}{:9} {vars-name}{} {vars}=> {reset}{}{reset}\n".format(
261			"",
262			"vars" if first else "...",
263			code,
264			printout,
265			align=self.filename_alignment,
266			**self.event_colors
267			))
268			first = False
269

ionelmc / python-hunter

Push — master ( bb545f...077227 )

src.hunter.ColorStreamAction.stream() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like