1#!/usr/bin/env python3
3import click
4from typing import List, Optional
5from .run_benchmarks.run_benchmarks import run_benchmarks
6from .algorithms import algorithms
7from .datasets import datasets
10def get_available_algorithms() -> List[str]:
11 """Get list of available algorithm names"""
12 return [alg.name for alg in algorithms]
15def get_available_datasets() -> List[str]:
16 """Get list of available dataset names"""
17 return [ds.name for ds in datasets]
20def filter_algorithms(selected: Optional[List[str]] = None) -> List[dict]:
21 """Filter algorithms based on selected names"""
22 if not selected:
23 return algorithms
24 return [alg for alg in algorithms if alg.name in selected]
27def filter_datasets(selected: Optional[List[str]] = None) -> List[dict]:
28 """Filter datasets based on selected names"""
29 if not selected:
30 return datasets
31 return [ds for ds in datasets if ds.name in selected]
34def validate_algorithms(ctx, param, value):
35 if not value:
36 return None
37 available = get_available_algorithms()
38 invalid = [alg for alg in value if alg not in available]
39 if invalid:
40 raise click.BadParameter(
41 f"Invalid algorithm(s): {', '.join(invalid)}. "
42 f"Available algorithms: {', '.join(available)}"
43 )
44 return value
47def validate_datasets(ctx, param, value):
48 if not value:
49 return None
50 available = get_available_datasets()
51 invalid = [ds for ds in value if ds not in available]
52 if invalid:
53 raise click.BadParameter(
54 f"Invalid dataset(s): {', '.join(invalid)}. "
55 f"Available datasets: {', '.join(available)}"
56 )
57 return value
61def cli():
62 """Benchmark compression algorithms for electrophysiology data"""
63 pass
67def list():
68 """List available algorithms and datasets"""
69 click.echo("\nAvailable Algorithms:")
70 for alg in algorithms:
71 desc = alg.description if alg.description else "No description"
72 click.echo(f" {alg.name:<20} - {desc}")
74 click.echo("\nAvailable Datasets:")
75 for ds in datasets:
76 desc = ds.description if ds.description else "No description"
77 click.echo(f" {ds.name:<20} - {desc}")
103def run(algorithm, dataset, cache_dir, quiet, force):
104 """Run benchmarks with specified options"""
105 # Filter algorithms and datasets
106 filtered_algorithms = filter_algorithms(algorithm)
107 filtered_datasets = filter_datasets(dataset)
109 if not filtered_algorithms:
110 click.echo("Error: No matching algorithms found", err=True)
111 ctx = click.get_current_context()
112 ctx.exit(1)
113 if not filtered_datasets:
114 click.echo("Error: No matching datasets found", err=True)
115 ctx = click.get_current_context()
116 ctx.exit(1)
118 # Run benchmarks with filtered options
119 results = run_benchmarks(
120 cache_dir=cache_dir,
121 verbose=not quiet,
122 selected_algorithms=filtered_algorithms,
123 selected_datasets=filtered_datasets,
124 force=force,
125 )
127 # Print summary
128 click.echo("\nBenchmark Summary:")
129 for result in results["results"]:
130 click.echo(
131 f"\n{result['dataset']} + {result['algorithm']}:"
132 f"\n Compression ratio: {result['compression_ratio']:.2f}x"
133 f"\n Encode speed: {result['encode_mb_per_sec']:.2f} MB/s"
134 f"\n Decode speed: {result['decode_mb_per_sec']:.2f} MB/s"
135 )
138def main():
139 cli()
142if __name__ == "__main__":
143 main()