src.pytest_benchmark.parse_columns() - Code Metrics - Inspection of "add an option to select which columns to display" - ionelmc/pytest-benchmark - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#34)

unknown

created 2015-11-27 16:22 UTC

src.pytest_benchmark.parse_columns() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	3
dl	0
loc	11
rs	9.4286

from __future__ import division
from __future__ import print_function

import argparse
import json
import os
import platform
import re
import subprocess
import sys
import types
from datetime import datetime
from decimal import Decimal

from .compat import PY3

try:
    from subprocess import check_output
except ImportError:
    def check_output(*popenargs, **kwargs):
        if 'stdout' in kwargs:
            raise ValueError('stdout argument not allowed, it will be overridden.')
        process = subprocess.Popen(stdout=subprocess.PIPE, *popenargs, **kwargs)
        output, unused_err = process.communicate()
        retcode = process.poll()
        if retcode:
            cmd = kwargs.get("args")
            if cmd is None:
                cmd = popenargs[0]
            raise subprocess.CalledProcessError(retcode, cmd)
        return output


class SecondsDecimal(Decimal):
    def __float__(self):
        return float(super(SecondsDecimal, self).__str__())

    def __str__(self):
        return "{0}s".format(format_time(float(super(SecondsDecimal, self).__str__())))

    @property
    def as_string(self):
        return super(SecondsDecimal, self).__str__()


class NameWrapper(object):
    def __init__(self, target):
        self.target = target

    def __str__(self):
        name = self.target.__module__ + "." if hasattr(self.target, '__module__') else ""
        name += self.target.__name__ if hasattr(self.target, '__name__') else repr(self.target)
        return name

    def __repr__(self):
        return "NameWrapper(%s)" % repr(self.target)


def get_tag():
    info = get_commit_info()
    return '%s_%s%s' % (info['id'], get_current_time(), '_uncommitted-changes' if info['dirty'] else '')


def get_commit_info():
    dirty = False
    commit = 'unversioned'
    try:
        if os.path.exists('.git'):
            desc = check_output('git describe --dirty --always --long --abbrev=40'.split(),
                                universal_newlines=True).strip()
            desc = desc.split('-')
            if desc[-1].strip() == 'dirty':
                dirty = True
                desc.pop()
            commit = desc[-1].strip('g')
        elif os.path.exists('.hg'):
            desc = check_output('hg id --id --debug'.split(), universal_newlines=True).strip()
            if desc[-1] == '+':
                dirty = True
            commit = desc.strip('+')
        return {
            'id': commit,
            'dirty': dirty
        }
    except Exception as exc:
        return {
            'id': 'unknown',
            'dirty': dirty,
            'error': repr(exc),
        }


def get_current_time():
    return datetime.now().strftime("%Y%m%d_%H%M%S")


def first_or_value(obj, value):
    if obj:
        value, = obj

    return value


def load_timer(string):
    if "." not in string:
        raise argparse.ArgumentTypeError("Value for --benchmark-timer must be in dotted form. Eg: 'module.attr'.")
    mod, attr = string.rsplit(".", 1)
    if mod == 'pep418':
        if PY3:
            import time
            return NameWrapper(getattr(time, attr))
        else:
            from . import pep418
            return NameWrapper(getattr(pep418, attr))
    else:
        __import__(mod)
        mod = sys.modules[mod]
        return NameWrapper(getattr(mod, attr))


class RegressionCheck(object):
    def __init__(self, field, threshold):
        self.field = field
        self.threshold = threshold

    def fails(self, current, compared):
        val = self.compute(current, compared)
        if val > self.threshold:
            return "Field %s has failed %s: %.9f > %.9f" % (
                self.field, self.__class__.__name__, val, self.threshold
            )


class PercentageRegressionCheck(RegressionCheck):
    def compute(self, current, compared):
        val = compared[self.field]
        if not val:
            return float("inf")
        return current[self.field] / val * 100 - 100


