Convert PyTorch output to CK profiler commands.

This commit is contained in:
Ville Pietilä
2026-01-19 06:10:00 -05:00
parent 2625758522
commit 1db3473ff7
3 changed files with 1013 additions and 0 deletions

View File

@@ -0,0 +1,281 @@
# PyTorch to CK Profiler Conversion Tools
This directory contains two Python scripts to convert PyTorch convolution operations to CK Profiler format and execute them.
## Overview
1. **`convert_pytorch_to_ck.py`** - Converts PyTorch JSON to CK Profiler configuration JSON
2. **`run_ck_profiler.py`** - Executes CK profilers for each configuration
## Workflow
```bash
# Step 1: Convert PyTorch JSON to CK Profiler JSON
python convert_pytorch_to_ck.py build/test-data/conv_repros_ir.json ck_profiler_configs.json
# Step 2: Execute profilers
python run_ck_profiler.py ck_profiler_configs.json --profiler-path ./build/bin
```
---
## Script 1: convert_pytorch_to_ck.py
### Description
Converts PyTorch convolution operations from `conv_repros_ir.json` to CK Profiler configuration format.
### Supported Operations
- `aten::miopen_convolution` → Forward convolution
- `aten::convolution_backward` → Backward data and/or backward weight
### Configuration
The script uses these hardcoded settings:
- **Data type**: FP16 (data_type=1)
- **Layout**: NGCHW_GKCYX_NGKHW (layout=3)
- **Index type**: 32-bit (index_type=0, for forward only)
- **Verification**: Disabled (verify=0)
- **Initialization**: Integer values (init=1)
- **Logging**: Disabled (log=0)
- **Timing**: Enabled (time=1)
- **Split-K**: "all" (for backward operations)
### Usage
```bash
# Basic usage (uses defaults)
python convert_pytorch_to_ck.py
# Specify input and output files
python convert_pytorch_to_ck.py input.json output.json
# Help
python convert_pytorch_to_ck.py --help
```
### Default Files
- **Input**: `build/test-data/conv_repros_ir.json`
- **Output**: `ck_profiler_configs.json`
### Output Format
The output JSON contains an array of configurations:
```json
[
{
"operation_type": "profile_grouped_conv_fwd",
"profiler_args": {
"data_type": 1,
"layout": 3,
"index_type": 0,
"verify": 0,
"init": 1,
"log": 0,
"time": 1,
"num_dim_spatial": 2,
"G": 32,
"N": 32,
"K": 4,
"C": 4,
"filter_spatial": [3, 3],
"input_spatial": [200, 200],
"strides": [1, 1],
"dilations": [1, 1],
"left_pads": [1, 1],
"right_pads": [1, 1]
},
"metadata": {
"priority_rank": 1,
"pytorch_op": "aten::miopen_convolution",
"description": "Forward conv: G=32, N=32, K=4, C=4, [3, 3] kernel, [200, 200] input"
}
}
]
```
### Example Output
```
Converting PyTorch convolutions to CK Profiler format...
✓ [1/78] Converted: aten::miopen_convolution (rank 1) → fwd
✓ [2/78] Converted: aten::convolution_backward (rank 2) → bwd_data
✓ [2/78] Converted: aten::convolution_backward (rank 2) → bwd_weight
...
============================================================
Conversion Summary
============================================================
Total PyTorch operations: 78
✓ Forward convolutions: 30
✓ Backward data convolutions: 35
✓ Backward weight convolutions: 35
⚠ Skipped operations: 0
Total CK profiler configs: 100
Output file: ck_profiler_configs.json
============================================================
Conversion completed successfully!
```
---
## Script 2: run_ck_profiler.py
### Description
Executes CK Profiler binaries for each configuration in the JSON file, with support for dry-run, filtering, and result collection.
### Usage
```bash
# Run all configurations
python run_ck_profiler.py ck_profiler_configs.json
# Run first 10 only (for testing)
python run_ck_profiler.py ck_profiler_configs.json --max-ops 10
# Dry run (show commands without executing)
python run_ck_profiler.py ck_profiler_configs.json --dry-run
# Specify profiler path
python run_ck_profiler.py ck_profiler_configs.json --profiler-path ./build/bin
# Quiet mode (less verbose output)
python run_ck_profiler.py ck_profiler_configs.json --quiet
# Don't save results
python run_ck_profiler.py ck_profiler_configs.json --no-save
# Custom results output file
python run_ck_profiler.py ck_profiler_configs.json --output my_results.json
```
### Command Line Options
| Option | Description | Default |
|--------|-------------|---------|
| `config_file` | CK Profiler configuration JSON file | (required) |
| `--profiler-path` | Path to CK profiler binaries | `./build/bin` |
| `--max-ops` | Maximum number of operations to execute | All |
| `--dry-run` | Show commands without executing | False |
| `--quiet` | Reduce output verbosity | False |
| `--no-save` | Do not save results to JSON | False |
| `--output` | Output file for results | `profiler_results.json` |
### Example Output
```
Executing CK Profiler for 100 configurations...
======================================================================
[1/100] Forward conv: G=32, N=32, K=4, C=4, [3, 3] kernel, [200, 200] input
Priority Rank: 1
PyTorch Op: aten::miopen_convolution
======================================================================
Command: ./build/bin/profile_grouped_conv_fwd 1 3 0 0 1 0 1 2 32 32 4 4 3 3 200 200 1 1 1 1 1 1 1 1
✓ SUCCESS (completed in 2.34s)
Output:
...profiler output...
...
======================================================================
Execution Summary
======================================================================
Total configurations: 100
✓ Successful: 95
✗ Failed: 5
Success rate: 95.0%
======================================================================
Results saved to: profiler_results.json
```
### Results File Format
The `profiler_results.json` file contains:
```json
{
"timestamp": "2026-01-19T05:48:00",
"stats": {
"total": 100,
"success": 95,
"failed": 5,
"skipped": 0
},
"results": [
{
"operation": "profile_grouped_conv_fwd",
"description": "Forward conv: G=32, N=32, K=4, C=4, [3, 3] kernel, [200, 200] input",
"priority_rank": 1,
"success": true,
"returncode": 0,
"elapsed_time": 2.34,
"command": "./build/bin/profile_grouped_conv_fwd ...",
"error": null,
"stdout": "...truncated..."
}
]
}
```
---
## Requirements
- Python 3.6+
- CK profiler binaries built in `./build/bin` (or specify with `--profiler-path`)
- Input JSON file with PyTorch convolution operations
## Notes
### Layout Mapping
PyTorch uses **NCHW** layout, which maps to CK's **NGCHW_GKCYX_NGKHW** (layout=3):
- Input: [N, G, C, Hi, Wi]
- Weight: [G, K, C, Y, X]
- Output: [N, G, K, Ho, Wo]
### Per-Group Channels
For grouped convolutions:
- `C_per_group = C_total / groups`
- `K_per_group = K_total / groups`
The converter automatically computes these values.
### Backward Operations
The `aten::convolution_backward` operation can compute:
- **Backward data** (gradient w.r.t. input) when `output_mask[0]=true`
- **Backward weight** (gradient w.r.t. weight) when `output_mask[1]=true`
- Both if both flags are true
### Split-K Parameter
For backward operations, split_k is set to "all" which instructs the profiler to test all split-K values: -1, 1, 2, 4, 8, 16, 32, 64, 128.
---
## Troubleshooting
### "Profiler executable not found"
Ensure CK profilers are built:
```bash
mkdir -p build && cd build
cmake .. -DCMAKE_BUILD_TYPE=Release
make profile_grouped_conv_fwd profile_grouped_conv_bwd_data profile_grouped_conv_bwd_weight
```
### "Input file not found"
Check that the PyTorch JSON file exists:
```bash
ls -l build/test-data/conv_repros_ir.json
```
### Conversion warnings
Yellow warnings indicate operations that couldn't be converted (e.g., non-convolution operations). These are expected and don't indicate errors.
---

