diff --git a/src/nwb_benchmarks/benchmarks/track_incremental_slicing.py b/src/nwb_benchmarks/benchmarks/track_incremental_slicing.py index 1c15639b..5625709d 100644 --- a/src/nwb_benchmarks/benchmarks/track_incremental_slicing.py +++ b/src/nwb_benchmarks/benchmarks/track_incremental_slicing.py @@ -123,9 +123,12 @@ def _track_cumulative_slice_times(self, params: dict[str, Any]) -> dict[str, lis return {"cumulative_time_in_seconds": cumulative_times} @skip_benchmark_if(not RUN_INCREMENTAL_SLICING_BENCHMARKS) - def track_cumulative_slice_times(self, params: dict[str, Any]) -> dict[str, list[float]]: + def track_cumulative_slice_times(self, params: dict[str, Any]) -> dict[str, Any]: """Track cumulative open + repeated slice timing.""" - return self._track_cumulative_slice_times(params=params) + # Same structure as `NetworkTracker.asv_network_statistics`: ASV stores 'samples' in the samples column of + # the raw results (with `--record-samples`), which is where `reduce_results` reads every benchmark from. + # 'number' is simply required, but needs to be None for custom track_ functions. + return dict(samples=self._track_cumulative_slice_times(params=params), number=None) class HDF5PyNWBRemfileWithCacheIncrementalSliceBenchmark(IncrementalSliceBenchmark): diff --git a/src/nwb_benchmarks/database/_models.py b/src/nwb_benchmarks/database/_models.py index 3bce5fc7..f44830c6 100644 --- a/src/nwb_benchmarks/database/_models.py +++ b/src/nwb_benchmarks/database/_models.py @@ -64,8 +64,6 @@ def parse_parameter_case(s): # if the parsed string is not a dict (older benchmarks results), convert it to one if not isinstance(output, dict): - if output == (): - return {} output = {"https_url": output[0].strip("'")} return output diff --git a/src/nwb_benchmarks/setup/_reduce_results.py b/src/nwb_benchmarks/setup/_reduce_results.py index cf2207ba..291186aa 100644 --- a/src/nwb_benchmarks/setup/_reduce_results.py +++ b/src/nwb_benchmarks/setup/_reduce_results.py @@ -3,7 +3,6 @@ import collections import datetime import hashlib -import itertools import json import math import pathlib @@ -17,86 +16,6 @@ from ..utils import get_dictionary_checksum -def _serialize_parameter_cases(serialized_params: list) -> list[str]: - """Return one serialized parameter-case key per ASV result. - - ASV stores parameters as a list of parameter axes. Preserve the historical - single-axis keys used by nwb_benchmarks results, while supporting ASV's - general zero-axis and multi-axis parameter layouts. - """ - - if len(serialized_params) == 0: - return ["()"] - if len(serialized_params) == 1: - return serialized_params[0] - - return [str(parameter_case) for parameter_case in itertools.product(*serialized_params)] - - -def _extract_successful_results(test_case: str, raw_results_list: list, result_columns: list[str] | None) -> dict: - """Extract successful ASV benchmark results keyed by serialized parameter case. - - ASV has used multiple result serialization layouts. Older result files store a - fixed-length list where parameters are at index 1 and samples are at index 11. - Newer result files include a top-level ``result_columns`` list and each result - row follows that column order. This helper normalizes both layouts to the - reduced results format used by nwb_benchmarks. - """ - - if raw_results_list is None: - return {} - - if result_columns is not None: - column_index = {column_name: index for index, column_name in enumerate(result_columns)} - if "result" not in column_index or "params" not in column_index: - return {} - - params_index = column_index["params"] - result_index = column_index["result"] - if params_index >= len(raw_results_list) or result_index >= len(raw_results_list): - return {} - - serialized_params = raw_results_list[params_index] - aggregate_results = raw_results_list[result_index] - raw_results = aggregate_results - if "samples" in column_index: - samples_index = column_index["samples"] - if samples_index < len(raw_results_list) and raw_results_list[samples_index] is not None: - raw_results = raw_results_list[samples_index] - # Older ASV layout used by the original reducer implementation. - elif len(raw_results_list) == 12: - aggregate_results = raw_results_list[0] - serialized_params = raw_results_list[1] - raw_results = raw_results_list[11] - else: - return {} - - # Skipped results in JSON are written as `null` and read back into Python as `None`. - if aggregate_results is None or raw_results is None: - return {} - if not isinstance(aggregate_results, list): - aggregate_results = [aggregate_results] - if not isinstance(raw_results, list): - raw_results = [raw_results] - serialized_params = _serialize_parameter_cases(serialized_params=serialized_params) - - if len(serialized_params) != len(aggregate_results) or len(serialized_params) != len(raw_results): - message = ( - f"In intermediate results for test case {test_case}: \n" - f"\tLength mismatch between parameters ({len(serialized_params)}) and " - f"results ({len(aggregate_results)}) or result samples ({len(raw_results)})!\n\n" - "Please raise an issue and share your intermediate results file." - ) - warnings.warn(message=message) - return {} - - return { - params: raw_result - for params, aggregate_result, raw_result in zip(serialized_params, aggregate_results, raw_results) - if not (isinstance(aggregate_result, float) and math.isnan(aggregate_result)) and raw_result is not None - } - - def _parse_environment_info(raw_environment_info: List[str]) -> Dict[str, List[Dict[str, str]]]: """Turn the results of `conda list` printout to a JSON dictionary.""" header_stripped = raw_environment_info[3:] @@ -124,15 +43,38 @@ def reduce_results(machine_id: str, raw_results_file_path: pathlib.Path, raw_env timestamp = datetime.datetime.now().strftime("%Y-%m-%d-%H-%M-%S") reduced_results = dict() - result_columns = raw_results_info.get("result_columns") for test_case, raw_results_list in raw_results_info["results"].items(): - extracted_results = _extract_successful_results( - test_case=test_case, - raw_results_list=raw_results_list, - result_columns=result_columns, - ) - if extracted_results: - reduced_results.update({test_case: extracted_results}) + + # Only successful runs have a results field of length 12 + if len(raw_results_list) != 12: + continue + + # This code assumes that test cases are only run with one parameter + assert len(raw_results_list[1]) == 1, "Unexpected length of serialized parameters list!" + serialized_params = raw_results_list[1][0] + + aggregate_results = raw_results_list[0] + samples_per_parameter_set = raw_results_list[11] + if not len(serialized_params) == len(aggregate_results) == len(samples_per_parameter_set): + message = ( + f"In intermediate results for test case {test_case}: \n" + f"\tLength mismatch between parameters ({len(serialized_params)}), results " + f"({len(aggregate_results)}) and result samples ({len(samples_per_parameter_set)})!\n\n" + "Please raise an issue and share your intermediate results file." + ) + warnings.warn(message=message) + continue + + # ASV records every parameter set of a benchmark, including those a `--bench` pattern did not select. + # Those were never run and have a NaN result. Samples of parameter sets that failed are written as `null` + # and read back into Python as `None`. Leave both out, keeping the parameter sets that succeeded. + successful_results = { + params: samples + for params, result, samples in zip(serialized_params, aggregate_results, samples_per_parameter_set) + if not (isinstance(result, float) and math.isnan(result)) and samples is not None + } + if len(successful_results) > 0: + reduced_results.update({test_case: successful_results}) if len(reduced_results) == 0: raise ValueError( diff --git a/tests/data/asv_0.6.1_raw_results.json b/tests/data/asv_0.6.1_raw_results.json index b4987d9c..7f4c59d4 100644 --- a/tests/data/asv_0.6.1_raw_results.json +++ b/tests/data/asv_0.6.1_raw_results.json @@ -18,13 +18,7 @@ "results": { "bench.Incremental.track_cumulative_slice_times": [ [ - { - "cumulative_time_in_seconds": [ - 0.5, - 1.0, - 1.5 - ] - }, + true, null ], [ @@ -33,9 +27,25 @@ "{'name': 'B'}" ] ], - "b3e7fa9dfc40a9240d5f4dc9d6e3e552a4fbbe90b8b7e12ee43c6aff1c810eef", - 1791349309228, - 0.023083 + "baffb96bc63f705ae883e56b15ebc74070d05406a779dc4ec4f382bfe6d84b36", + 1791350742012, + 0.017311, + null, + null, + null, + null, + null, + null, + [ + { + "cumulative_time_in_seconds": [ + 0.5, + 1.0, + 1.5 + ] + }, + null + ] ], "bench.Network.track_network": [ [ @@ -49,8 +59,8 @@ ] ], "26e3a7e66482e998b12768ed82102e95f8770e4e2f3737de978309feb7d94f69", - 1791349309252, - 0.012449, + 1791350742031, + 0.013156, null, null, null, @@ -72,7 +82,7 @@ ], "bench.Timed.time_read": [ [ - 1.2534999996205443e-05, + 1.3246999998273168e-05, NaN ], [ @@ -82,8 +92,8 @@ ] ], "267b65a446ea0bd2b4564e2747f943211b97455e145f7675e0abcc11247da3ad", - 1791349309265, - 0.006144, + 1791350742044, + 0.007133, [ -Infinity, null @@ -93,11 +103,11 @@ null ], [ - 1.2535e-05, + 1.3247e-05, null ], [ - 1.2535e-05, + 1.3247e-05, null ], [ @@ -110,10 +120,37 @@ ], [ [ - 1.2534999996205443e-05 + 1.3246999998273168e-05 ], null ] + ], + "bench.Unwrapped.track_unwrapped": [ + [ + { + "cumulative_time_in_seconds": [ + 0.5, + 1.0, + 1.5 + ] + }, + { + "cumulative_time_in_seconds": [ + 0.5, + 1.0, + 1.5 + ] + } + ], + [ + [ + "{'name': 'A'}", + "{'name': 'B'}" + ] + ], + "c89cd8f48cba96f3dc897029408daf80651d6bfa6c5477abd567a34b6c5f0c37", + 1791350742052, + 0.011107 ] }, "version": 2 diff --git a/tests/test_reduce_results.py b/tests/test_reduce_results.py index ce1be40e..0932cacd 100644 --- a/tests/test_reduce_results.py +++ b/tests/test_reduce_results.py @@ -1,28 +1,31 @@ import json import math import pathlib -import shutil import pytest from nwb_benchmarks.setup import _reduce_results -from nwb_benchmarks.setup._reduce_results import ( - _extract_successful_results, - _serialize_parameter_cases, - reduce_results, -) +from nwb_benchmarks.setup._reduce_results import reduce_results # Written by `asv run --record-samples` (as `nwb_benchmarks run` calls it) with asv 0.6.1, trimmed to the keys the -# reducer reads. It holds one benchmark of each result shape the suite produces: -# - `Incremental`: a `track_` benchmark returning a plain dict, as `track_incremental_slicing` does. ASV stores it in -# the `result` column without samples. Parameter set B raised, so its result is `null`. +# reducer reads. It holds one benchmark of each result shape below: +# - `Incremental`: a `track_` benchmark returning `dict(samples=..., number=None)` with a list of cumulative times, as +# `track_incremental_slicing` does. Parameter set B raised, so its samples are `null`. # - `Network`: a `track_` benchmark returning `dict(samples=..., number=None)`, as the network tracking benchmarks do. # - `Timed`: a `time_` benchmark run with a `--bench` pattern selecting only parameter set A, so B's result is NaN. +# - `Unwrapped`: a `track_` benchmark returning a plain dict, which ASV stores in the `result` column without samples. RAW_RESULTS_FILE_PATH = pathlib.Path(__file__).parent / "data" / "asv_0.6.1_raw_results.json" INCREMENTAL = "bench.Incremental.track_cumulative_slice_times" NETWORK = "bench.Network.track_network" TIMED = "bench.Timed.time_read" +UNWRAPPED = "bench.Unwrapped.track_unwrapped" + +NETWORK_STATISTICS = { + "total_transfer_in_bytes": 100, + "total_traffic_in_number_of_web_packets": 4, + "total_transfer_time_in_seconds": 2.0, +} @pytest.fixture @@ -31,125 +34,102 @@ def raw_results_info() -> dict: return json.load(fp=file_stream) -def extract(raw_results_info: dict, test_case: str, result_columns="from file") -> dict: - return _extract_successful_results( - test_case=test_case, - raw_results_list=raw_results_info["results"][test_case], - result_columns=raw_results_info["result_columns"] if result_columns == "from file" else result_columns, +@pytest.fixture +def reduce(tmp_path, monkeypatch): + """Return a function that runs `reduce_results` on raw results, writing into a temporary home directory.""" + for name in ("RESULTS_DIR", "ENVIRONMENTS_DIR", "MACHINES_DIR"): + directory = tmp_path / name.lower() + directory.mkdir() + monkeypatch.setattr(_reduce_results, name, directory) + (tmp_path / "machines_dir" / "machine-test.json").write_text("{}") + + raw_environment_info_file_path = tmp_path / "environment.txt" + raw_environment_info_file_path.write_text( + "# packages in environment at /conda/envs/nwb_benchmarks:\n" + "#\n" + "# Name Version Build Channel\n" + "pynwb 3.1.0 pypi_0 pypi\n" ) + def _reduce(raw_results_info: dict) -> pathlib.Path: + raw_results_file_path = tmp_path / "raw_results.json" + with open(raw_results_file_path, mode="w") as file_stream: + json.dump(obj=raw_results_info, fp=file_stream) -def test_incremental_results_keep_the_successful_parameter_set(raw_results_info): - assert extract(raw_results_info, INCREMENTAL) == {"{'name': 'A'}": {"cumulative_time_in_seconds": [0.5, 1.0, 1.5]}} + reduce_results( + machine_id="test", + raw_results_file_path=raw_results_file_path, + raw_environment_info_file_path=raw_environment_info_file_path, + ) + assert not raw_results_file_path.exists() + (reduced_results_file_path,) = (tmp_path / "results_dir").glob("*_machine-test_results.json") + return reduced_results_file_path -def test_network_results_are_read_from_the_samples(raw_results_info): - network_statistics = { - "total_transfer_in_bytes": 100, - "total_traffic_in_number_of_web_packets": 4, - "total_transfer_time_in_seconds": 2.0, - } + return _reduce - assert extract(raw_results_info, NETWORK) == { - "{'name': 'A'}": network_statistics, - "{'name': 'B'}": network_statistics, - } +def reduced_results(reduced_results_file_path: pathlib.Path) -> dict: + with open(reduced_results_file_path) as file_stream: + return json.load(fp=file_stream)["results"] -def test_time_results_leave_out_parameter_sets_that_did_not_run(raw_results_info): - timed_samples = raw_results_info["results"][TIMED][11] - assert extract(raw_results_info, TIMED) == {"{'name': 'A'}": timed_samples[0]} +def test_incremental_results_keep_the_successful_parameter_set(reduce, raw_results_info): + results = reduced_results(reduce(raw_results_info)) + assert results[INCREMENTAL] == {"{'name': 'A'}": {"cumulative_time_in_seconds": [0.5, 1.0, 1.5]}} -def test_rows_without_result_columns_use_the_fixed_layout(raw_results_info): - # Rows with samples have 12 entries and are read the same way; the incremental row has none, so it is left out. - assert extract(raw_results_info, NETWORK, result_columns=None) == extract(raw_results_info, NETWORK) - assert extract(raw_results_info, TIMED, result_columns=None) == extract(raw_results_info, TIMED) - assert extract(raw_results_info, INCREMENTAL, result_columns=None) == {} +def test_network_results(reduce, raw_results_info): + results = reduced_results(reduce(raw_results_info)) -def test_missing_and_skipped_rows_are_left_out(raw_results_info): - result_columns = raw_results_info["result_columns"] + assert results[NETWORK] == {"{'name': 'A'}": NETWORK_STATISTICS, "{'name': 'B'}": NETWORK_STATISTICS} - assert _extract_successful_results(test_case="x", raw_results_list=None, result_columns=result_columns) == {} - assert ( - _extract_successful_results(test_case="x", raw_results_list=[None, [["a"]]], result_columns=result_columns) - == {} - ) +def test_time_results_leave_out_parameter_sets_that_did_not_run(reduce, raw_results_info): + timed_samples = raw_results_info["results"][TIMED][11] + + results = reduced_results(reduce(raw_results_info)) -def test_length_mismatch_warns_and_is_left_out(raw_results_info): - raw_results_list = [[1.0], [["{'name': 'A'}", "{'name': 'B'}"]]] + assert results[TIMED] == {"{'name': 'A'}": timed_samples[0]} - with pytest.warns(UserWarning, match="Length mismatch"): - result = _extract_successful_results( - test_case="x", raw_results_list=raw_results_list, result_columns=raw_results_info["result_columns"] - ) - assert result == {} +def test_track_results_without_samples_are_left_out(reduce, raw_results_info): + results = reduced_results(reduce(raw_results_info)) -@pytest.mark.parametrize( - ("serialized_params", "expected"), - [ - ([], ["()"]), - ([["'a'", "'b'"]], ["'a'", "'b'"]), - ([["'a'", "'b'"], ["1"]], ["(\"'a'\", '1')", "(\"'b'\", '1')"]), - ], -) -def test_serialize_parameter_cases(serialized_params, expected): - assert _serialize_parameter_cases(serialized_params=serialized_params) == expected + assert sorted(results) == [INCREMENTAL, NETWORK, TIMED] -@pytest.fixture -def reduced_results_file_path(tmp_path, monkeypatch) -> pathlib.Path: - """Run `reduce_results` on a copy of the raw results, writing into a temporary home directory.""" - for name in ("RESULTS_DIR", "ENVIRONMENTS_DIR", "MACHINES_DIR"): - directory = tmp_path / name.lower() - directory.mkdir() - monkeypatch.setattr(_reduce_results, name, directory) - (tmp_path / "machines_dir" / "machine-test.json").write_text("{}") +def test_length_mismatch_warns_and_leaves_out_only_that_benchmark(reduce, raw_results_info): + raw_results_info["results"][NETWORK][11] = raw_results_info["results"][NETWORK][11][:1] - raw_results_file_path = tmp_path / "raw_results.json" - shutil.copy(RAW_RESULTS_FILE_PATH, raw_results_file_path) - raw_environment_info_file_path = tmp_path / "environment.txt" - raw_environment_info_file_path.write_text( - "# packages in environment at /conda/envs/nwb_benchmarks:\n" - "#\n" - "# Name Version Build Channel\n" - "pynwb 3.1.0 pypi_0 pypi\n" - ) + with pytest.warns(UserWarning, match=f"test case {NETWORK}: \n\tLength mismatch"): + results = reduced_results(reduce(raw_results_info)) - reduce_results( - machine_id="test", - raw_results_file_path=raw_results_file_path, - raw_environment_info_file_path=raw_environment_info_file_path, - ) + assert sorted(results) == [INCREMENTAL, TIMED] - assert not raw_results_file_path.exists() - (reduced_results_file_path,) = (tmp_path / "results_dir").glob("*_machine-test_results.json") - return reduced_results_file_path +def test_no_successful_results_raises(reduce, raw_results_info): + raw_results_info["results"] = {UNWRAPPED: raw_results_info["results"][UNWRAPPED]} -def test_reduce_results_writes_every_successful_result(reduced_results_file_path): - with open(reduced_results_file_path) as file_stream: + with pytest.raises(ValueError, match="failed to find any successful results"): + reduce(raw_results_info) + + +def test_reduced_results_file(reduce, raw_results_info): + with open(reduce(raw_results_info)) as file_stream: reduced_results_info = json.load(fp=file_stream) assert reduced_results_info["commit_hash"] == "0123456789abcdef0123456789abcdef01234567" assert reduced_results_info["machine_id"] == "test" - assert sorted(reduced_results_info["results"]) == [INCREMENTAL, NETWORK, TIMED] - assert reduced_results_info["results"][INCREMENTAL] == { - "{'name': 'A'}": {"cumulative_time_in_seconds": [0.5, 1.0, 1.5]} - } -def test_database_reads_the_reduced_results(reduced_results_file_path): +def test_database_reads_the_reduced_results(reduce, raw_results_info): pytest.importorskip("polars") pytest.importorskip("seaborn") from nwb_benchmarks.database import Results - results = Results.safe_load_from_json(file_path=reduced_results_file_path) - data_frame = results.to_dataframe() + data_frame = Results.safe_load_from_json(file_path=reduce(raw_results_info)).to_dataframe() incremental = data_frame.filter(data_frame["benchmark_name"] == INCREMENTAL) assert incremental["variable"].to_list() == ["cumulative_time_in_seconds"] * 3 diff --git a/tests/test_track_incremental_slicing.py b/tests/test_track_incremental_slicing.py index 615ab0d5..14fe39b1 100644 --- a/tests/test_track_incremental_slicing.py +++ b/tests/test_track_incremental_slicing.py @@ -91,6 +91,16 @@ def test_cumulative_times_for_icephys_strategy(): np.testing.assert_array_equal(benchmark._temp, np.arange(3)) +def test_tracked_result_is_wrapped_as_asv_samples(): + # ASV only writes a `track_` result to the samples column, which `reduce_results` reads, in this structure. + benchmark = InMemoryIncrementalSliceBenchmark(nwbfile=make_nwbfile()) + result = benchmark.track_cumulative_slice_times(params=dict(slice_strategy="iterate_icephys_timeseries")) + + assert list(result) == ["samples", "number"] + assert result["number"] is None + assert_cumulative_times(result=result["samples"], expected_length=1 + 3) + + def test_unsupported_strategy_raises(): benchmark = InMemoryIncrementalSliceBenchmark(nwbfile=make_nwbfile())