import warnings
from collections.abc import Collection, Iterable
from typing import Any
import pandas as pd
from ._multi_case_timer import CandidateMethodArg, MultiCaseTimer, TestDataArg
from ._util import to_dataframe
from .types import DataFunc, DataType, SetupFunc, StratifyArg, Ts
[docs]
def run_multivariate_test(
candidate_method: CandidateMethodArg[DataType],
test_data: TestDataArg[DataType] | DataFunc[*Ts, DataType], # DataFunc[DataFuncP, DataType]
*,
time_per_candidate: float = 6.0,
repeat: int = 5,
plot: bool = True,
show: bool = True,
names: Iterable[str] | None = (),
progress: bool = False,
setup: SetupFunc[DataType] | None = None,
warmup: int = 0,
stratify: StratifyArg = None,
case_args: Collection[tuple[*Ts]] | None = None,
kwargs: Any | None = None,
**plot_kwargs: Any,
) -> pd.DataFrame:
"""Run performance tests for multiple candidate methods on collections of test data.
This is a convenience method which combines :meth:`MultiCaseTimer.run() <rics.performance.MultiCaseTimer.run>`,
:meth:`~rics.performance.to_dataframe` and -- if plotting is enabled -- :meth:`~rics.performance.plot_run`. For full
functionally these methods should be use directly.
Args:
candidate_method: A dict ``{label: function}``. Alternatively, you may pass a collection of functions or a
single function.
test_data: A ``{label: data}`` to evaluate candidates on. You may also pass a list of data, which will be
converted to a dict as above. Data may also be generated by passing a callable.
time_per_candidate: Minimum runtime per repetition and candidate label. When `stratify` is set this budget
applies **per** ``(candidate, stratum)`` **instead**, so total runtime scales with the number of strata.
repeat: Number of times to repeat for all candidates per data label.
plot: If ``True``, plot a figure using :meth:`~rics.performance.plot_run`.
show: If ``True``, attempt to display the figure. Ignored when ``plot=False``.
names: Level names for tuple keys in the data (creates new columns). See :func:`.plot_run` for details. Set to
``None`` to disable derived names when `test_data` is callable.
progress: If ``True``, display progress. Uses ``tqdm`` on a TTY and falls back to periodic logging otherwise
(so ``tqdm`` is optional).
setup: A callable ``(data) -> data`` invoked -- **not** measured -- before each timed repetition to produce a
fresh input (mirrors :py:class:`timeit.Timer`'s ``setup``). Use for candidates that mutate their input, or
to reset shared state (e.g. caches) between repetitions.
warmup: Number of untimed calls per candidate/data pair before timing begins (warms caches/JIT/imports).
stratify: Groups variants of comparable cost so that the iteration count is calibrated once per ``(candidate,
stratum)`` instead of once per candidate function. Using ``"auto"`` implicitly calls
:meth:`~rics.performance.MultiCaseTimer.fit_strata` the first time. Set to ``None`` to disable.
case_args: Collection of positional arguments for a `test_data` callable.
kwargs: Shared keyword arguments for a `test_data` callable.
**plot_kwargs: See :func:`.plot_run` for details. Ignored if ``plot=False``.
Returns:
A long-form DataFrame of results.
Raises:
RuntimeError: If Seaborn isn't installed and ``plot=True``.
TypeError: If `case_args` or `kwargs` are set when `test_data` is not a callable.
ValueError: If `case_args` is empty and `test_data` is a callable.
See Also:
The :func:`~rics.performance.plot_run` and :func:`~rics.performance.get_best` functions.
"""
if plot:
_verify_can_plot()
elif plot_kwargs:
# TODO(7.0.0): Raise.
warnings.warn(
f"Ignoring {plot_kwargs=} since {plot=}; this will raise in a future version.",
FutureWarning,
stacklevel=2,
)
timer: MultiCaseTimer[DataType, *Ts] = MultiCaseTimer(
candidate_method,
test_data,
case_args=case_args,
kwargs=kwargs,
setup=setup,
warmup=warmup,
)
run_results = timer.run(
time_per_candidate=time_per_candidate,
repeat=repeat,
progress=progress,
stratify=stratify,
)
if names is None:
names = ()
else:
names = [*names]
if not names and timer.is_data_generated:
names = timer.derive_names()
data = to_dataframe(run_results, names=names)
if plot:
from matplotlib.pyplot import show as show_figure
from ._plot import plot_run
plot_run(data, names=names, **plot_kwargs)
if show:
show_figure(block=False)
return data
def _verify_can_plot() -> None:
from importlib.util import find_spec
if find_spec("seaborn") is None:
msg = (
"Package `seaborn` not installed. Run of one:"
"\n pip install seaborn"
"\n pip install rics[plotting]"
"\nto enable plots."
)
raise RuntimeError(msg)
