src.hunter.ColorStreamAction.stream() - Code Metrics - Inspection of "Fix a bit the classes (we only need printables)." - ionelmc/python-hunter - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( f77446...965385 )

by Ionel Cristian

created 2016-01-22 23:57 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 PrintableMixin
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,
    '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(PrintableMixin.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(PrintableMixin.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=DEFAULT_MIN_FILENAME_ALIGNMENT,
                 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.
    """
    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.

    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.
    """

    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(PrintableMixin.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).
        force_colors (bool): Force coloring.
    """

    def __init__(self, *names, **options):
        if not names:
            raise TypeError("VarsPrinter requires at least one variable name/expression.")
        super(VarsPrinter, self).__init__(**options)
        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 PrintableMixin
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			'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(PrintableMixin.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(PrintableMixin.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=DEFAULT_MIN_FILENAME_ALIGNMENT,
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.
146			"""
147			def _safe_source(self, event):
148			try:
149			lines = event._raw_fullsource.rstrip().splitlines()
150			if lines:
151			return lines
152			else:
153			return "{source-failure}??? NO SOURCE: {source-detail}" \
154			"Source code string for module {!r} is empty.".format(event.module, **self.event_colors),
155			return lines
156			except Exception as exc:
157			return "{source-failure}??? NO SOURCE: {source-detail}{!r}".format(exc, **self.event_colors),
158
159			def _format_filename(self, event):
160			filename = event.filename or "<???>"
161			if len(filename) > self.filename_alignment:
162			filename = '[...]{}'.format(filename[5 - self.filename_alignment:])
163			return filename
164
165			def __call__(self, event, sep=os.path.sep, join=os.path.join):
166			"""
167			Handle event and print filename, line number and source code. If event.kind is a `return` or `exception` also
168			prints values.
169			"""
170
171			# context = event.tracer
172			# alignment = context.filename_alignment = max(
173			# getattr(context, 'filename_alignment', 5),
174			# len(filename)
175			# )
176			lines = self._safe_source(event)
177			self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {code}{}{reset}\n".format(
178			self._format_filename(event),
179			event.lineno,
180			event.kind,
181			lines[0],
182			align=self.filename_alignment,
183			code=self.code_colors[event.kind],
184			**self.event_colors
185			))
186			for line in lines[1:]:
187			self.stream.write("{:>{align}} {kind}{:9} {code}{}{reset}\n".format(
188			"",
189			r" \|",
190			line,
191			align=self.filename_alignment,
192			code=self.code_colors[event.kind],
193			**self.event_colors
194			))
195
196			if event.kind in ('return', 'exception'):
197			self.stream.write("{:>{align}} {continuation}{:9} {color}{} value: {detail}{}{reset}\n".format(
198			"",
199			"...",
200			event.kind,
201			self._safe_repr(event.arg),
202			align=self.filename_alignment,
203			color=self.event_colors[event.kind],
204			**self.event_colors
205			))
206
207
208			class CallPrinter(CodePrinter):
209			"""
210			An action that just prints the code being executed.
211
212			Args:
213			stream (file-like): Stream to write to. Default: ``sys.stderr``.
214			filename_alignment (int): Default size for the filename column (files are right-aligned). Default: ``40``.
215			force_colors (bool): Force coloring.
216			"""
217
218			def __init__(self, **options):
219			super(CallPrinter, self).__init__(**options)
220			self.stack = []
221
222			def __call__(self, event, sep=os.path.sep, join=os.path.join):
223			"""
224			Handle event and print filename, line number and source code. If event.kind is a `return` or `exception` also
225			prints values.
226			"""
227			filename = self._format_filename(event)
228			ident = event.module, event.function
229
230			if event.kind == 'call':
231			code = event.code
232			self.stack.append(ident)
233			self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {}{call}=>{normal} {}({}{call}{normal}){reset}\n".format(
234			filename,
235			event.lineno,
236			event.kind,
237			' ' * (len(self.stack) - 1),
238			event.function,
239			', '.join('{vars}{vars-name}{0}{vars}={reset}{1}'.format(
240			var,
241			self._safe_repr(event.locals.get(var, MISSING)),
242			**self.event_colors
243			) for var in code.co_varnames[:code.co_argcount]),
244			align=self.filename_alignment,
245			**self.event_colors
246			))
247			elif event.kind in ('return', 'exception'):
248			self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {code}{}{}{normal} {}: {reset}{}\n".format(
249			filename,
250			event.lineno,
251			event.kind,
252			' ' * (len(self.stack) - 1),
253			{'return': '<=', 'exception': '<!'}[event.kind],
254			event.function,
255			self._safe_repr(event.arg),
256			align=self.filename_alignment,
257			code=self.event_colors[event.kind],
258			**self.event_colors
259			))
260			if self.stack and self.stack[-1] == ident:
261			self.stack.pop()
262			else:
263			self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {reset}{}{}\n".format(
264			filename,
265			event.lineno,
266			event.kind,
267			' ' * len(self.stack),
268			event.source.strip(),
269			align=self.filename_alignment,
270			code=self.code_colors[event.kind],
271			**self.event_colors
272			))
273
274
275			class VarsPrinter(PrintableMixin.names.globals.stream.filename_alignment, ColorStreamAction):
276			"""
277			An action that prints local variables and optionally global variables visible from the current executing frame.
278
279			Args:
280			*names (strings): Names to evaluate. Expressions can be used (will only try to evaluate if all the variables are
281			present on the frame.
282			stream (file-like): Stream to write to. Default: ``sys.stderr``.
283			filename_alignment (int): Default size for the filename column (files are right-aligned). Default: ``40``.
284			globals (bool): Allow access to globals. Default: ``False`` (only looks at locals).
285			force_colors (bool): Force coloring.
286			"""
287
288			def __init__(self, names, *options):
289			if not names:
290			raise TypeError("VarsPrinter requires at least one variable name/expression.")
291			super(VarsPrinter, self).__init__(**options)
292			self.names = {
293			name: set(self._iter_symbols(name))
294			for name in names
295			}
296			self.globals = options.pop('globals', False)
297
298			@staticmethod
299			def _iter_symbols(code):
300			"""
301			Iterate all the variable names in the given expression.
302
303			Example:
304
305			* ``self.foobar`` yields ``self``
306			* ``self[foobar]`` yields `self`` and ``foobar``
307			"""
308			for node in ast.walk(ast.parse(code)):
309			if isinstance(node, ast.Name):
310			yield node.id
311
312			def _safe_eval(self, code, event):
313			"""
314			Try to evaluate the given code on the given frame. If failure occurs, returns some ugly string with exception.
315			"""
316			try:
317			return eval(code, event.globals if self.globals else {}, event.locals)
318			except Exception as exc:
319			return "{internal-failure}FAILED EVAL: {internal-detail}{!r}".format(exc, **self.event_colors)
320
321			def __call__(self, event):
322			"""
323			Handle event and print the specified variables.
324			"""
325			first = True
326			frame_symbols = set(event.locals)
327			if self.globals:
328			frame_symbols \|= set(event.globals)
329
330			for code, symbols in self.names.items():
331			try:
332			obj = eval(code, event.globals if self.globals else {}, event.locals)
333			except AttributeError:
334			continue
335			except Exception as exc:
336			printout = "{internal-failure}FAILED EVAL: {internal-detail}{!r}".format(exc, **self.event_colors)
337			else:
338			printout = self._safe_repr(obj)
339
340			if frame_symbols >= symbols:
341			self.stream.write("{:>{align}} {vars}{:9} {vars-name}{} {vars}=> {reset}{}{reset}\n".format(
342			"",
343			"vars" if first else "...",
344			code,
345			printout,
346			align=self.filename_alignment,
347			**self.event_colors
348			))
349			first = False
350

ionelmc / python-hunter

Push — master ( f77446...965385 )

src.hunter.ColorStreamAction.stream() C

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like