class DifferenceRegressionCheck(RegressionCheck):
    def compute(self, current, compared):
        return current[self.field] - compared[self.field]


def parse_compare_fail(string,
                       rex=re.compile('^(?P<field>min|max|mean|median|stddev|iqr):'
                                      '((?P<percentage>[0-9]?[0-9])%|(?P<difference>[0-9]*\.?[0-9]+([eE][-+]?[0-9]+)?))$')):
    m = rex.match(string)
    if m:
        g = m.groupdict()
        if g['percentage']:
            return PercentageRegressionCheck(g['field'], int(g['percentage']))
        elif g['difference']:
            return DifferenceRegressionCheck(g['field'], float(g['difference']))

    raise argparse.ArgumentTypeError("Could not parse value: %r." % string)


def parse_warmup(string):
    string = string.lower().strip()
    if string == "auto":
        return platform.python_implementation() == "PyPy"
    elif string in ["off", "false", "no"]:
        return False
    elif string in ["on", "true", "yes", ""]:
        return True
    else:
        raise argparse.ArgumentTypeError("Could not parse value: %r." % string)


def parse_timer(string):
    return str(load_timer(string))


def parse_sort(string):
    string = string.lower().strip()
    if string not in ("min", "max", "mean", "stddev"):
        raise argparse.ArgumentTypeError(
            "Unacceptable value: %r. "
            "Value for --benchmark-sort must be one of: 'min', 'max', 'mean' or 'stddev'." % string)
    return string


def parse_columns(string):
    allowed_columns = ["min", "max", "mean", "stddev", "median", "iqr",
                       "outliers", "rounds", "iterations"]
    columns = [str.strip(s) for s in string.lower().split(',')]
    invalid = set(columns) - set(allowed_columns)
    if invalid:
        # there are extra items in columns!
        msg = "Invalid column name(s): %s. " % ', '.join(invalid)
        msg += "The only valid column names are: %s" % ', '.join(allowed_columns)
        raise argparse.ArgumentTypeError(msg)
    return columns


def parse_rounds(string):
    try:
        value = int(string)
    except ValueError as exc:
        raise argparse.ArgumentTypeError(exc)
    else:
        if value < 1:
            raise argparse.ArgumentTypeError("Value for --benchmark-rounds must be at least 1.")
        return value


def parse_seconds(string):
    try:
        return SecondsDecimal(string).as_string
    except Exception as exc:
        raise argparse.ArgumentTypeError("Invalid decimal value %r: %r" % (string, exc))


def parse_save(string):
    if not string:
        raise argparse.ArgumentTypeError("Can't be empty.")
    illegal = ''.join(c for c in r"\/:*?<>|" if c in string)
    if illegal:
        raise argparse.ArgumentTypeError("Must not contain any of these characters: /:*?<>|\\ (it has %r)" % illegal)
    return string


def time_unit(value):
    if value < 1e-6:
        return "n", 1e9
    elif value < 1e-3:
        return "u", 1e6
    elif value < 1:
        return "m", 1e3
    else:
        return "", 1.


def format_time(value):
    unit, adjustment = time_unit(value)
    return "{0:.2f}{1:s}".format(value * adjustment, unit)


class cached_property(object):
    def __init__(self, func):
        self.__doc__ = getattr(func, '__doc__')
        self.func = func

    def __get__(self, obj, cls):
        if obj is None:
            return self
        value = obj.__dict__[self.func.__name__] = self.func(obj)
        return value


