Merge pull request #3 from mutating/develop

0.0.3

Author: pomponchik

README.md

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import importlib
+import sys
+from pathlib import Path
+
+from microbenchmark.scenario import Scenario
+from microbenchmark.scenario_group import ScenarioGroup
+
+
+def main(argv: list[str] | None = None) -> None:
+    args = argv if argv is not None else sys.argv[1:]
+
+    if not args or ':' not in args[0]:
+        sys.stderr.write(
+            'microbenchmark: expected TARGET in the form module.path:attr\n'
+            'Usage: microbenchmark module.path:attr [OPTIONS]\n',
+        )
+        sys.exit(3)
+
+    spec, *rest = args
+    colon_pos = spec.index(':')
+    module_path = spec[:colon_pos]
+    attr_name = spec[colon_pos + 1:]
+
+    if not module_path or not attr_name:
+        sys.stderr.write(
+            f'microbenchmark: invalid target {spec!r} — '
+            'expected non-empty module and attribute\n',
+        )
+        sys.exit(3)
+
+    # Ensure CWD is importable so local (non-installed) modules can be found.
+    cwd = str(Path.cwd())
+    if cwd not in sys.path and '' not in sys.path:
+        sys.path.insert(0, cwd)  # pragma: no cover
+
+    try:
+        module = importlib.import_module(module_path)
+    except ModuleNotFoundError as exc:
+        sys.stderr.write(f'microbenchmark: cannot import module {module_path!r}: {exc}\n')
+        sys.exit(3)
+
+    sentinel = object()
+    target: object = getattr(module, attr_name, sentinel)
+    if target is sentinel:
+        sys.stderr.write(
+            f'microbenchmark: module {module_path!r} has no attribute {attr_name!r}\n',
+        )
+        sys.exit(3)
+
+    if not isinstance(target, (Scenario, ScenarioGroup)):
+        sys.stderr.write(
+            f'microbenchmark: {module_path}:{attr_name} is a '
+            f'{type(target).__name__!r}, expected Scenario or ScenarioGroup\n',
+        )
+        sys.exit(3)
+
+    target.cli(argv=rest)
+
+
+if __name__ == '__main__':
+    main()

microbenchmark/__main__.py

