liblaf.grapes ¶

def __attrs_post_init__(self) -> None:
    _warnings_hide = True
    if self.label is None and self.name is not None:
        warnings.warn(
            "'name' parameter is deprecated. Please use 'label' instead.",
            DeprecationWarning,
        )
        self.label = self.name

def __bool__(self) -> bool:
    return True

def __len__(self) -> int:
    return len(self.timings[self.default_clock])

def clear(self) -> None:
    self.timings.clear()
    self._start_time.clear()
    self._stop_time.clear()

def elapsed(self, clock_name: ClockName | None = None) -> float:
    clock_name = clock_name or self.default_clock
    stop_time: float
    if clock_name in self._stop_time:
        stop_time = self._stop_time[clock_name]
    else:
        stop_time = clock(clock_name)
    return stop_time - self._start_time[clock_name]

def finish(self) -> None:
    _logging_hide = True
    if self.cb_finish is not None:
        self.cb_finish(self)

def log_record(
    self,
    *,
    index: int = -1,
    level: int = LOG_RECORD_DEFAULT_LEVEL,
    limits: str | limits.RateLimitItem | None = "1/second",
) -> None:
    _logging_hide = True
    autolog.log(
        level,
        "%s",
        LazyStr(lambda: self.pretty_record(index=index)),
        extra={
            "limits": LimitsHitArgs(
                item=limits, identifiers=(self.label or "Timer",)
            )
        },
    )

def log_summary(
    self,
    *,
    level: int = LOG_SUMMARY_DEFAULT_LEVEL,
    stats: Iterable[StatisticName] = LOG_SUMMARY_DEFAULT_STATISTICS,
    limits: str | limits.RateLimitItem | None = None,
) -> None:
    _logging_hide = True
    autolog.log(
        level,
        "%s",
        LazyStr(lambda: self.pretty_summary(stats=stats)),
        extra={
            "limits": LimitsHitArgs(
                item=limits, identifiers=(self.label or "Timer",)
            ),
            "markup": lambda: self.pretty_summary(stats=stats, rich_markup=True),
        },
    )

def pretty_record(self, index: int = -1) -> str:
    name: str = self.label or "Timer"
    items: list[str] = [
        f"{clock_name}: {pretty.pretty_duration(self.timings[clock_name][index])}"
        for clock_name in self.clocks
    ]
    items_str: str = ", ".join(items)
    return f"{name} > {items_str}"

def pretty_summary(
    self,
    stats: Iterable[StatisticName] = LOG_SUMMARY_DEFAULT_STATISTICS,
    *,
    rich_markup: bool = False,
) -> str:
    name: str = self.label or "Timer"
    header: str = f"{name} (count: {len(self)})"
    if len(self) == 0:
        return header
    lines: list[str] = []
    for clock_name in self.clocks:
        stats_str: list[str] = []
        for stat in stats:
            stat_name: str
            value: str
            stat_name, value = pretty_statistic(
                self.timings[clock_name], stat, rich_markup=rich_markup
            )
            stats_str.append(f"{stat_name}: {value}")
        line: str = f"{clock_name} > {', '.join(stats_str)}"
        lines.append(line)
    if len(self.clocks) == 1:
        return f"{header} {lines[0]}"
    return f"{header}\n" + "\n".join(lines)

def start(self) -> None:
    _logging_hide = True
    for clock_name in self.clocks:
        self._start_time[clock_name] = clock(clock_name)
    self._stop_time.clear()
    if self.cb_start is not None:
        self.cb_start(self)

def stop(self) -> None:
    _logging_hide = True
    for clock_name in self.clocks:
        stop_time: float = clock(clock_name)
        self._stop_time[clock_name] = stop_time
        self.timings[clock_name].append(stop_time - self._start_time[clock_name])
    if self.cb_stop is not None:
        self.cb_stop(self)

def plot(
    self,
    *,
    relative_to: str | None = None,
    xlabel: str = "Size",
    ylabel: str = "Time (sec)",
    log_scale: bool = True,
) -> Figure:
    with deps.optional_deps("liblaf-grapes", "bench"):
        import matplotlib.pyplot as plt

    if relative_to is None:
        relative_to = min(self.timings.keys(), key=lambda k: self.timings[k][-1])

    fig: Figure
    ax0: Axes
    ax1: Axes
    fig, (ax0, ax1) = plt.subplots(1, 2, sharex="all", figsize=(12.8, 4.8))

    for label, timings in self.timings.items():
        ax0.plot(self.sizes, timings, label=label)
        base: list[float] = self.timings[relative_to]
        relative: list[float] = [t / b for t, b in zip(timings, base, strict=True)]
        ax1.plot(self.sizes, relative, label=label)

    ax0.grid(which="both", linestyle="--")
    ax0.legend()
    ax0.set_xlabel(xlabel)
    ax0.set_ylabel(ylabel)

    ax1.grid(which="both", linestyle="--")
    ax1.legend()
    ax1.set_xlabel(xlabel)
    ax1.set_ylabel(f"{ylabel} (relative to {relative_to})")

    if log_scale:
        ax0.set_xscale("log")
        ax0.set_yscale("log")
        ax1.set_xscale("log")

    fig.tight_layout()
    return fig