def clonefunc(f):
    """Deep clone the given function to create a new one.

    By default, the PyPy JIT specializes the assembler based on f.__code__:
    clonefunc makes sure that you will get a new function with a **different**
    __code__, so that PyPy will produce independent assembler. This is useful
    e.g. for benchmarks and microbenchmarks, so you can make sure to compare
    apples to apples.

    Use it with caution: if abused, this might easily produce an explosion of
    produced assembler.

    from: https://bitbucket.org/antocuni/pypytools/src/tip/pypytools/util.py?at=default
    """

    # first of all, we clone the code object
    try:
        co = f.__code__
        if PY3:
            co2 = types.CodeType(co.co_argcount, co.co_kwonlyargcount,
                                 co.co_nlocals, co.co_stacksize, co.co_flags, co.co_code,
                                 co.co_consts, co.co_names, co.co_varnames, co.co_filename, co.co_name,
                                 co.co_firstlineno, co.co_lnotab, co.co_freevars, co.co_cellvars)
        else:
            co2 = types.CodeType(co.co_argcount, co.co_nlocals, co.co_stacksize, co.co_flags, co.co_code,
                                 co.co_consts, co.co_names, co.co_varnames, co.co_filename, co.co_name,
                                 co.co_firstlineno, co.co_lnotab, co.co_freevars, co.co_cellvars)
        #
        # then, we clone the function itself, using the new co2
        return types.FunctionType(co2, f.__globals__, f.__name__, f.__defaults__, f.__closure__)
    except AttributeError:
        return f


def format_dict(obj):
    return "{%s}" % ", ".join("%s: %s" % (k, json.dumps(v)) for k, v in sorted(obj.items()))


def report_progress(iterable, terminal_reporter, format_string, **kwargs):
    total = len(iterable)

    def progress_reporting_wrapper():
        for pos, item in enumerate(iterable):
            string = format_string.format(pos=pos + 1, total=total, value=item, **kwargs)
            terminal_reporter.rewrite(string, black=True, bold=True)
            yield string, item
    return progress_reporting_wrapper()