@@ -0,0 +1,198 @@
+from __future__ import annotations
+
+import shutil
+from collections.abc import Sequence
+
+_BLOCKS = ' ▁▂▃▄▅▆▇█'
+
+
+def _format_duration(t: float) -> str:
+    """Format *t* seconds into a compact human-readable string."""
+    if t >= 1.0:
+        return f'{t:.3f}s'
+    if t >= 1e-3:
+        return f'{t * 1e3:.2f}ms'
+    if t >= 1e-6:
+        return f'{t * 1e6:.2f}\u03bcs'
+    return f'{t * 1e9:.2f}ns'
+
+
+def terminal_width() -> int:
+    """Return the current terminal width, clamped to a minimum of 20."""
+    cols = shutil.get_terminal_size((80, 24)).columns
+    return max(cols, 20)
+
+
+def draw_box(lines: list[str], width: int) -> list[str]:
+    """Draw a Unicode border box around *lines*.
+
+    Each output line is exactly *width* characters wide.  Content lines are
+    padded with spaces or truncated to ``width - 4`` characters (2 chars for
+    the border + space on each side).
+
+    Box drawing uses ASCII + box-drawing characters (``╭╮╰╯│─``).  Width is
+    measured by ``len()``, which counts every character as 1 column.  Emoji or
+    CJK characters may cause visual misalignment.
+
+    Args:
+        lines: Content lines to render inside the box.
+        width: Total width of each output line, including borders.
+    """
+    inner_width = width - 4  # 2 for │ and 1 space on each side
+    top = '╭' + '─' * (width - 2) + '╮'
+    bottom = '╰' + '─' * (width - 2) + '╯'
+    result = [top]
+    for line in lines:
+        if len(line) > inner_width:
+            content = line[:inner_width]
+        else:
+            content = line.ljust(inner_width)
+        result.append('│ ' + content + ' │')
+    result.append(bottom)
+    return result
+
+
+def draw_nested(
+    inner_blocks: list[list[str]],
+    extras: list[str],
+    width: int,
+) -> list[str]:
+    """Draw an outer border box wrapping pre-rendered *inner_blocks*.
+
+    Each inner block is already a list of strings (e.g. from :func:`draw_box`).
+    They are placed inside the outer border with 1-space padding on each side.
+    *extras* are additional plain-text lines rendered after the inner blocks,
+    also padded.
+
+    Args:
+        inner_blocks: List of already-rendered blocks (each a list of strings).
+        extras: Extra plain-text lines appended after the inner blocks.
+        width: Total width of the outer box.
+    """
+    inner_width = width - 4  # space for │ + space on each side
+
+    def pad_inner_line(line: str) -> str:
+        if len(line) > inner_width:
+            content = line[:inner_width]
+        else:
+            content = line.ljust(inner_width)
+        return '│ ' + content + ' │'
+
+    top = '╭' + '─' * (width - 2) + '╮'
+    bottom = '╰' + '─' * (width - 2) + '╯'
+    result = [top]
+    for block in inner_blocks:
+        for line in block:
+            result.append(pad_inner_line(line))
+    for line in extras:
+        result.append(pad_inner_line(line))
+    result.append(bottom)
+    return result
+
+
+def histogram_bounds(durations: Sequence[float]) -> tuple[float, float]:
+    """Return ``(lo, hi)`` bounds for a histogram: minimum and p99 value.
+
+    The p99 clip prevents extreme outliers from compressing the bulk of the
+    distribution into the leftmost column.
+
+    Args:
+        durations: Sequence of per-call timings in seconds.
+
+    Raises:
+        ValueError: If *durations* is empty.
+    """
+    if not durations:
+        raise ValueError('durations must not be empty')
+    sorted_durs = sorted(durations)
+    lo = sorted_durs[0]
+    p99_idx = min(len(sorted_durs) - 1, int(len(sorted_durs) * 0.99))
+    return lo, sorted_durs[p99_idx]
+
+
+def draw_histogram_axis(lo: float, hi: float, width: int) -> str:
+    """Return a single-line x-axis label for a histogram with bounds *lo*/*hi*.
+
+    The minimum value is left-aligned; the p99 clip value is right-aligned.
+    Both are formatted with auto-selected time units (ns / μs / ms / s).
+
+    Args:
+        lo: Minimum value displayed on the x-axis.
+        hi: Maximum value (p99 clip point) displayed on the x-axis.
+        width: Total width of the label in characters.
+    """
+    if width < 1:
+        return ''
+    left = _format_duration(lo)
+    right = _format_duration(hi) + ' (p99)'
+    if len(left) + 1 + len(right) > width:
+        return left[:width]
+    spaces = width - len(left) - len(right)
+    return left + ' ' * spaces + right
+
+
+def draw_histogram(durations: Sequence[float], width: int, height: int) -> list[str]:
+    """Render an ASCII bar chart of *durations* as a ``height`` x ``width`` grid.
+
+    The output is ``width`` characters wide and ``height`` rows tall.  Each
+    cell uses one of the Unicode block characters ``' ▁▂▃▄▅▆▇█'`` so bar
+    heights are resolved at 1/8-row precision, giving smooth transitions
+    between adjacent buckets.  Returns an empty list when any dimension is
+    zero/negative or when *durations* is empty.
+
+    Internally the data is bucketed into at most 20 bins regardless of
+    *width*.  Each bin is then rendered as ``width // n_buckets`` characters
+    wide.  This prevents timer-quantisation artefacts from producing a single
+    spike with stray isolated pixels.
+
+    The x-axis upper bound is clipped to the p99 value via
+    :func:`histogram_bounds`.  Values above the clip point are omitted.
+
+    When all values are identical (``hi == lo``), the middle bin is drawn
+    full-height and all others are left empty.
+
+    Args:
+        durations: Sequence of per-call timings in seconds.
+        width: Total number of output characters per row.
+        height: Number of rows in the chart.
+    """
+    if width < 1 or height < 1 or len(durations) == 0:
+        return []
+
+    lo, hi = histogram_bounds(durations)
+
+    # Cap the number of data buckets at 20 so that adjacent quantised timer
+    # values are merged into the same wider bar.
+    n_buckets = min(width, 20)
+    counts = [0] * n_buckets
+
+    if hi == lo:
+        counts[n_buckets // 2] = 1
+    else:
+        span = hi - lo
+        for d in durations:
+            if d > hi:
+                continue
+            idx = min(n_buckets - 1, int((d - lo) / span * n_buckets))
+            counts[idx] += 1
+
+    max_count = max(counts)
+    bar_heights_float = [
+        c / max_count * height if c > 0 else 0.0
+        for c in counts
+    ]
+
+    rows: list[str] = []
+    for row in range(height - 1, -1, -1):
+        line_chars: list[str] = []
+        for col in range(width):
+            bh = bar_heights_float[col * n_buckets // width]
+            if bh >= row + 1:
+                line_chars.append('█')
+            elif bh > row:
+                frac = bh - row
+                line_chars.append(_BLOCKS[max(1, min(8, round(frac * 8)))])
+            else:
+                line_chars.append(' ')
+        rows.append(''.join(line_chars))
+    return rows

microbenchmark/_render.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 from printo import describe_data_object
+from sigmatch import PossibleCallMatcher
+from sigmatch.errors import SignatureNotFoundError, UnsupportedSignatureError
 
 
 class arguments:  # noqa: N801
@@ -28,3 +30,25 @@ def __eq__(self, other: object) -> bool:
 
     def __hash__(self) -> int:
         return hash((self.args, tuple(sorted(self.kwargs.items()))))
+
+    def match(self, function: object) -> bool:
+        """Check whether *function* can be called with these arguments.
+
+        Returns ``True`` if the call is compatible with the function's
+        signature, ``False`` if not.
+
+        **Limitation:** if Python cannot introspect the signature of
+        *function* (e.g. built-in / C-extension functions such as ``len``),
+        the check is silently skipped and ``True`` is returned.  The function
+        will be validated at runtime when the benchmark actually runs.
+        """
+        from sigmatch import SignatureMismatchError  # noqa: PLC0415
+        shape = ('.',) * len(self.args) + tuple(self.kwargs)
+        matcher: PossibleCallMatcher = PossibleCallMatcher(*shape)
+        try:
+            matcher.match(function, raise_exception=True)  # type: ignore[arg-type]
+            return True
+        except SignatureMismatchError:
+            return False
+        except (SignatureNotFoundError, UnsupportedSignatureError):
+            return True

microbenchmark/arguments.py

@@ -19,6 +19,7 @@ class _ScenarioMeta(TypedDict):
 
 class _ResultJson(TypedDict):
     durations: list[float]
+    total_duration: float
     is_primary: bool
     scenario: _ScenarioMeta | None
 
@@ -27,16 +28,19 @@ class _ResultJson(TypedDict):
 class BenchmarkResult:
     scenario: Scenario | None
     durations: tuple[float, ...]
+    total_duration: float
     is_primary: bool = True
 
     mean: float = field(init=False)
     best: float = field(init=False)
     worst: float = field(init=False)
+    functions_duration: float = field(init=False)
 
     def __post_init__(self) -> None:
         self.mean = math.fsum(self.durations) / len(self.durations)
         self.best = min(self.durations)
         self.worst = max(self.durations)
+        self.functions_duration = math.fsum(self.durations)
 
     def percentile(self, p: float) -> BenchmarkResult:
         if not (0 < p <= 100):
@@ -46,6 +50,7 @@ def percentile(self, p: float) -> BenchmarkResult:
         return BenchmarkResult(
             scenario=self.scenario,
             durations=trimmed,
+            total_duration=0.0,
             is_primary=False,
         )
 
@@ -73,6 +78,7 @@ def to_json(self) -> str:
             scenario_meta = None
         data: _ResultJson = _ResultJson(
             durations=list(self.durations),
+            total_duration=self.total_duration,
             is_primary=self.is_primary,
             scenario=scenario_meta,
         )
@@ -91,8 +97,14 @@ def from_json(cls, data: str) -> BenchmarkResult:
             raise ValueError('durations must be a list')
         if not isinstance(raw_is_primary, bool):
             raise ValueError('is_primary must be a bool')
+        raw_total: object = raw.get('total_duration')
+        if raw_total is None:
+            total_duration = math.fsum(float(d) for d in raw_durations)
+        else:
+            total_duration = float(raw_total)  # type: ignore[arg-type]
         return cls(
             scenario=None,
             durations=tuple(float(d) for d in raw_durations),
+            total_duration=total_duration,
             is_primary=raw_is_primary,
         )