def bench(
    self, func: Callable | None = None, *, label: str | None = None
) -> Callable:
    if func is None:
        return functools.partial(self.bench, label=label)
    if label is None:
        label = func.__name__
    self._registry[label] = func
    return func

def run(self) -> BenchResults:
    inputs: Sequence[tuple[Sequence, Mapping]] = list(self._setup())
    sizes_and_inputs: list[tuple[float, tuple[Sequence, Mapping]]] = [
        (self._size_fn(*args, **kwargs), (args, kwargs)) for args, kwargs in inputs
    ]
    sizes_and_inputs.sort(key=lambda x: x[0])
    sizes: Sequence[float] = [size for size, _ in sizes_and_inputs]
    inputs = [(args, kwargs) for _, (args, kwargs) in sizes_and_inputs]
    outputs: dict[str, list[Any]] = {}
    timings: dict[str, list[float]] = {}
    for label, func in self._registry.items():
        outputs_name: list[Any] = []
        timings_name: list[float] = []
        for size, (args, kwargs) in zip(sizes, inputs, strict=True):
            output: Any
            elapsed: float
            output, elapsed = self._bench(func, args, kwargs)
            outputs_name.append(output)
            timings_name.append(elapsed)
            logger.debug("Bench %s(%g) took %g sec", label, size, elapsed)
            if elapsed > self.timeout:
                logger.warning("Bench %s(%g) timed out", label, size)
                break
        if len(timings_name) < len(sizes):
            n_remain: int = len(sizes) - len(timings_name)
            outputs_name.extend([None] * n_remain)
            timings_name.extend([math.inf] * n_remain)
        timings[label] = timings_name
        outputs[label] = outputs_name
    return BenchResults(outputs=outputs, sizes=list(sizes), timings=timings)

def setup[C: Callable[..., Iterable[tuple[Sequence, Mapping]]]](self, func: C) -> C:
    self._setup = func
    return func

def size[C: Callable[..., float]](self, func: C) -> C:
    self._size_fn = func
    return func

def __str__(self) -> str:
    from liblaf.grapes import pretty

    pretty_call: str = pretty.pretty_call(
        self.func, self.params.args, self.params.kwargs
    )
    return f"`{pretty_call}` could not be resolved."

def __repr__(self) -> str:
    return self.__wrapped__

def __str__(self) -> str:
    return self.__wrapped__

def __str__(self) -> str:
    cls: str = self.typ if isinstance(self.typ, str) else self.typ.__qualname__
    return f"{self.value!r} is not a valid {cls}."

@override
def advance(self, task_id: TaskID, advance: float = 1) -> None:
    _logging_hide = True
    self.stale = True
    super().advance(task_id, advance)
    self.refresh()

@override
@classmethod
def get_default_columns(cls) -> tuple[str | ProgressColumn, ...]:  # pyright: ignore[reportIncompatibleMethodOverride]
    return (
        # SpinnerColumn(),
        TextColumn("{task.description}", style="progress.description"),
        BarColumn(),
        TaskProgressColumn(),
        MofNCompleteColumn(),
        "[",
        TimeElapsedColumn(),
        "<",
        TimeRemainingColumn(),
        ",",
        RateColumn(),
        "]",
    )

@override
def refresh(self, *, force: bool = False) -> None:
    _logging_hide = True
    if self.stale and (self.limiter.hit(self.limit) or force):
        self.stale = False
        self.logger.info(self.get_renderable())

@override
def reset(
    self,
    task_id: TaskID,
    *,
    start: bool = True,
    total: float | None = None,
    completed: int = 0,
    visible: bool | None = None,
    description: str | None = None,
    **fields: Any,
) -> None:
    _logging_hide = True
    self.stale = True
    super().reset(
        task_id,
        start=start,
        total=total,
        completed=completed,
        visible=visible,
        description=description,
        **fields,
    )

@override
def start(self) -> None:
    pass

@override
def stop(self) -> None:
    _logging_hide = True
    self.refresh(force=True)

@override
def update(
    self,
    task_id: TaskID,
    *,
    total: float | None = None,
    completed: float | None = None,
    advance: float | None = None,
    description: str | None = None,
    visible: bool | None = None,
    refresh: bool = False,
    **fields: Any,
) -> None:
    _logging_hide = True
    self.stale = True
    super().update(
        task_id,
        total=total,
        completed=completed,
        advance=advance,
        description=description,
        visible=visible,
        refresh=refresh,
        **fields,
    )