1			from __future__ import division
2			from __future__ import print_function
3
4			import argparse
5			import json
6			import os
7			import platform
8			import re
9			import subprocess
10			import sys
11			import types
12			from datetime import datetime
13			from decimal import Decimal
14
15			from .compat import PY3
16
17			try:
18			from subprocess import check_output
19			except ImportError:
20			def check_output(popenargs, *kwargs):
21			if 'stdout' in kwargs:
22			raise ValueError('stdout argument not allowed, it will be overridden.')
23			process = subprocess.Popen(stdout=subprocess.PIPE, popenargs, *kwargs)
24			output, unused_err = process.communicate()
25			retcode = process.poll()
26			if retcode:
27			cmd = kwargs.get("args")
28			if cmd is None:
29			cmd = popenargs[0]
30			raise subprocess.CalledProcessError(retcode, cmd)
31			return output
32
33
34			class SecondsDecimal(Decimal):
35			def __float__(self):
36			return float(super(SecondsDecimal, self).__str__())
37
38			def __str__(self):
39			return "{0}s".format(format_time(float(super(SecondsDecimal, self).__str__())))
40
41			@property
42			def as_string(self):
43			return super(SecondsDecimal, self).__str__()
44
45
46			class NameWrapper(object):
47			def __init__(self, target):
48			self.target = target
49
50			def __str__(self):
51			name = self.target.__module__ + "." if hasattr(self.target, '__module__') else ""
52			name += self.target.__name__ if hasattr(self.target, '__name__') else repr(self.target)
53			return name
54
55			def __repr__(self):
56			return "NameWrapper(%s)" % repr(self.target)
57
58
59			def get_tag():
60			info = get_commit_info()
61			return '%s_%s%s' % (info['id'], get_current_time(), '_uncommitted-changes' if info['dirty'] else '')
62
63
64			def get_commit_info():
65			dirty = False
66			commit = 'unversioned'
67			try:
68			if os.path.exists('.git'):
69			desc = check_output('git describe --dirty --always --long --abbrev=40'.split(),
70			universal_newlines=True).strip()
71			desc = desc.split('-')
72			if desc[-1].strip() == 'dirty':
73			dirty = True
74			desc.pop()
75			commit = desc[-1].strip('g')
76			elif os.path.exists('.hg'):
77			desc = check_output('hg id --id --debug'.split(), universal_newlines=True).strip()
78			if desc[-1] == '+':
79			dirty = True
80			commit = desc.strip('+')
81			return {
82			'id': commit,
83			'dirty': dirty
84			}
85			except Exception as exc:
86			return {
87			'id': 'unknown',
88			'dirty': dirty,
89			'error': repr(exc),
90			}
91
92
93			def get_current_time():
94			return datetime.now().strftime("%Y%m%d_%H%M%S")
95
96
97			def first_or_value(obj, value):
98			if obj:
99			value, = obj
100
101			return value
102
103
104			def load_timer(string):
105			if "." not in string:
106			raise argparse.ArgumentTypeError("Value for --benchmark-timer must be in dotted form. Eg: 'module.attr'.")
107			mod, attr = string.rsplit(".", 1)
108			if mod == 'pep418':
109			if PY3:
110			import time
111			return NameWrapper(getattr(time, attr))
112			else:
113			from . import pep418
114			return NameWrapper(getattr(pep418, attr))
115			else:
116			__import__(mod)
117			mod = sys.modules[mod]
118			return NameWrapper(getattr(mod, attr))
119
120
121			class RegressionCheck(object):
122			def __init__(self, field, threshold):
123			self.field = field
124			self.threshold = threshold
125
126			def fails(self, current, compared):
127			val = self.compute(current, compared)
128			if val > self.threshold:
129			return "Field %s has failed %s: %.9f > %.9f" % (
130			self.field, self.__class__.__name__, val, self.threshold
131			)
132
133
134			class PercentageRegressionCheck(RegressionCheck):
135			def compute(self, current, compared):
136			val = compared[self.field]
137			if not val:
138			return float("inf")
139			return current[self.field] / val * 100 - 100
140
141
142			class DifferenceRegressionCheck(RegressionCheck):
143			def compute(self, current, compared):
144			return current[self.field] - compared[self.field]
145
146
147			def parse_compare_fail(string,
148			rex=re.compile('^(?P<field>min\|max\|mean\|median\|stddev\|iqr):'
149			'((?P<percentage>[0-9]?[0-9])%\|(?P<difference>[0-9]*\.?[0-9]+([eE][-+]?[0-9]+)?))$')):
150			m = rex.match(string)
151			if m:
152			g = m.groupdict()
153			if g['percentage']:
154			return PercentageRegressionCheck(g['field'], int(g['percentage']))
155			elif g['difference']:
156			return DifferenceRegressionCheck(g['field'], float(g['difference']))
157
158			raise argparse.ArgumentTypeError("Could not parse value: %r." % string)
159
160
161			def parse_warmup(string):
162			string = string.lower().strip()
163			if string == "auto":
164			return platform.python_implementation() == "PyPy"
165			elif string in ["off", "false", "no"]:
166			return False
167			elif string in ["on", "true", "yes", ""]:
168			return True
169			else:
170			raise argparse.ArgumentTypeError("Could not parse value: %r." % string)
171
172
173			def parse_timer(string):
174			return str(load_timer(string))
175
176
177			def parse_sort(string):
178			string = string.lower().strip()
179			if string not in ("min", "max", "mean", "stddev"):
180			raise argparse.ArgumentTypeError(
181			"Unacceptable value: %r. "
182			"Value for --benchmark-sort must be one of: 'min', 'max', 'mean' or 'stddev'." % string)
183			return string
184
185
186			def parse_columns(string):
187			allowed_columns = ["min", "max", "mean", "stddev", "median", "iqr",
188			"outliers", "rounds", "iterations"]
189			columns = [str.strip(s) for s in string.lower().split(',')]
190			invalid = set(columns) - set(allowed_columns)
191			if invalid:
192			# there are extra items in columns!
193			msg = "Invalid column name(s): %s. " % ', '.join(invalid)
194			msg += "The only valid column names are: %s" % ', '.join(allowed_columns)
195			raise argparse.ArgumentTypeError(msg)
196			return columns
197
198
199			def parse_rounds(string):
200			try:
201			value = int(string)
202			except ValueError as exc:
203			raise argparse.ArgumentTypeError(exc)
204			else:
205			if value < 1:
206			raise argparse.ArgumentTypeError("Value for --benchmark-rounds must be at least 1.")
207			return value
208
209
210			def parse_seconds(string):
211			try:
212			return SecondsDecimal(string).as_string
213			except Exception as exc:
214			raise argparse.ArgumentTypeError("Invalid decimal value %r: %r" % (string, exc))
215
216
217			def parse_save(string):
218			if not string:
219			raise argparse.ArgumentTypeError("Can't be empty.")
220			illegal = ''.join(c for c in r"\/:*?<>\|" if c in string)
221			if illegal:
222			raise argparse.ArgumentTypeError("Must not contain any of these characters: /:*?<>\|\\ (it has %r)" % illegal)
223			return string
224
225
226			def time_unit(value):
227			if value < 1e-6:
228			return "n", 1e9
229			elif value < 1e-3:
230			return "u", 1e6
231			elif value < 1:
232			return "m", 1e3
233			else:
234			return "", 1.
235
236
237			def format_time(value):
238			unit, adjustment = time_unit(value)
239			return "{0:.2f}{1:s}".format(value * adjustment, unit)
240
241
242			class cached_property(object):
243			def __init__(self, func):
244			self.__doc__ = getattr(func, '__doc__')
245			self.func = func
246
247			def __get__(self, obj, cls):
248			if obj is None:
249			return self
250			value = obj.__dict__[self.func.__name__] = self.func(obj)
251			return value
252
253
254			def clonefunc(f):
255			"""Deep clone the given function to create a new one.
256
257			By default, the PyPy JIT specializes the assembler based on f.__code__:
258			clonefunc makes sure that you will get a new function with a different
259			__code__, so that PyPy will produce independent assembler. This is useful
260			e.g. for benchmarks and microbenchmarks, so you can make sure to compare
261			apples to apples.
262
263			Use it with caution: if abused, this might easily produce an explosion of
264			produced assembler.
265
266			from: https://bitbucket.org/antocuni/pypytools/src/tip/pypytools/util.py?at=default
267			"""
268
269			# first of all, we clone the code object
270			try:
271			co = f.__code__
272			if PY3:
273			co2 = types.CodeType(co.co_argcount, co.co_kwonlyargcount,
274			co.co_nlocals, co.co_stacksize, co.co_flags, co.co_code,
275			co.co_consts, co.co_names, co.co_varnames, co.co_filename, co.co_name,
276			co.co_firstlineno, co.co_lnotab, co.co_freevars, co.co_cellvars)
277			else:
278			co2 = types.CodeType(co.co_argcount, co.co_nlocals, co.co_stacksize, co.co_flags, co.co_code,
279			co.co_consts, co.co_names, co.co_varnames, co.co_filename, co.co_name,
280			co.co_firstlineno, co.co_lnotab, co.co_freevars, co.co_cellvars)
281			#
282			# then, we clone the function itself, using the new co2
283			return types.FunctionType(co2, f.__globals__, f.__name__, f.__defaults__, f.__closure__)
284			except AttributeError:
285			return f
286
287
288			def format_dict(obj):
289			return "{%s}" % ", ".join("%s: %s" % (k, json.dumps(v)) for k, v in sorted(obj.items()))
290
291
292			def report_progress(iterable, terminal_reporter, format_string, **kwargs):
293			total = len(iterable)
294
295			def progress_reporting_wrapper():
296			for pos, item in enumerate(iterable):
297			string = format_string.format(pos=pos + 1, total=total, value=item, **kwargs)
298			terminal_reporter.rewrite(string, black=True, bold=True)
299			yield string, item
300			return progress_reporting_wrapper()
301

ionelmc / pytest-benchmark

Pull Request — master (#34)

src.pytest_benchmark.parse_columns() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like