View File

@@ -0,0 +1,362 @@
#!/usr/bin/env python3
"""
Convert PyTorch convolution operations JSON to CK Profiler configuration JSON.
"""
import json
import sys
from pathlib import Path
from typing import Dict, List, Optional, Tuple
# ANSI color codes
class Colors:
GREEN = '\033[92m'
YELLOW = '\033[93m'
RED = '\033[91m'
BLUE = '\033[94m'
RESET = '\033[0m'
BOLD = '\033[1m'
# Configuration constants
DATA_TYPE = 1 # FP16
LAYOUT = 3 # NGCHW_GKCYX_NGKHW
INDEX_TYPE = 0 # 32-bit (only for forward)
VERIFY = 0 # No verification
INIT = 1 # Integer initialization
LOG = 0 # No log printing
TIME = 1 # Time kernels
SPLIT_K = "all" # For backward ops
class PyTorchToCKConverter:
"""Convert PyTorch convolution operations to CK Profiler format."""
def __init__(self):
self.converted_ops = []
self.skipped_ops = []
self.stats = {
'total': 0,
'forward': 0,
'backward_data': 0,
'backward_weight': 0,
'skipped': 0
}
def compute_output_spatial_dims(self, input_spatial: List[int],
filter_spatial: List[int],
strides: List[int],
paddings: List[int],
dilations: List[int]) -> List[int]:
"""Compute output spatial dimensions."""
output_spatial = []
for i in range(len(input_spatial)):
# Formula: output = floor((input + 2*pad - dilation*(kernel-1) - 1) / stride) + 1
kernel_eff = (filter_spatial[i] - 1) * dilations[i] + 1
output = (input_spatial[i] + 2 * paddings[i] - kernel_eff) // strides[i] + 1
output_spatial.append(output)
return output_spatial
def extract_forward_conv_params(self, op: Dict) -> Optional[Dict]:
"""Extract parameters for forward convolution (aten::miopen_convolution)."""
try:
args = op['replay_ir']['list_pos_args']
# Find arguments by name
arg_map = {arg['arg_name']: arg for arg in args}
# Extract tensor shapes
input_shape = arg_map['self']['value']['shape'] # [N, C_total, H, W]
weight_shape = arg_map['weight']['value']['shape'] # [K_total, C_per_group, Y, X]
# Extract convolution parameters
groups = arg_map['groups']['value']
stride = arg_map['stride']['value']
padding = arg_map['padding']['value']
dilation = arg_map['dilation']['value']
# Compute per-group channels
N = input_shape[0]
C_total = input_shape[1]
K_total = weight_shape[0]
G = groups
C = C_total // G # Per-group input channels
K = K_total // G # Per-group output channels
# Spatial dimensions
num_dim_spatial = len(stride)
input_spatial = input_shape[2:]
filter_spatial = weight_shape[2:]
# Convert single values to lists if needed
if isinstance(padding, int):
padding = [padding] * num_dim_spatial
if isinstance(stride, int):
stride = [stride] * num_dim_spatial
if isinstance(dilation, int):
dilation = [dilation] * num_dim_spatial
return {
'operation_type': 'profile_grouped_conv_fwd',
'profiler_args': {
'data_type': DATA_TYPE,
'layout': LAYOUT,
'index_type': INDEX_TYPE,
'verify': VERIFY,
'init': INIT,
'log': LOG,
'time': TIME,
'num_dim_spatial': num_dim_spatial,
'G': G,
'N': N,
'K': K,
'C': C,
'filter_spatial': filter_spatial,
'input_spatial': input_spatial,
'strides': stride,
'dilations': dilation,
'left_pads': padding,
'right_pads': padding
},
'metadata': {
'priority_rank': op['metadata']['priority_rank'],
'pytorch_op': op['op_name'],
'description': f"Forward conv: G={G}, N={N}, K={K}, C={C}, {filter_spatial} kernel, {input_spatial} input"
}
}
except Exception as e:
print(f"{Colors.YELLOW}⚠ Warning: Failed to extract forward conv params: {e}{Colors.RESET}")
return None
def extract_backward_conv_params(self, op: Dict) -> Optional[List[Dict]]:
"""Extract parameters for backward convolution (aten::convolution_backward)."""
try:
args = op['replay_ir']['list_pos_args']
# Find arguments by name
arg_map = {arg['arg_name']: arg for arg in args}
# Extract tensor shapes
grad_output_shape = arg_map['grad_output']['value']['shape'] # [N, K_total, Ho, Wo]
input_shape = arg_map['input']['value']['shape'] # [N, C_total, Hi, Wi]
weight_shape = arg_map['weight']['value']['shape'] # [K_total, C_per_group, Y, X]
# Extract convolution parameters
groups = arg_map['groups']['value']
stride = arg_map['stride']['value']
padding = arg_map['padding']['value']
dilation = arg_map['dilation']['value']
output_mask = arg_map['output_mask']['value']
# Compute per-group channels
N = input_shape[0]
C_total = input_shape[1]
K_total = weight_shape[0]
G = groups
C = C_total // G
K = K_total // G
# Spatial dimensions
num_dim_spatial = len(stride)
input_spatial = input_shape[2:]
filter_spatial = weight_shape[2:]
output_spatial = grad_output_shape[2:]
# Convert single values to lists if needed
if isinstance(padding, int):
padding = [padding] * num_dim_spatial
if isinstance(stride, int):
stride = [stride] * num_dim_spatial
if isinstance(dilation, int):
dilation = [dilation] * num_dim_spatial
results = []
# Backward data (computing gradient w.r.t. input)
if output_mask[0]: # grad_input
results.append({
'operation_type': 'profile_grouped_conv_bwd_data',
'profiler_args': {
'data_type': DATA_TYPE,
'layout': LAYOUT,
'verify': VERIFY,
'init': INIT,
'log': LOG,
'time': TIME,
'num_dim_spatial': num_dim_spatial,
'G': G,
'N': N,
'K': K,
'C': C,
'filter_spatial': filter_spatial,
'input_spatial': input_spatial,
'strides': stride,
'dilations': dilation,
'left_pads': padding,
'right_pads': padding,
'split_k': SPLIT_K
},
'metadata': {
'priority_rank': op['metadata']['priority_rank'],
'pytorch_op': op['op_name'],
'description': f"Backward data: G={G}, N={N}, K={K}, C={C}, {filter_spatial} kernel, {input_spatial} input"
}
})
# Backward weight (computing gradient w.r.t. weight)
if output_mask[1]: # grad_weight
results.append({
'operation_type': 'profile_grouped_conv_bwd_weight',
'profiler_args': {
'data_type': DATA_TYPE,
'layout': LAYOUT,
'verify': VERIFY,
'init': INIT,
'log': LOG,
'time': TIME,
'num_dim_spatial': num_dim_spatial,
'G': G,
'N': N,
'K': K,
'C': C,
'filter_spatial': filter_spatial,
'input_spatial': input_spatial,
'strides': stride,
'dilations': dilation,
'left_pads': padding,
'right_pads': padding,
'split_k': SPLIT_K
},
'metadata': {
'priority_rank': op['metadata']['priority_rank'],
'pytorch_op': op['op_name'],
'description': f"Backward weight: G={G}, N={N}, K={K}, C={C}, {filter_spatial} kernel, {input_spatial} input"
}
})
return results if results else None
except Exception as e:
print(f"{Colors.YELLOW}⚠ Warning: Failed to extract backward conv params: {e}{Colors.RESET}")
return None
def convert_operation(self, op: Dict) -> Optional[List[Dict]]:
"""Convert a single PyTorch operation to CK profiler config(s)."""
op_name = op['op_name']
if op_name == 'aten::miopen_convolution':
result = self.extract_forward_conv_params(op)
if result:
self.stats['forward'] += 1
return [result]
elif op_name == 'aten::convolution_backward':
results = self.extract_backward_conv_params(op)
if results:
for result in results:
if result['operation_type'] == 'profile_grouped_conv_bwd_data':
self.stats['backward_data'] += 1
elif result['operation_type'] == 'profile_grouped_conv_bwd_weight':
self.stats['backward_weight'] += 1
return results
else:
print(f"{Colors.YELLOW}⚠ Warning: Skipping unsupported operation: {op_name}{Colors.RESET}")
self.skipped_ops.append({
'op_name': op_name,
'priority_rank': op['metadata']['priority_rank']
})
self.stats['skipped'] += 1
return None
return None
def convert_file(self, input_path: str, output_path: str):
"""Convert PyTorch JSON file to CK Profiler JSON file."""
print(f"{Colors.BOLD}Converting PyTorch convolutions to CK Profiler format...{Colors.RESET}\n")
# Load input JSON
with open(input_path, 'r') as f:
pytorch_ops = json.load(f)
self.stats['total'] = len(pytorch_ops)
# Convert each operation
for i, op in enumerate(pytorch_ops, 1):
op_name = op['op_name']
priority = op['metadata']['priority_rank']
results = self.convert_operation(op)
if results:
self.converted_ops.extend(results)
for result in results:
op_type = result['operation_type'].replace('profile_grouped_conv_', '')
print(f"{Colors.GREEN}{Colors.RESET} [{i}/{len(pytorch_ops)}] Converted: {op_name} (rank {priority}) → {op_type}")
else:
print(f"{Colors.YELLOW}{Colors.RESET} [{i}/{len(pytorch_ops)}] Skipped: {op_name} (rank {priority})")
# Write output JSON
with open(output_path, 'w') as f:
json.dump(self.converted_ops, f, indent=2)
# Print summary
self.print_summary(output_path)
def print_summary(self, output_path: str):
"""Print conversion summary."""
print(f"\n{Colors.BOLD}{'='*60}{Colors.RESET}")
print(f"{Colors.BOLD}Conversion Summary{Colors.RESET}")
print(f"{Colors.BOLD}{'='*60}{Colors.RESET}")
print(f"Total PyTorch operations: {self.stats['total']}")
print(f" {Colors.GREEN}{Colors.RESET} Forward convolutions: {self.stats['forward']}")
print(f" {Colors.GREEN}{Colors.RESET} Backward data convolutions: {self.stats['backward_data']}")
print(f" {Colors.GREEN}{Colors.RESET} Backward weight convolutions: {self.stats['backward_weight']}")
if self.stats['skipped'] > 0:
print(f" {Colors.YELLOW}{Colors.RESET} Skipped operations: {self.stats['skipped']}")
print(f"\nTotal CK profiler configs: {len(self.converted_ops)}")
print(f"Output file: {output_path}")
print(f"{Colors.BOLD}{'='*60}{Colors.RESET}\n")
def main():
"""Main entry point."""
import argparse
parser = argparse.ArgumentParser(
description='Convert PyTorch convolution JSON to CK Profiler configuration JSON'
)
parser.add_argument(
'input',
nargs='?',
default='build/test-data/conv_repros_ir.json',
help='Input PyTorch JSON file (default: build/test-data/conv_repros_ir.json)'
)
parser.add_argument(
'output',
nargs='?',
default='ck_profiler_configs.json',
help='Output CK Profiler JSON file (default: ck_profiler_configs.json)'
)
args = parser.parse_args()
# Check if input file exists
if not Path(args.input).exists():
print(f"{Colors.RED}Error: Input file not found: {args.input}{Colors.RESET}")
sys.exit(1)
# Run conversion
converter = PyTorchToCKConverter()
try:
converter.convert_file(args.input, args.output)
print(f"{Colors.GREEN}Conversion completed successfully!{Colors.RESET}")
except Exception as e:
print(f"{Colors.RED}Error during conversion: {e}{Colors.RESET}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == '__main__':
main()

View File

@@ -0,0 +1,370 @@
#!/usr/bin/env python3
"""
Execute CK Profiler for each configuration in the JSON file.
"""
import json
import subprocess
import sys
import argparse
from pathlib import Path
from typing import Dict, List, Optional
from datetime import datetime
import time
# ANSI color codes
class Colors:
GREEN = '\033[92m'
YELLOW = '\033[93m'
RED = '\033[91m'
BLUE = '\033[94m'
CYAN = '\033[96m'
RESET = '\033[0m'
BOLD = '\033[1m'
class CKProfilerExecutor:
"""Execute CK Profiler for each configuration."""
def __init__(self, profiler_path: str = './build/bin', dry_run: bool = False):
self.profiler_path = Path(profiler_path)
self.dry_run = dry_run
self.results = []
self.stats = {
'total': 0,
'success': 0,
'failed': 0,
'skipped': 0
}
def build_command(self, config: Dict) -> List[str]:
"""Build command line from JSON config."""
op_type = config['operation_type']
args = config['profiler_args']
# Build argument list
cmd = [str(self.profiler_path / op_type)]
# Add arguments based on operation type
if op_type == 'profile_grouped_conv_fwd':
# Forward convolution arguments
cmd.extend([
str(args['data_type']),
str(args['layout']),
str(args['index_type']),
str(args['verify']),
str(args['init']),
str(args['log']),
str(args['time']),
str(args['num_dim_spatial']),
str(args['G']),
str(args['N']),
str(args['K']),
str(args['C'])
])
else:
# Backward convolution arguments (no index_type)
cmd.extend([
str(args['data_type']),
str(args['layout']),
str(args['verify']),
str(args['init']),
str(args['log']),
str(args['time']),
str(args['num_dim_spatial']),
str(args['G']),
str(args['N']),
str(args['K']),
str(args['C'])
])
# Add spatial parameters (same order for all operation types)
# filter_spatial, input_spatial, strides, dilations, left_pads, right_pads
cmd.extend([str(x) for x in args['filter_spatial']])
cmd.extend([str(x) for x in args['input_spatial']])
cmd.extend([str(x) for x in args['strides']])
cmd.extend([str(x) for x in args['dilations']])
cmd.extend([str(x) for x in args['left_pads']])
cmd.extend([str(x) for x in args['right_pads']])
# Add split_k for backward ops
if 'split_k' in args:
cmd.append(str(args['split_k']))
return cmd
def run_profiler(self, config: Dict, index: int, total: int, verbose: bool = True) -> Dict:
"""Run profiler for a single config."""
cmd = self.build_command(config)
metadata = config['metadata']
if verbose:
print(f"\n{Colors.BOLD}{'='*70}{Colors.RESET}")
print(f"{Colors.BOLD}[{index}/{total}] {metadata['description']}{Colors.RESET}")
print(f"{Colors.CYAN}Priority Rank: {metadata['priority_rank']}{Colors.RESET}")
print(f"{Colors.CYAN}PyTorch Op: {metadata['pytorch_op']}{Colors.RESET}")
print(f"{Colors.BOLD}{'='*70}{Colors.RESET}")
# Print command
cmd_str = ' '.join(cmd)
if self.dry_run:
print(f"{Colors.YELLOW}[DRY RUN]{Colors.RESET} Would execute:")
print(f" {cmd_str}")
return {
'success': True,
'dry_run': True,
'command': cmd_str
}
if verbose:
print(f"{Colors.BLUE}Command:{Colors.RESET} {cmd_str}\n")
# Execute command
start_time = time.time()
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=600 # 10 minute timeout
)
elapsed_time = time.time() - start_time
success = result.returncode == 0
if verbose:
if success:
print(f"{Colors.GREEN}✓ SUCCESS{Colors.RESET} (completed in {elapsed_time:.2f}s)")
else:
print(f"{Colors.RED}✗ FAILED{Colors.RESET} (return code: {result.returncode})")
# Print stdout if present
if result.stdout:
print(f"\n{Colors.BOLD}Output:{Colors.RESET}")
print(result.stdout)
# Print stderr if present and failed
if result.stderr and not success:
print(f"\n{Colors.RED}Error Output:{Colors.RESET}")
print(result.stderr)
return {
'success': success,
'returncode': result.returncode,
'stdout': result.stdout,
'stderr': result.stderr,
'elapsed_time': elapsed_time,
'command': cmd_str
}
except subprocess.TimeoutExpired:
elapsed_time = time.time() - start_time
if verbose:
print(f"{Colors.RED}✗ TIMEOUT{Colors.RESET} after {elapsed_time:.2f}s")
return {
'success': False,
'error': f'Timeout after {elapsed_time:.2f}s',
'command': cmd_str
}
except FileNotFoundError:
if verbose:
print(f"{Colors.RED}✗ ERROR{Colors.RESET}: Profiler executable not found")
print(f" Looking for: {cmd[0]}")
print(f" Please check that CK profilers are built in: {self.profiler_path}")
return {
'success': False,
'error': f'Profiler executable not found: {cmd[0]}',
'command': cmd_str
}
except Exception as e:
if verbose:
print(f"{Colors.RED}✗ ERROR{Colors.RESET}: {str(e)}")
return {
'success': False,
'error': str(e),
'command': cmd_str
}
def run_all(self, config_file: str, max_ops: Optional[int] = None,
verbose: bool = True, save_results: bool = True) -> List[Dict]:
"""Execute all profiler configurations."""
# Load configurations
with open(config_file) as f:
configs = json.load(f)
total = len(configs)
if max_ops:
total = min(max_ops, total)
configs = configs[:max_ops]
self.stats['total'] = total
print(f"{Colors.BOLD}Executing CK Profiler for {total} configurations...{Colors.RESET}\n")
# Execute each configuration
for i, config in enumerate(configs, 1):
result = self.run_profiler(config, i, total, verbose)
# Track result
self.results.append({
'config': config,
'result': result
})
# Update stats
if result.get('success'):
self.stats['success'] += 1
else:
self.stats['failed'] += 1
# Print summary
self.print_summary()
# Save results if requested
if save_results and not self.dry_run:
self.save_results()
return self.results
def print_summary(self):
"""Print execution summary."""
print(f"\n{Colors.BOLD}{'='*70}{Colors.RESET}")
print(f"{Colors.BOLD}Execution Summary{Colors.RESET}")
print(f"{Colors.BOLD}{'='*70}{Colors.RESET}")
print(f"Total configurations: {self.stats['total']}")
print(f" {Colors.GREEN}{Colors.RESET} Successful: {self.stats['success']}")
if self.stats['failed'] > 0:
print(f" {Colors.RED}{Colors.RESET} Failed: {self.stats['failed']}")
if self.stats['success'] > 0:
success_rate = (self.stats['success'] / self.stats['total']) * 100
print(f"\nSuccess rate: {success_rate:.1f}%")
print(f"{Colors.BOLD}{'='*70}{Colors.RESET}\n")
def save_results(self, output_file: str = 'profiler_results.json'):
"""Save execution results to JSON file."""
timestamp = datetime.now().isoformat()
output_data = {
'timestamp': timestamp,
'stats': self.stats,
'results': []
}
for item in self.results:
config = item['config']
result = item['result']
output_data['results'].append({
'operation': config['operation_type'],
'description': config['metadata']['description'],
'priority_rank': config['metadata']['priority_rank'],
'success': result.get('success', False),
'returncode': result.get('returncode'),
'elapsed_time': result.get('elapsed_time'),
'command': result.get('command'),
'error': result.get('error'),
'stdout': result.get('stdout', '')[:500] if result.get('stdout') else None # Truncate long output
})
with open(output_file, 'w') as f:
json.dump(output_data, f, indent=2)
print(f"{Colors.GREEN}Results saved to: {output_file}{Colors.RESET}")
def main():
"""Main entry point."""
parser = argparse.ArgumentParser(
description='Execute CK Profiler for configurations in JSON file',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Run all configurations
python run_ck_profiler.py ck_profiler_configs.json
# Run first 10 only
python run_ck_profiler.py ck_profiler_configs.json --max-ops 10
# Dry run (show commands without executing)
python run_ck_profiler.py ck_profiler_configs.json --dry-run
# Specify profiler path
python run_ck_profiler.py ck_profiler_configs.json --profiler-path ./build/bin
"""
)
parser.add_argument(
'config_file',
help='CK Profiler configuration JSON file'
)
parser.add_argument(
'--profiler-path',
default='./build/bin',
help='Path to CK profiler binaries (default: ./build/bin)'
)
parser.add_argument(
'--max-ops',
type=int,
help='Maximum number of operations to execute'
)
parser.add_argument(
'--dry-run',
action='store_true',
help='Show commands without executing them'
)
parser.add_argument(
'--quiet',
action='store_true',
help='Reduce output verbosity'
)
parser.add_argument(
'--no-save',
action='store_true',
help='Do not save results to JSON file'
)
parser.add_argument(
'--output',
default='profiler_results.json',
help='Output file for results (default: profiler_results.json)'
)
args = parser.parse_args()
# Check if config file exists
if not Path(args.config_file).exists():
print(f"{Colors.RED}Error: Config file not found: {args.config_file}{Colors.RESET}")
sys.exit(1)
# Create executor
executor = CKProfilerExecutor(
profiler_path=args.profiler_path,
dry_run=args.dry_run
)
# Run profilers
try:
executor.run_all(
args.config_file,
max_ops=args.max_ops,
verbose=not args.quiet,
save_results=not args.no_save
)
# Exit with error code if any failed
if executor.stats['failed'] > 0 and not args.dry_run:
sys.exit(1)
except KeyboardInterrupt:
print(f"\n{Colors.YELLOW}Interrupted by user{Colors.RESET}")
sys.exit(1)
except Exception as e:
print(f"{Colors.RED}Error: {e}{Colors.RESET}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == '__main__':
main()