refactor run_benchmarks.py
11 changed files+679−486
benchcompress/src/benchcompress/cli.pymodified+1−1View file
@@ -2,7 +2,7 @@
22
33 import click
44 from typing import List, Optional
5-from .run_benchmarks import run_benchmarks
5+from .run_benchmarks.run_benchmarks import run_benchmarks
66 from .algorithms import algorithms
77 from .datasets import datasets
88
benchcompress/src/benchcompress/run_benchmarks.pydeleted+0−485View file
@@ -1,485 +0,0 @@
1-import time
2-import json
3-import os
4-from typing import Dict, Any, Tuple, List, Optional
5-from datetime import datetime
6-import numpy as np
7-from statistics import median
8-from .algorithms import algorithms
9-from .datasets import datasets
10-from ._memobin import (
11- construct_memobin_url,
12- construct_dataset_url,
13- upload_to_memobin,
14- download_from_memobin,
15- exists_in_memobin,
16-)
17-
18-
19-system_version = "v6"
20-GITHUB_ALGORITHMS_PREFIX = "https://github.com/magland/benchcompress/blob/main/benchcompress/src/benchcompress/algorithms/"
21-GITHUB_DATASETS_PREFIX = "https://github.com/magland/benchcompress/blob/main/benchcompress/src/benchcompress/datasets/"
22-
23-
24-def is_compatible(algorithm_tags: List[str], dataset_tags: List[str]) -> bool:
25- """Check if an algorithm is compatible with a dataset based on their tags.
26-
27- Args:
28- algorithm_tags: List of tags for the algorithm
29- dataset_tags: List of tags for the dataset
30-
31- Returns:
32- True if the algorithm should be applied to the dataset
33- """
34- # If algorithm has delta_encoding or markov_prediction, dataset must have continuous, timeseries, 1d, integer
35- if "delta_encoding" in algorithm_tags or "markov_prediction" in algorithm_tags:
36- if (
37- "continuous" not in dataset_tags
38- or "timeseries" not in dataset_tags
39- or "1d" not in dataset_tags
40- or "integer" not in dataset_tags
41- ):
42- return False
43-
44- # If algorithm has zero_rle, dataset must have sparse, timeseries, 1d
45- if "zero_rle" in algorithm_tags:
46- if (
47- "sparse" not in dataset_tags
48- or "timeseries" not in dataset_tags
49- or "1d" not in dataset_tags
50- ):
51- return False
52-
53- # If algorithm has integer, dataset must have integer
54- if "integer" in algorithm_tags:
55- if "integer" not in dataset_tags:
56- return False
57-
58- return True
59-
60-
61-def upload_benchmark_status(
62- memobin_api_key: str,
63- current_dataset: str,
64- current_algorithm: str,
65- completed_benchmarks: List[Dict[str, Any]],
66- total_benchmarks: int,
67- start_time: float,
68-) -> None:
69- """Upload current benchmark status to memobin.
70-
71- Args:
72- memobin_api_key: API key for memobin authentication
73- current_dataset: Name of the current dataset being processed
74- current_algorithm: Name of the current algorithm being tested
75- completed_benchmarks: List of completed benchmark results
76- total_benchmarks: Total number of benchmarks to run
77- start_time: Timestamp when the benchmark run started
78- """
79- status = {
80- "current_dataset": current_dataset,
81- "current_algorithm": current_algorithm,
82- "completed_count": len(completed_benchmarks),
83- "total_count": total_benchmarks,
84- "progress_percentage": (len(completed_benchmarks) / total_benchmarks) * 100,
85- "elapsed_time": time.time() - start_time,
86- "last_update": datetime.now().isoformat(),
87- "completed_benchmarks": completed_benchmarks,
88- }
89-
90- status_url = "https://tempory.net/f/memobin/benchmark_status/current.json"
91- upload_to_memobin(status, status_url, memobin_api_key)
92-
93-
94-def run_benchmarks(
95- cache_dir: str = ".benchmark_cache",
96- verbose: bool = True,
97- selected_algorithms: Optional[List[dict]] = None,
98- selected_datasets: Optional[List[dict]] = None,
99- force: bool = False,
100-) -> Dict[str, Any]:
101- """Run all benchmarks, with caching based on algorithm and dataset versions.
102-
103- Results are stored in separate directories for each dataset/algorithm combination:
104- cache_dir/
105- dataset_name/
106- algorithm_name/
107- metadata.json # Contains algorithm version, dataset version, and results
108- compressed.dat # The actual compressed data
109-
110- Args:
111- cache_dir: Directory to store cached results
112-
113- Returns:
114- Dictionary containing benchmark results and metadata
115- """
116- print("\n=== Starting Benchmark Run ===")
117- print(f"Cache directory: {cache_dir}")
118-
119- os.makedirs(cache_dir, exist_ok=True)
120-
121- start_time = time.time()
122- last_status_upload = 0 # Track last status upload time
123- results = []
124- print("\nRunning benchmarks for all dataset-algorithm combinations...")
125-
126- # Use selected datasets/algorithms or fall back to all
127- datasets_to_run = selected_datasets if selected_datasets is not None else datasets
128- algorithms_to_run = (
129- selected_algorithms if selected_algorithms is not None else algorithms
130- )
131-
132- # Calculate total number of benchmarks
133- total_benchmarks = sum(
134- 1
135- for dataset in datasets_to_run
136- for algorithm in algorithms_to_run
137- if is_compatible(algorithm.get("tags", []), dataset.get("tags", []))
138- )
139-
140- # Run benchmarks for each dataset and algorithm combination
141- memobin_api_key = os.environ.get("MEMOBIN_API_KEY")
142- upload_enabled = os.environ.get("UPLOAD_TO_MEMOBIN") == "1"
143-
144- for dataset in datasets_to_run:
145- dataset_tags = dataset.get("tags", [])
146- print(f"\n*** Dataset: {dataset['name']} (tags: {dataset_tags}) ***")
147-
148- # only create the dataset if it is needed
149- data = None
150-
151- for algorithm in algorithms_to_run:
152- alg_name = algorithm["name"]
153- alg_tags = algorithm.get("tags", [])
154-
155- # Skip if algorithm and dataset are not compatible based on tags
156- if not is_compatible(alg_tags, dataset_tags):
157- if verbose:
158- print(
159- f"\nSkipping algorithm {alg_name} (tags: {alg_tags}) - incompatible with dataset tags"
160- )
161- continue
162-
163- print(f"\nTesting algorithm: {alg_name} (tags: {alg_tags})")
164-
165- # Upload current status to memobin if enabled (once per minute)
166- current_time = time.time()
167- if (
168- memobin_api_key
169- and upload_enabled
170- and (current_time - last_status_upload >= 60)
171- ): # Check if 60 seconds have passed
172- try:
173- upload_benchmark_status(
174- memobin_api_key,
175- dataset["name"],
176- alg_name,
177- results,
178- total_benchmarks,
179- start_time,
180- )
181- last_status_upload = current_time # Update last upload time
182- except Exception as e:
183- print(f" Warning: Failed to upload status to memobin: {str(e)}")
184-
185- # Check if we can use cached result (unless force flag is set)
186- test_dir = os.path.join(cache_dir, dataset["name"], alg_name)
187- metadata_file = os.path.join(test_dir, "metadata.json")
188- compressed_file = os.path.join(test_dir, "compressed.dat")
189-
190- # First try local cache (unless force flag is set)
191- cached_data = None
192- if not force and os.path.exists(metadata_file):
193- with open(metadata_file, "r") as f:
194- cached_data = json.load(f)
195- # if versions do not match, then set to None
196- if isinstance(cached_data, dict) and "result" in cached_data:
197- result = cached_data["result"]
198- if (
199- result["algorithm_version"] != algorithm["version"]
200- or result["dataset_version"] != dataset["version"]
201- or result.get("system_version", "") != system_version
202- ):
203- cached_data = None
204-
205- # If not in local cache, try memobin (unless force flag is set)
206- if cached_data is None and not force:
207- memobin_url = construct_memobin_url(
208- alg_name,
209- dataset["name"],
210- algorithm["version"],
211- dataset["version"],
212- system_version,
213- "metadata.json",
214- )
215- if verbose:
216- print(" Looking for cached result in memobin...")
217- cached_data = download_from_memobin(memobin_url)
218- if cached_data is not None:
219- if verbose:
220- print(" Found result in memobin, saving locally...")
221- # Save to local cache
222- os.makedirs(test_dir, exist_ok=True)
223- with open(metadata_file, "w") as f:
224- json.dump(cached_data, f, indent=2)
225-
226- if (
227- cached_data is not None
228- and isinstance(cached_data, dict)
229- and "result" in cached_data
230- ):
231- result = cached_data["result"]
232- if (
233- isinstance(result, dict)
234- and result.get("algorithm_version") == algorithm["version"]
235- and result.get("dataset_version") == dataset["version"]
236- and result.get("system_version", "") == system_version
237- ):
238- print(" Using cached result:")
239- result["cache_status"] = "cached"
240- results.append(result)
241- continue
242-
243- print(f" Running benchmark for {alg_name} on {dataset['name']}...")
244- if data is None:
245- data = dataset["create"]()
246- print(f"Created dataset: shape={data.shape}, dtype={data.dtype}")
247- else:
248- print("Dataset already created")
249- dtype = str(data.dtype)
250- original_size = len(data.tobytes())
251- print(f"Dataset: shape={data.shape}, dtype={dtype}")
252- print(f"Original size: {original_size:,} bytes")
253-
254- # Upload dataset to memobin if enabled
255- memobin_api_key = os.environ.get("MEMOBIN_API_KEY")
256- upload_enabled = os.environ.get("UPLOAD_TO_MEMOBIN") == "1"
257- if memobin_api_key and upload_enabled:
258- try:
259- # Upload array metadata as JSON
260- dataset_url_json = construct_dataset_url(
261- dataset["name"], dataset["version"], "json"
262- )
263- if not exists_in_memobin(dataset_url_json):
264- if verbose:
265- print(" Uploading dataset metadata to memobin...")
266- metadata = {"dtype": str(data.dtype), "shape": data.shape}
267- upload_to_memobin(
268- metadata,
269- dataset_url_json,
270- memobin_api_key,
271- content_type="application/json",
272- )
273- if verbose:
274- print(" Successfully uploaded metadata")
275-
276- # Upload raw .dat format
277- dataset_url_raw = construct_dataset_url(
278- dataset["name"], dataset["version"], "dat"
279- )
280- if not exists_in_memobin(dataset_url_raw):
281- if verbose:
282- print(" Uploading dataset (raw) to memobin...")
283- upload_to_memobin(
284- data.tobytes(),
285- dataset_url_raw,
286- memobin_api_key,
287- content_type="application/octet-stream",
288- )
289- if verbose:
290- print(" Successfully uploaded raw dataset")
291-
292- # Upload .npy format
293- dataset_url_npy = construct_dataset_url(
294- dataset["name"], dataset["version"], "npy"
295- )
296- if not exists_in_memobin(dataset_url_npy):
297- if verbose:
298- print(" Uploading dataset (npy) to memobin...")
299- # Save array to a temporary .npy file
300- temp_npy = os.path.join(cache_dir, "temp.npy")
301- np.save(temp_npy, data)
302- with open(temp_npy, "rb") as f:
303- npy_bytes = f.read()
304- os.remove(temp_npy) # Clean up temp file
305-
306- upload_to_memobin(
307- npy_bytes,
308- dataset_url_npy,
309- memobin_api_key,
310- content_type="application/octet-stream",
311- )
312- if verbose:
313- print(" Successfully uploaded npy dataset")
314- except Exception as e:
315- print(f" Warning: Failed to upload dataset to memobin: {str(e)}")
316-
317- assert data is not None
318- assert isinstance(data, np.ndarray)
319- assert isinstance(original_size, int)
320- assert isinstance(dtype, str)
321-
322- def run_timed_trials(operation, *args) -> Tuple[float, float, Any]:
323- """Run multiple trials of an operation until total time exceeds 1 second.
324- Returns (median_time, mb_per_sec)"""
325- assert data is not None
326- assert isinstance(data, np.ndarray)
327- times = []
328- total_time = 0
329- array_size_mb = data.nbytes / (1024 * 1024) # Convert to MB
330-
331- ret = None
332- while total_time < 1.0:
333- start_time = time.perf_counter()
334- ret = operation(*args) # Execute operation but discard result
335- trial_time = time.perf_counter() - start_time
336- times.append(trial_time)
337- total_time += trial_time
338-
339- median_time = median(times)
340- mb_per_sec = array_size_mb / median_time
341- return median_time, mb_per_sec, ret
342-
343- print(" Encoding...")
344- encode_time, encode_mb_per_sec, encoded = run_timed_trials(
345- algorithm["encode"], data
346- )
347- compressed_size = len(encoded)
348- compression_ratio = original_size / compressed_size
349- print(" Compression complete:")
350- print(f" Compressed size: {compressed_size:,} bytes")
351- print(f" Compression ratio: {compression_ratio:.2f}x")
352- print(f" Encode time: {encode_time*1000:.2f}ms")
353- print(f" Encode throughput: {encode_mb_per_sec:.2f} MB/s")
354-
355- print(" Decoding...")
356- # Measure decoding with multiple trials
357- decode_time, decode_mb_per_sec, decoded = run_timed_trials(
358- algorithm["decode"], encoded, dtype, data.shape
359- )
360- print(f" Decode time: {decode_time*1000:.2f}ms")
361- print(f" Decode throughput: {decode_mb_per_sec:.2f} MB/s")
362-
363- if len(data) != len(decoded):
364- raise ValueError(
365- f"Decompression failed: decoded length {len(decoded)} != original length {len(data)}"
366- )
367-
368- # Verify correctness
369- if not np.array_equal(data, decoded):
370- print(data[:100])
371- print(decoded[:100])
372- for j in range(len(data)):
373- if data[j] != decoded[j]:
374- print(f"Error at index {j}: {data[j]} != {decoded[j]}")
375- break
376- raise ValueError(
377- f"Decompression verification failed for {alg_name} on {dataset['name']}"
378- )
379- print(" Verification successful!")
380-
381- # Store result
382- result = {
383- "dataset": dataset["name"],
384- "algorithm": alg_name,
385- "algorithm_version": algorithm["version"],
386- "dataset_version": dataset["version"],
387- "system_version": system_version,
388- "compression_ratio": compression_ratio,
389- "encode_time": encode_time,
390- "decode_time": decode_time,
391- "encode_mb_per_sec": encode_mb_per_sec,
392- "decode_mb_per_sec": decode_mb_per_sec,
393- "original_size": original_size,
394- "compressed_size": compressed_size,
395- "array_shape": data.shape,
396- "array_dtype": dtype,
397- "timestamp": time.time(),
398- "cache_status": "new",
399- }
400- results.append(result)
401-
402- # Save result and compressed data
403- os.makedirs(test_dir, exist_ok=True)
404- cache_data = {"result": result}
405- with open(metadata_file, "w") as f:
406- json.dump(cache_data, f, indent=2)
407- with open(compressed_file, "wb") as f:
408- f.write(encoded)
409- print(f" Results saved to: {test_dir}")
410-
411- # Upload to memobin if API key is set and upload is enabled
412- memobin_api_key = os.environ.get("MEMOBIN_API_KEY")
413- upload_enabled = os.environ.get("UPLOAD_TO_MEMOBIN") == "1"
414- if memobin_api_key and upload_enabled:
415- if verbose:
416- print(" Uploading results to memobin...")
417- try:
418- memobin_url = construct_memobin_url(
419- alg_name,
420- dataset["name"],
421- algorithm["version"],
422- dataset["version"],
423- system_version,
424- )
425- upload_to_memobin(
426- cache_data,
427- memobin_url,
428- memobin_api_key,
429- )
430- if verbose:
431- print(" Successfully uploaded to memobin")
432- except Exception as e:
433- print(f" Warning: Failed to upload to memobin: {str(e)}")
434-
435- print("\n=== Benchmark Run Complete ===\n")
436-
437- # Collect algorithm and dataset information as lists
438- algorithm_info = []
439- for algorithm in algorithms:
440- info = {
441- "name": algorithm["name"],
442- "description": algorithm.get("description", ""),
443- "version": algorithm["version"],
444- "tags": algorithm.get("tags", []),
445- }
446- if "source_file" in algorithm:
447- info["source_file"] = GITHUB_ALGORITHMS_PREFIX + algorithm["source_file"]
448- algorithm_info.append(info)
449-
450- dataset_info = []
451- for dataset in datasets:
452- info = {
453- "name": dataset["name"],
454- "description": dataset.get("description", ""),
455- "version": dataset["version"],
456- "tags": dataset.get("tags", []),
457- "data_url_raw": construct_dataset_url(
458- dataset["name"], dataset["version"], "dat"
459- ),
460- "data_url_npy": construct_dataset_url(
461- dataset["name"], dataset["version"], "npy"
462- ),
463- "data_url_json": construct_dataset_url(
464- dataset["name"], dataset["version"], "json"
465- ),
466- }
467- if "source_file" in dataset:
468- info["source_file"] = GITHUB_DATASETS_PREFIX + dataset["source_file"]
469- dataset_info.append(info)
470-
471- # Upload final benchmark status
472- if memobin_api_key and upload_enabled:
473- try:
474- upload_benchmark_status(
475- memobin_api_key,
476- "All datasets",
477- "All algorithms",
478- results,
479- total_benchmarks,
480- start_time,
481- )
482- except Exception as e:
483- print(f" Warning: Failed to upload final status to memobin: {str(e)}")
484-
485- return {"results": results, "algorithms": algorithm_info, "datasets": dataset_info}
benchcompress/src/benchcompress/run_benchmarks/__init__.pyadded+1−0View file
@@ -0,0 +1 @@
1+from .run_benchmarks import run_benchmarks
benchcompress/src/benchcompress/_memobin.py →benchcompress/src/benchcompress/run_benchmarks/_memobin.pyrenamed+0−0View file
No changes to the file's content.
benchcompress/src/benchcompress/run_benchmarks/benchmark_timing.pyadded+118−0View file
@@ -0,0 +1,118 @@
1+from typing import Any, Tuple, Callable, Dict
2+from statistics import median
3+import time
4+import numpy as np
5+
6+
7+def run_timed_trials(
8+ data: np.ndarray, operation: Callable, *args
9+) -> Tuple[float, float, Any]:
10+ """Run multiple trials of an operation until total time exceeds 1 second.
11+
12+ Args:
13+ data: Input numpy array for calculating throughput
14+ operation: Function to benchmark
15+ *args: Arguments to pass to the operation
16+
17+ Returns:
18+ Tuple containing:
19+ - median_time: Median execution time across trials
20+ - mb_per_sec: Throughput in MB/s
21+ - result: Result from the last trial execution
22+ """
23+ times = []
24+ total_time = 0
25+ array_size_mb = data.nbytes / (1024 * 1024) # Convert to MB
26+
27+ ret = None
28+ while total_time < 1.0:
29+ start_time = time.perf_counter()
30+ ret = operation(*args) # Execute operation
31+ trial_time = time.perf_counter() - start_time
32+ times.append(trial_time)
33+ total_time += trial_time
34+
35+ median_time = median(times)
36+ mb_per_sec = array_size_mb / median_time
37+ return median_time, mb_per_sec, ret
38+
39+
40+def run_compression_benchmark(
41+ data: np.ndarray,
42+ algorithm_name: str,
43+ encode_fn: Callable,
44+ decode_fn: Callable,
45+ verbose: bool = True,
46+) -> Tuple[Dict[str, Any], bytes]:
47+ """Run compression and decompression benchmarks for an algorithm.
48+
49+ Args:
50+ data: Input numpy array to compress
51+ algorithm_name: Name of the algorithm being benchmarked
52+ encode_fn: Compression function
53+ decode_fn: Decompression function
54+ verbose: Whether to print progress messages
55+
56+ Returns:
57+ Tuple containing:
58+ - result: Dictionary with benchmark metrics
59+ - encoded: Compressed data bytes
60+ """
61+ original_size = len(data.tobytes())
62+ dtype = str(data.dtype)
63+
64+ if verbose:
65+ print(" Encoding...")
66+ encode_time, encode_mb_per_sec, encoded = run_timed_trials(data, encode_fn, data)
67+ compressed_size = len(encoded)
68+ compression_ratio = original_size / compressed_size
69+
70+ if verbose:
71+ print(" Compression complete:")
72+ print(f" Compressed size: {compressed_size:,} bytes")
73+ print(f" Compression ratio: {compression_ratio:.2f}x")
74+ print(f" Encode time: {encode_time*1000:.2f}ms")
75+ print(f" Encode throughput: {encode_mb_per_sec:.2f} MB/s")
76+ print(" Decoding...")
77+
78+ decode_time, decode_mb_per_sec, decoded = run_timed_trials(
79+ data, decode_fn, encoded, dtype, data.shape
80+ )
81+
82+ if verbose:
83+ print(f" Decode time: {decode_time*1000:.2f}ms")
84+ print(f" Decode throughput: {decode_mb_per_sec:.2f} MB/s")
85+
86+ # Verify correctness
87+ if len(data) != len(decoded):
88+ raise ValueError(
89+ f"Decompression failed: decoded length {len(decoded)} != original length {len(data)}"
90+ )
91+
92+ if not np.array_equal(data, decoded):
93+ print(data[:100])
94+ print(decoded[:100])
95+ for j in range(len(data)):
96+ if data[j] != decoded[j]:
97+ print(f"Error at index {j}: {data[j]} != {decoded[j]}")
98+ break
99+ raise ValueError(f"Decompression verification failed for {algorithm_name}")
100+
101+ if verbose:
102+ print(" Verification successful!")
103+
104+ result = {
105+ "compression_ratio": compression_ratio,
106+ "encode_time": encode_time,
107+ "decode_time": decode_time,
108+ "encode_mb_per_sec": encode_mb_per_sec,
109+ "decode_mb_per_sec": decode_mb_per_sec,
110+ "original_size": original_size,
111+ "compressed_size": compressed_size,
112+ "array_shape": data.shape,
113+ "array_dtype": dtype,
114+ "timestamp": time.time(),
115+ "cache_status": "new",
116+ }
117+
118+ return result, encoded
benchcompress/src/benchcompress/run_benchmarks/cache_management.pyadded+118−0View file
@@ -0,0 +1,118 @@
1+import os
2+import json
3+from typing import Optional, Dict, Any
4+from ._memobin import (
5+ construct_memobin_url,
6+ download_from_memobin,
7+)
8+
9+
10+def check_cached_result(
11+ cache_dir: str,
12+ dataset_name: str,
13+ algorithm_name: str,
14+ algorithm_version: str,
15+ dataset_version: str,
16+ system_version: str,
17+ force: bool = False,
18+ verbose: bool = True,
19+) -> Optional[Dict[str, Any]]:
20+ """Check for cached benchmark results locally and in memobin.
21+
22+ Args:
23+ cache_dir: Directory containing cached results
24+ dataset_name: Name of the dataset
25+ algorithm_name: Name of the algorithm
26+ algorithm_version: Version of the algorithm
27+ dataset_version: Version of the dataset
28+ system_version: Version of the system
29+ force: If True, ignore cached results
30+ verbose: Whether to print progress messages
31+
32+ Returns:
33+ Cached result dictionary if found and valid, None otherwise
34+ """
35+ test_dir = os.path.join(cache_dir, dataset_name, algorithm_name)
36+ metadata_file = os.path.join(test_dir, "metadata.json")
37+
38+ # First try local cache (unless force flag is set)
39+ cached_data = None
40+ if not force and os.path.exists(metadata_file):
41+ with open(metadata_file, "r") as f:
42+ cached_data = json.load(f)
43+ # if versions do not match, then set to None
44+ if isinstance(cached_data, dict) and "result" in cached_data:
45+ result = cached_data["result"]
46+ if (
47+ result["algorithm_version"] != algorithm_version
48+ or result["dataset_version"] != dataset_version
49+ or result.get("system_version", "") != system_version
50+ ):
51+ cached_data = None
52+
53+ # If not in local cache, try memobin (unless force flag is set)
54+ if cached_data is None and not force:
55+ memobin_url = construct_memobin_url(
56+ algorithm_name,
57+ dataset_name,
58+ algorithm_version,
59+ dataset_version,
60+ system_version,
61+ "metadata.json",
62+ )
63+ if verbose:
64+ print(" Looking for cached result in memobin...")
65+ cached_data = download_from_memobin(memobin_url)
66+ if cached_data is not None:
67+ if verbose:
68+ print(" Found result in memobin, saving locally...")
69+ # Save to local cache
70+ os.makedirs(test_dir, exist_ok=True)
71+ with open(metadata_file, "w") as f:
72+ json.dump(cached_data, f, indent=2)
73+
74+ if (
75+ cached_data is not None
76+ and isinstance(cached_data, dict)
77+ and "result" in cached_data
78+ ):
79+ result = cached_data["result"]
80+ if (
81+ isinstance(result, dict)
82+ and result.get("algorithm_version") == algorithm_version
83+ and result.get("dataset_version") == dataset_version
84+ and result.get("system_version", "") == system_version
85+ ):
86+ result["cache_status"] = "cached"
87+ return result
88+
89+ return None
90+
91+
92+def save_result_to_cache(
93+ result: Dict[str, Any],
94+ encoded_data: bytes,
95+ cache_dir: str,
96+ dataset_name: str,
97+ algorithm_name: str,
98+) -> None:
99+ """Save benchmark result and compressed data to cache.
100+
101+ Args:
102+ result: Benchmark result dictionary
103+ encoded_data: Compressed data bytes
104+ cache_dir: Directory to store cached results
105+ dataset_name: Name of the dataset
106+ algorithm_name: Name of the algorithm
107+ """
108+ test_dir = os.path.join(cache_dir, dataset_name, algorithm_name)
109+ metadata_file = os.path.join(test_dir, "metadata.json")
110+ compressed_file = os.path.join(test_dir, "compressed.dat")
111+
112+ os.makedirs(test_dir, exist_ok=True)
113+ cache_data = {"result": result}
114+
115+ with open(metadata_file, "w") as f:
116+ json.dump(cache_data, f, indent=2)
117+ with open(compressed_file, "wb") as f:
118+ f.write(encoded_data)
benchcompress/src/benchcompress/run_benchmarks/collect_info.pyadded+60−0View file
@@ -0,0 +1,60 @@
1+from typing import List, Dict, Any
2+from ._memobin import construct_dataset_url
3+
4+GITHUB_ALGORITHMS_PREFIX = "https://github.com/magland/benchcompress/blob/main/benchcompress/src/benchcompress/algorithms/"
5+GITHUB_DATASETS_PREFIX = "https://github.com/magland/benchcompress/blob/main/benchcompress/src/benchcompress/datasets/"
6+
7+
8+def collect_algorithm_info(algorithms: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
9+ """Collect information about compression algorithms.
10+
11+ Args:
12+ algorithms: List of algorithm dictionaries
13+
14+ Returns:
15+ List of algorithm information dictionaries
16+ """
17+ algorithm_info = []
18+ for algorithm in algorithms:
19+ info = {
20+ "name": algorithm["name"],
21+ "description": algorithm.get("description", ""),
22+ "version": algorithm["version"],
23+ "tags": algorithm.get("tags", []),
24+ }
25+ if "source_file" in algorithm:
26+ info["source_file"] = GITHUB_ALGORITHMS_PREFIX + algorithm["source_file"]
27+ algorithm_info.append(info)
28+ return algorithm_info
29+
30+
31+def collect_dataset_info(datasets: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
32+ """Collect information about benchmark datasets.
33+
34+ Args:
35+ datasets: List of dataset dictionaries
36+
37+ Returns:
38+ List of dataset information dictionaries
39+ """
40+ dataset_info = []
41+ for dataset in datasets:
42+ info = {
43+ "name": dataset["name"],
44+ "description": dataset.get("description", ""),
45+ "version": dataset["version"],
46+ "tags": dataset.get("tags", []),
47+ "data_url_raw": construct_dataset_url(
48+ dataset["name"], dataset["version"], "dat"
49+ ),
50+ "data_url_npy": construct_dataset_url(
51+ dataset["name"], dataset["version"], "npy"
52+ ),
53+ "data_url_json": construct_dataset_url(
54+ dataset["name"], dataset["version"], "json"
55+ ),
56+ }
57+ if "source_file" in dataset:
58+ info["source_file"] = GITHUB_DATASETS_PREFIX + dataset["source_file"]
59+ dataset_info.append(info)
60+ return dataset_info
benchcompress/src/benchcompress/run_benchmarks/is_compatible.pyadded+38−0View file
@@ -0,0 +1,38 @@
1+from typing import List
2+
3+
4+def is_compatible(algorithm_tags: List[str], dataset_tags: List[str]) -> bool:
5+ """Check if an algorithm is compatible with a dataset based on their tags.
6+
7+ Args:
8+ algorithm_tags: List of tags for the algorithm
9+ dataset_tags: List of tags for the dataset
10+
11+ Returns:
12+ True if the algorithm should be applied to the dataset
13+ """
14+ # If algorithm has delta_encoding or markov_prediction, dataset must have continuous, timeseries, 1d, integer
15+ if "delta_encoding" in algorithm_tags or "markov_prediction" in algorithm_tags:
16+ if (
17+ "continuous" not in dataset_tags
18+ or "timeseries" not in dataset_tags
19+ or "1d" not in dataset_tags
20+ or "integer" not in dataset_tags
21+ ):
22+ return False
23+
24+ # If algorithm has zero_rle, dataset must have sparse, timeseries, 1d
25+ if "zero_rle" in algorithm_tags:
26+ if (
27+ "sparse" not in dataset_tags
28+ or "timeseries" not in dataset_tags
29+ or "1d" not in dataset_tags
30+ ):
31+ return False
32+
33+ # If algorithm has integer, dataset must have integer
34+ if "integer" in algorithm_tags:
35+ if "integer" not in dataset_tags:
36+ return False
37+
38+ return True
benchcompress/src/benchcompress/run_benchmarks/run_benchmarks.pyadded+225−0View file
@@ -0,0 +1,225 @@
1+import os
2+import time
3+from typing import Dict, Any, List, Optional
4+import numpy as np
5+
6+from ..algorithms import algorithms
7+from ..datasets import datasets
8+from ._memobin import construct_memobin_url, upload_to_memobin
9+from .upload_dataset import upload_dataset_to_memobin
10+from .cache_management import check_cached_result, save_result_to_cache
11+from .benchmark_timing import run_compression_benchmark
12+from .collect_info import collect_algorithm_info, collect_dataset_info
13+from .is_compatible import is_compatible
14+from .upload_benchmark_status import upload_benchmark_status
15+
16+system_version = "v6"
17+
18+
19+def run_benchmarks(
20+ cache_dir: str = ".benchmark_cache",
21+ verbose: bool = True,
22+ selected_algorithms: Optional[List[dict]] = None,
23+ selected_datasets: Optional[List[dict]] = None,
24+ force: bool = False,
25+) -> Dict[str, Any]:
26+ """Run all benchmarks, with caching based on algorithm and dataset versions.
27+
28+ Results are stored in separate directories for each dataset/algorithm combination:
29+ cache_dir/
30+ dataset_name/
31+ algorithm_name/
32+ metadata.json # Contains algorithm version, dataset version, and results
33+ compressed.dat # The actual compressed data
34+
35+ Args:
36+ cache_dir: Directory to store cached results
37+ verbose: Whether to print progress messages
38+ selected_algorithms: Optional list of specific algorithms to run
39+ selected_datasets: Optional list of specific datasets to run
40+ force: If True, ignore cached results
41+
42+ Returns:
43+ Dictionary containing benchmark results and metadata
44+ """
45+ print("\n=== Starting Benchmark Run ===")
46+ print(f"Cache directory: {cache_dir}")
47+
48+ os.makedirs(cache_dir, exist_ok=True)
49+
50+ start_time = time.time()
51+ last_status_upload = 0 # Track last status upload time
52+ results = []
53+ print("\nRunning benchmarks for all dataset-algorithm combinations...")
54+
55+ # Use selected datasets/algorithms or fall back to all
56+ datasets_to_run = selected_datasets if selected_datasets is not None else datasets
57+ algorithms_to_run = (
58+ selected_algorithms if selected_algorithms is not None else algorithms
59+ )
60+
61+ # Calculate total number of benchmarks
62+ total_benchmarks = sum(
63+ 1
64+ for dataset in datasets_to_run
65+ for algorithm in algorithms_to_run
66+ if is_compatible(algorithm.get("tags", []), dataset.get("tags", []))
67+ )
68+
69+ # Run benchmarks for each dataset and algorithm combination
70+ memobin_api_key = os.environ.get("MEMOBIN_API_KEY")
71+ upload_enabled = os.environ.get("UPLOAD_TO_MEMOBIN") == "1"
72+
73+ for dataset in datasets_to_run:
74+ dataset_tags = dataset.get("tags", [])
75+ print(f"\n*** Dataset: {dataset['name']} (tags: {dataset_tags}) ***")
76+
77+ # only create the dataset if it is needed
78+ data = None
79+
80+ for algorithm in algorithms_to_run:
81+ alg_name = algorithm["name"]
82+ alg_tags = algorithm.get("tags", [])
83+
84+ # Skip if algorithm and dataset are not compatible based on tags
85+ if not is_compatible(alg_tags, dataset_tags):
86+ if verbose:
87+ print(
88+ f"\nSkipping algorithm {alg_name} (tags: {alg_tags}) - incompatible with dataset tags"
89+ )
90+ continue
91+
92+ print(f"\nTesting algorithm: {alg_name} on dataset: {dataset['name']}")
93+
94+ # Upload current status to memobin if enabled (once per minute)
95+ current_time = time.time()
96+ if (
97+ memobin_api_key
98+ and upload_enabled
99+ and (current_time - last_status_upload >= 60)
100+ ): # Check if 60 seconds have passed
101+ try:
102+ upload_benchmark_status(
103+ memobin_api_key,
104+ dataset["name"],
105+ alg_name,
106+ results,
107+ total_benchmarks,
108+ start_time,
109+ )
110+ last_status_upload = current_time # Update last upload time
111+ except Exception as e:
112+ print(f" Warning: Failed to upload status to memobin: {str(e)}")
113+
114+ # Check if we can use cached result
115+ cached_result = check_cached_result(
116+ cache_dir,
117+ dataset["name"],
118+ alg_name,
119+ algorithm["version"],
120+ dataset["version"],
121+ system_version,
122+ force,
123+ verbose,
124+ )
125+
126+ if cached_result is not None:
127+ print(" Using cached result")
128+ results.append(cached_result)
129+ continue
130+
131+ print(f" Running benchmark for {alg_name} on {dataset['name']}...")
132+ if data is None:
133+ data = dataset["create"]()
134+ print(f"Created dataset: shape={data.shape}, dtype={data.dtype}")
135+ else:
136+ print("Dataset already created")
137+
138+ # Upload dataset to memobin if enabled
139+ if memobin_api_key and upload_enabled:
140+ try:
141+ upload_dataset_to_memobin(
142+ data,
143+ dataset["name"],
144+ dataset["version"],
145+ memobin_api_key,
146+ cache_dir,
147+ verbose,
148+ )
149+ except Exception as e:
150+ print(f" Warning: Failed to upload dataset to memobin: {str(e)}")
151+
152+ # Run the benchmark
153+ result, encoded = run_compression_benchmark(
154+ data,
155+ alg_name,
156+ algorithm["encode"],
157+ algorithm["decode"],
158+ verbose,
159+ )
160+
161+ # Add metadata to result
162+ result.update(
163+ {
164+ "dataset": dataset["name"],
165+ "algorithm": alg_name,
166+ "algorithm_version": algorithm["version"],
167+ "dataset_version": dataset["version"],
168+ "system_version": system_version,
169+ }
170+ )
171+ results.append(result)
172+
173+ # Save result and compressed data
174+ save_result_to_cache(
175+ result,
176+ encoded,
177+ cache_dir,
178+ dataset["name"],
179+ alg_name,
180+ )
181+ print(
182+ f" Results saved to: {os.path.join(cache_dir, dataset['name'], alg_name)}"
183+ )
184+
185+ # Upload to memobin if enabled
186+ if memobin_api_key and upload_enabled:
187+ try:
188+ memobin_url = construct_memobin_url(
189+ alg_name,
190+ dataset["name"],
191+ algorithm["version"],
192+ dataset["version"],
193+ system_version,
194+ )
195+ upload_to_memobin(
196+ {"result": result},
197+ memobin_url,
198+ memobin_api_key,
199+ )
200+ if verbose:
201+ print(" Successfully uploaded to memobin")
202+ except Exception as e:
203+ print(f" Warning: Failed to upload to memobin: {str(e)}")
204+
205+ print("\n=== Benchmark Run Complete ===\n")
206+
207+ # Collect algorithm and dataset information
208+ algorithm_info = collect_algorithm_info(algorithms)
209+ dataset_info = collect_dataset_info(datasets)
210+
211+ # Upload final benchmark status
212+ if memobin_api_key and upload_enabled:
213+ try:
214+ upload_benchmark_status(
215+ memobin_api_key,
216+ "All datasets",
217+ "All algorithms",
218+ results,
219+ total_benchmarks,
220+ start_time,
221+ )
222+ except Exception as e:
223+ print(f" Warning: Failed to upload final status to memobin: {str(e)}")
224+
225+ return {"results": results, "algorithms": algorithm_info, "datasets": dataset_info}
benchcompress/src/benchcompress/run_benchmarks/upload_benchmark_status.pyadded+39−0View file
@@ -0,0 +1,39 @@
1+from typing import Any, Dict, List
2+import time
3+from datetime import datetime
4+from ._memobin import (
5+ upload_to_memobin,
6+)
7+
8+
9+def upload_benchmark_status(
10+ memobin_api_key: str,
11+ current_dataset: str,
12+ current_algorithm: str,
13+ completed_benchmarks: List[Dict[str, Any]],
14+ total_benchmarks: int,
15+ start_time: float,
16+) -> None:
17+ """Upload current benchmark status to memobin.
18+
19+ Args:
20+ memobin_api_key: API key for memobin authentication
21+ current_dataset: Name of the current dataset being processed
22+ current_algorithm: Name of the current algorithm being tested
23+ completed_benchmarks: List of completed benchmark results
24+ total_benchmarks: Total number of benchmarks to run
25+ start_time: Timestamp when the benchmark run started
26+ """
27+ status = {
28+ "current_dataset": current_dataset,
29+ "current_algorithm": current_algorithm,
30+ "completed_count": len(completed_benchmarks),
31+ "total_count": total_benchmarks,
32+ "progress_percentage": (len(completed_benchmarks) / total_benchmarks) * 100,
33+ "elapsed_time": time.time() - start_time,
34+ "last_update": datetime.now().isoformat(),
35+ "completed_benchmarks": completed_benchmarks,
36+ }
37+
38+ status_url = "https://tempory.net/f/memobin/benchmark_status/current.json"
39+ upload_to_memobin(status, status_url, memobin_api_key)
benchcompress/src/benchcompress/run_benchmarks/upload_dataset.pyadded+79−0View file
@@ -0,0 +1,79 @@
1+import os
2+import numpy as np
3+from ._memobin import (
4+ construct_dataset_url,
5+ exists_in_memobin,
6+ upload_to_memobin,
7+)
8+
9+
10+def upload_dataset_to_memobin(
11+ data: np.ndarray,
12+ dataset_name: str,
13+ dataset_version: str,
14+ memobin_api_key: str,
15+ cache_dir: str,
16+ verbose: bool = True,
17+) -> None:
18+ """Upload dataset to memobin in multiple formats.
19+
20+ Args:
21+ data: The numpy array dataset to upload
22+ dataset_name: Name of the dataset
23+ dataset_version: Version of the dataset
24+ memobin_api_key: API key for memobin
25+ cache_dir: Directory for temporary files
26+ verbose: Whether to print progress messages
27+ """
28+ try:
29+ # Upload array metadata as JSON
30+ dataset_url_json = construct_dataset_url(dataset_name, dataset_version, "json")
31+ if not exists_in_memobin(dataset_url_json):
32+ if verbose:
33+ print(" Uploading dataset metadata to memobin...")
34+ metadata = {"dtype": str(data.dtype), "shape": data.shape}
35+ upload_to_memobin(
36+ metadata,
37+ dataset_url_json,
38+ memobin_api_key,
39+ content_type="application/json",
40+ )
41+ if verbose:
42+ print(" Successfully uploaded metadata")
43+
44+ # Upload raw .dat format
45+ dataset_url_raw = construct_dataset_url(dataset_name, dataset_version, "dat")
46+ if not exists_in_memobin(dataset_url_raw):
47+ if verbose:
48+ print(" Uploading dataset (raw) to memobin...")
49+ upload_to_memobin(
50+ data.tobytes(),
51+ dataset_url_raw,
52+ memobin_api_key,
53+ content_type="application/octet-stream",
54+ )
55+ if verbose:
56+ print(" Successfully uploaded raw dataset")
57+
58+ # Upload .npy format
59+ dataset_url_npy = construct_dataset_url(dataset_name, dataset_version, "npy")
60+ if not exists_in_memobin(dataset_url_npy):
61+ if verbose:
62+ print(" Uploading dataset (npy) to memobin...")
63+ # Save array to a temporary .npy file
64+ temp_npy = os.path.join(cache_dir, "temp.npy")
65+ np.save(temp_npy, data)
66+ with open(temp_npy, "rb") as f:
67+ npy_bytes = f.read()
68+ os.remove(temp_npy) # Clean up temp file
69+
70+ upload_to_memobin(
71+ npy_bytes,
72+ dataset_url_npy,
73+ memobin_api_key,
74+ content_type="application/octet-stream",
75+ )
76+ if verbose:
77+ print(" Successfully uploaded npy dataset")
78+ except Exception as e:
79+ print(f" Warning: Failed to upload dataset to memobin: {str(e)}")