import os
import time
import csv
import shutil
import traceback
import yaml
from datetime import datetime
from typing import Type, Dict, Any, List
import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor, as_completed
import importlib
from ddmtolab.Methods.Algo_Methods.algo_utils import set_seed
[docs]
class BatchExperiment:
"""
Batch Experiment Module
This class provides a framework to define and run batch experiments for multiple
optimization algorithms on multiple benchmark problems. It supports:
- Adding multiple problems via problem creator functions.
- Adding multiple optimization algorithm classes with fixed parameters.
- Running experiments in parallel using multiple CPU cores.
- Logging execution time, status, and errors for each run.
- Saving timing summaries to CSV files.
- Printing experiment configuration summaries to console.
- Optional folder clearing before experiments.
- Saving and loading experiment configuration from YAML files.
.. warning::
``run()`` uses ``ProcessPoolExecutor``. On Windows (and macOS), which
use the ``spawn`` start method, the calling script **must** wrap the
call in an ``if __name__ == '__main__':`` guard. Calling ``run()`` at
module top level re-imports the script in every worker process and
spawns processes recursively.
Example
-------
>>> if __name__ == '__main__':
... batch_exp = BatchExperiment(base_path='./Data', clear_folder=True)
... batch_exp.add_problem(problem_creator, 'P1')
... batch_exp.add_algorithm(GA, 'GA', n=100, max_nfes=10000)
... batch_exp.run(n_runs=5, max_workers=6)
Author: Jiangtao Shen
Email: j.shen5@exeter.ac.uk
Date: 2025.11.25
Version: 1.0
"""
def __init__(self, base_path: str = './Data', clear_folder: bool = False):
"""
Initialize batch experiment
Args:
base_path: Base path for data storage, defaults to './Data'
clear_folder: If True, clear the base_path folder before initialization, defaults to False
"""
self.base_path = base_path
self.problems = [] # Store problems: [(problem_creator, problem_name, problem_params), ...]
self.algorithms = [] # Store algorithms: [(algo_class, algo_name, params), ...]
self.experiment_config: Dict[str, Any] = {
'created_at': datetime.now().isoformat(),
'base_path': str(base_path),
'clear_folder': clear_folder,
'problems': [],
'algorithms': []
}
# Clear folder if requested
if clear_folder and os.path.exists(self.base_path):
self._clear_folder(self.base_path)
print(f"♻️ Clearing existing data folder: {self.base_path}")
# Create base directory if it doesn't exist
if not os.path.exists(self.base_path):
os.makedirs(self.base_path)
print(f"Created base path: {self.base_path}")
def _clear_folder(self, folder_path: str):
"""
Clear all contents in the specified folder
Args:
folder_path: Path to the folder to be cleared
"""
try:
if os.path.exists(folder_path):
# Remove all contents in the folder
for item in os.listdir(folder_path):
item_path = os.path.join(folder_path, item)
if os.path.isfile(item_path) or os.path.islink(item_path):
os.unlink(item_path)
elif os.path.isdir(item_path):
shutil.rmtree(item_path)
except Exception as e:
print(f"Warning: Failed to clear folder {folder_path}: {str(e)}")
[docs]
def add_problem(self, problem_creator, problem_name: str, **problem_params):
"""
Add an experiment problem using a creator function
Args:
problem_creator: Function that creates the problem instance
problem_name: Problem name (used for file naming, must be unique)
**problem_params: Parameters to pass to problem creator
Raises:
ValueError: If a problem with the same name was already added
(duplicate names would silently overwrite result files).
"""
self._validate_name(problem_name, "Problem")
if any(name == problem_name for _, name, _ in self.problems):
raise ValueError(
f"Duplicate problem name '{problem_name}': result files are keyed "
f"by name, so runs would overwrite each other. Use a distinct name."
)
self.problems.append((problem_creator, problem_name, problem_params))
# Get the class name from the problem creator's __self__ attribute
class_name = 'unknown'
if hasattr(problem_creator, '__self__'):
class_name = problem_creator.__self__.__class__.__name__
# Save problem configuration
self.experiment_config['problems'].append({
'name': problem_name,
'class': class_name,
'creator_name': problem_creator.__name__ if hasattr(problem_creator, '__name__') else str(problem_creator),
'module': problem_creator.__module__ if hasattr(problem_creator, '__module__') else 'unknown',
'params': problem_params
})
[docs]
def add_algorithm(self, algorithm_class: Type, algorithm_name: str, **params):
"""
Add an optimization algorithm class
Args:
algorithm_class: Algorithm class (e.g., GA, DE, PSO, etc.)
algorithm_name: Algorithm name (used for file naming and folder
creation, must be unique)
**params: Fixed parameters for the algorithm (e.g., n, max_nfes, muc, mum, etc.)
Note: problem, save_path, and name will be set automatically
Raises:
ValueError: If an algorithm with the same name was already added
(duplicate names would silently overwrite result files).
"""
self._validate_name(algorithm_name, "Algorithm")
if any(name == algorithm_name for _, name, _ in self.algorithms):
raise ValueError(
f"Duplicate algorithm name '{algorithm_name}': result files are "
f"keyed by name, so runs would overwrite each other. Use a "
f"distinct name (e.g. '{algorithm_name}_2')."
)
self.algorithms.append((algorithm_class, algorithm_name, params))
# Create folder for this algorithm
algo_folder = os.path.join(self.base_path, algorithm_name)
if not os.path.exists(algo_folder):
os.makedirs(algo_folder)
# Save algorithm configuration
self.experiment_config['algorithms'].append({
'name': algorithm_name,
'class': algorithm_class.__name__,
'module': algorithm_class.__module__,
'parameters': params
})
@staticmethod
def _validate_name(name: str, kind: str):
"""
Reject names that cannot be used as file/directory components.
Algorithm and problem names become folder names and pkl filename parts;
characters like ':' or '*' fail at save time on Windows with an obscure
OS error, so fail fast with an actionable message instead.
"""
forbidden = set('<>:"/\\|?*')
bad = sorted(set(name) & forbidden)
if not name or not name.strip():
raise ValueError(f"{kind} name must be a non-empty string")
if bad:
raise ValueError(
f"{kind} name {name!r} contains character(s) {bad} that are "
f"invalid in file names. Use letters, digits, '-' or '_'."
)
@staticmethod
def _yaml_scalar(value) -> str:
"""
Render a single value as a YAML-safe inline scalar/flow string.
Plain f-string interpolation breaks the config file for values that
YAML treats specially (names containing ':', '#', quotes, ...); this
delegates quoting decisions to the YAML library itself.
"""
try:
dumped = yaml.safe_dump(value, default_flow_style=True,
allow_unicode=True, width=10 ** 9).strip()
except yaml.YAMLError:
dumped = yaml.safe_dump(str(value)).strip()
# Scalar documents carry an explicit end marker ("5\n..."); drop it
if dumped.endswith('\n...'):
dumped = dumped[:-4].strip()
return dumped
def _previous_run_history(self, config_path: str) -> List[Dict[str, Any]]:
"""
Read the run history already recorded in a configuration file.
Args:
config_path: Path to the configuration file, which need not exist
Returns:
The recorded history, oldest call first, or an empty list when there
is nothing to carry forward. A file written before the history was
kept still records its last call, so that call is preserved rather
than dropped on the first re-run.
"""
if not os.path.exists(config_path):
return []
try:
with open(config_path, 'r', encoding='utf-8') as f:
existing = yaml.safe_load(f) or {}
except Exception as e:
print(f"⚠️ Warning: Could not read the previous configuration "
f"({str(e)}); its run history is lost\n")
return []
history = existing.get('run_history')
if isinstance(history, list):
return [entry for entry in history if isinstance(entry, dict)]
previous = existing.get('run_settings')
return [previous] if isinstance(previous, dict) and previous else []
def save_config(self, n_runs: int, max_workers: int, base_seed: int = None):
"""
Save experiment configuration to YAML file with custom formatting
Every call to :meth:`run` appends its settings to ``run_history`` instead
of discarding the ones before it. A folder is often filled by several
calls -- completed runs are skipped on a re-run, so a batch extended from
20 to 30 runs, or resumed under a different seed, mixes results produced
under different settings. Keeping only the latest would leave the file
describing settings that never produced most of the data next to it.
``run_settings`` still holds the most recent call, so anything reading
it, :meth:`from_config` included, is unaffected.
Args:
n_runs: Number of independent runs
max_workers: Maximum number of worker processes
base_seed: Base seed used for system-level random-seed control, or None
if the runs were not seeded
"""
s = self._yaml_scalar # YAML-safe scalar renderer
# Save to YAML file
config_path = os.path.join(self.base_path, 'experiment_config.yaml')
# Add run settings, keeping the calls that came before this one
current_run = {'n_runs': n_runs, 'max_workers': max_workers,
'base_seed': base_seed,
'start_time': datetime.now().isoformat()}
self.experiment_config['run_settings'] = current_run
self.experiment_config['run_history'] = (
self._previous_run_history(config_path) + [current_run]
)
try:
with open(config_path, 'w', encoding='utf-8') as f:
# Write basic info
f.write(f"created_at: {s(self.experiment_config['created_at'])}\n")
f.write(f"base_path: {s(self.experiment_config['base_path'])}\n")
f.write(f"clear_folder: {s(self.experiment_config['clear_folder'])}\n\n")
# Write problems with blank lines between each
f.write("problems:\n")
for i, prob in enumerate(self.experiment_config['problems']):
if i > 0:
f.write("\n") # Add blank line before each problem (except first)
f.write(f" - name: {s(prob['name'])}\n")
f.write(f" class: {s(prob['class'])}\n")
f.write(f" creator_name: {s(prob['creator_name'])}\n")
f.write(f" module: {s(prob['module'])}\n")
f.write(f" params: {s(dict(prob['params']))}\n")
# Write algorithms with blank lines between each
f.write("\nalgorithms:\n")
for i, algo in enumerate(self.experiment_config['algorithms']):
if i > 0:
f.write("\n") # Add blank line before each algorithm (except first)
f.write(f" - name: {s(algo['name'])}\n")
f.write(f" class: {s(algo['class'])}\n")
f.write(f" module: {s(algo['module'])}\n")
f.write(f" parameters:\n")
for key, value in algo['parameters'].items():
f.write(f" {s(key)}: {s(value)}\n")
# Write run settings
f.write("\nrun_settings:\n")
f.write(f" n_runs: {s(self.experiment_config['run_settings']['n_runs'])}\n")
f.write(f" max_workers: {s(self.experiment_config['run_settings']['max_workers'])}\n")
f.write(f" base_seed: {s(self.experiment_config['run_settings']['base_seed'])}\n")
f.write(f" start_time: {s(self.experiment_config['run_settings']['start_time'])}\n")
# Write every call that has filled this folder, oldest first;
# the last entry is the run_settings above
f.write("\n# Every run() call that has written into this folder,"
" oldest first.\n")
f.write("# The data on disk may come from several of them, since"
" completed runs\n# are skipped when a batch is resumed"
" or extended.\n")
f.write("run_history:\n")
for i, entry in enumerate(self.experiment_config['run_history']):
if i > 0:
f.write("\n")
f.write(f" - n_runs: {s(entry.get('n_runs'))}\n")
f.write(f" max_workers: {s(entry.get('max_workers'))}\n")
f.write(f" base_seed: {s(entry.get('base_seed'))}\n")
f.write(f" start_time: {s(entry.get('start_time'))}\n")
print(f"💾 Configuration saved to: {config_path}\n")
except Exception as e:
print(f"⚠️ Warning: Failed to save configuration: {str(e)}\n")
@classmethod
def from_config(cls, config_path: str):
"""
Load experiment configuration from YAML file and create BatchExperiment instance
Args:
config_path: Path to the YAML configuration file
Returns:
BatchExperiment: Configured experiment instance ready to run
"""
print(f"📂 Loading configuration from: {config_path}")
# Read configuration file
try:
with open(config_path, 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)
except Exception as e:
raise RuntimeError(f"Failed to load configuration file: {str(e)}")
print(f"📅 Original creation time: {config['created_at']}")
# Create BatchExperiment instance
batch_exp = cls(
base_path=config['base_path'],
clear_folder=config['clear_folder']
)
# Load problems
for prob_config in config['problems']:
try:
# Import module
module_name = prob_config['module']
creator_name = prob_config['creator_name']
class_name = prob_config.get('class', 'unknown')
# Import the module
module = importlib.import_module(module_name)
# Try to get the problem class
problem_class = None
if class_name != 'unknown':
# If we have the class name, try to get it directly
if hasattr(module, class_name):
problem_class = getattr(module, class_name)
# If we still don't have the class, try to find it by looking for the creator method
if problem_class is None:
for attr_name in dir(module):
attr = getattr(module, attr_name)
if hasattr(attr, creator_name):
problem_class = attr
break
# If not found, skip this problem with a warning
if problem_class is None:
print(
f" ⚠️ Warning: Could not find creator '{creator_name}' in module '{module_name}', skipping problem '{prob_config['name']}'")
continue
# Create instance and get problem creator
instance = problem_class()
problem_creator = getattr(instance, creator_name)
batch_exp.add_problem(
problem_creator,
prob_config['name'],
**prob_config['params']
)
except Exception as e:
print(f" ⚠️ Warning: Failed to load {prob_config['name']}: {str(e)}, skipping...")
continue
# Load algorithms
for algo_config in config['algorithms']:
try:
# Import module and get algorithm class
module = importlib.import_module(algo_config['module'])
algorithm_class = getattr(module, algo_config['class'])
batch_exp.add_algorithm(
algorithm_class,
algo_config['name'],
**algo_config['parameters']
)
except Exception as e:
print(f" ❌ Failed to load {algo_config['name']}: {str(e)}")
raise
# Store run settings for later use
batch_exp._loaded_run_settings = config.get('run_settings', {})
print("✅ Configuration loaded successfully!")
return batch_exp
def _scan_completed(self, tasks: List[Dict[str, Any]]) -> set:
"""
Scan data folder for already completed experiments.
Args:
tasks: List of all task dicts (used to know which algo folders to check)
Returns:
Set of (algo_name, problem_name, run_id) tuples that are already done
"""
completed = set()
for task in tasks:
save_path = task['save_path']
file_name = task['file_name']
# Check for .pkl file matching this experiment
pkl_path = os.path.join(save_path, f"{file_name}.pkl")
if os.path.isfile(pkl_path):
completed.add((task['algo_name'], task['problem_name'], task['run_id']))
return completed
def _run_single_experiment(self, task: Dict[str, Any]) -> Dict[str, Any]:
"""
Run a single experiment task - recreate problem and algorithm in child process
Args:
task: Dictionary containing experiment parameters
Returns:
Dictionary with experiment results and timing information
"""
algo_class = task['algo_class']
algo_name = task['algo_name']
problem_creator = task['problem_creator']
problem_name = task['problem_name']
problem_params = task['problem_params']
run_id = task['run_id']
algo_params = task['algo_params']
save_path = task['save_path']
file_name = task['file_name']
seed = task.get('seed')
# Record start time for performance measurement
exp_start_time = time.time()
status = "Success"
error_msg = ""
try:
# Fix every random stream before anything stochastic happens, so the
# whole run is reproducible from this single seed
if seed is not None:
set_seed(seed)
# Recreate problem instance in child process to avoid pickling issues
problem_instance = problem_creator(**problem_params)
# Create algorithm instance with the problem and parameters
algorithm = algo_class(
problem=problem_instance,
save_path=save_path,
name=file_name,
**algo_params
)
# Execute the optimization
algorithm.optimize()
except Exception as e:
status = "Failed"
# Keep the full traceback so failed runs remain diagnosable
error_msg = f"{e}\n{traceback.format_exc()}"
# Calculate execution duration
exp_end_time = time.time()
exp_duration = exp_end_time - exp_start_time
return {
'Algorithm': algo_name,
'Problem': problem_name,
'Run': run_id,
'Filename': file_name,
'Time(s)': round(exp_duration, 4),
'Status': status,
'Error': error_msg,
'Seed': seed if seed is not None else ''
}
[docs]
def run(self, n_runs: int = None, verbose: bool = True, max_workers: int = None,
base_seed: int = None):
"""
Run all experiments using multi-core parallel processing
Args:
n_runs: Number of independent runs for each algorithm on each problem
If None and loaded from config, uses config value
verbose: Whether to print detailed progress information
max_workers: Maximum number of worker processes, defaults to CPU count if None
If None and loaded from config, uses config value
base_seed: Base seed for system-level random-seed control. When given, run
``r`` (1-indexed) is executed under seed ``base_seed + r - 1``,
applied to the global ``random``, NumPy and PyTorch (including
CUDA) generators before the problem and algorithm are built, so
each individual run is reproducible from its seed. The per-run
seed is recorded in the timing summary. If None and loaded from
config, uses config value; if still None, no seeding is performed
and runs use the interpreter's default random state.
Returns:
List[Dict]: One timing/status record per executed run, with keys
'Algorithm', 'Problem', 'Run', 'Filename', 'Time(s)', 'Status',
'Error', 'Seed'. Empty list if nothing was run.
"""
if not self.problems:
print("Error: No problems added!")
return []
if not self.algorithms:
print("Error: No algorithms added!")
return []
# Use loaded settings if available
if hasattr(self, '_loaded_run_settings'):
if n_runs is None:
n_runs = self._loaded_run_settings.get('n_runs', 30)
if max_workers is None:
max_workers = self._loaded_run_settings.get('max_workers', mp.cpu_count())
if base_seed is None:
base_seed = self._loaded_run_settings.get('base_seed', None)
else:
# Use default values
if n_runs is None:
n_runs = 30
if max_workers is None:
max_workers = mp.cpu_count()
if not isinstance(n_runs, int) or n_runs < 1:
raise ValueError(f"n_runs must be a positive integer, got {n_runs!r}")
if not isinstance(max_workers, int) or max_workers < 1:
raise ValueError(f"max_workers must be a positive integer, got {max_workers!r}")
if base_seed is not None and not isinstance(base_seed, int):
raise ValueError(f"base_seed must be an integer or None, got {base_seed!r}")
# Save configuration before running
self.save_config(n_runs, max_workers, base_seed)
total_experiments = len(self.problems) * len(self.algorithms) * n_runs
timing_records = []
# Display experiment configuration
print(f"=" * 60)
print("🚀🚀🚀 Starting Batch Experiment (Parallel Mode)! 🚀🚀🚀")
print(f"=" * 60)
print(f"\n1️⃣ Number of problems: {len(self.problems)}")
print(f"2️⃣ Number of algorithms: {len(self.algorithms)}")
print(f"3️⃣ Number of independent runs: {n_runs}")
print(f"🔢 Total experiments: {total_experiments}")
print(f"⚙️ Max workers: {max_workers}\n")
start_time = time.time()
# Prepare all tasks for parallel execution
all_tasks = []
for problem_creator, problem_name, problem_params in self.problems:
for algo_class, algo_name, algo_params in self.algorithms:
for run_id in range(1, n_runs + 1):
# Generate unique filename for each experiment
file_name = f"{algo_name}_{problem_name}_{run_id}"
save_path = os.path.join(self.base_path, algo_name)
task = {
'algo_class': algo_class,
'algo_name': algo_name,
'problem_creator': problem_creator,
'problem_name': problem_name,
'problem_params': problem_params,
'run_id': run_id,
'algo_params': algo_params,
'save_path': save_path,
'file_name': file_name,
'seed': None if base_seed is None else base_seed + run_id - 1
}
all_tasks.append(task)
# Scan for completed experiments and skip them
completed_set = self._scan_completed(all_tasks)
tasks = [t for t in all_tasks
if (t['algo_name'], t['problem_name'], t['run_id']) not in completed_set]
n_skipped = len(completed_set)
if n_skipped > 0:
print(f"📂 Found {n_skipped}/{total_experiments} completed experiments, skipping.")
if not tasks:
print("✅ All experiments already completed, nothing to run.")
return []
total_remaining = len(tasks)
print(f"▶️ Remaining experiments to run: {total_remaining}\n")
# Execute experiments in parallel using process pool
completed_count = 0
with ProcessPoolExecutor(max_workers=max_workers) as executor:
# Submit all tasks to the process pool
future_to_task = {
executor.submit(self._run_single_experiment, task): task
for task in tasks
}
# Process completed tasks as they finish
for future in as_completed(future_to_task):
task = future_to_task[future]
try:
result = future.result()
timing_records.append(result)
completed_count += 1
# Display progress information if verbose mode is enabled
if verbose:
progress = (completed_count / total_remaining) * 100
if completed_count % max(1, total_remaining // 100) == 0:
print(f"⏳ Progress: {completed_count}/{total_remaining} ({progress:.1f}%)")
except Exception as e:
# Handle task execution failures
print(f"Task failed with exception: {e}")
timing_records.append({
'Algorithm': task['algo_name'],
'Problem': task['problem_name'],
'Run': task['run_id'],
'Filename': task['file_name'],
'Time(s)': 0.0,
'Status': "Failed",
'Error': str(e),
'Seed': task['seed'] if task.get('seed') is not None else ''
})
completed_count += 1
# Calculate total execution time
end_time = time.time()
elapsed_time = end_time - start_time
# Generate and save timing summary CSV file
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
csv_filename = f"time_summary_{timestamp}.csv"
csv_path = os.path.join(self.base_path, csv_filename)
self._save_timing_summary(timing_records, csv_path)
# Display final summary
print(f"\n⏰ Total time: {elapsed_time:.2f} seconds ({elapsed_time / 60:.2f} minutes)")
if n_skipped > 0:
print(f"⏩ Skipped: {n_skipped}, Executed: {total_remaining}")
if elapsed_time > 0 and max_workers:
throughput = total_remaining / max_workers / (elapsed_time / 60)
print(f"💥 Throughput: {throughput:.2f} runs/worker/minute")
print(f"📊 Timing summary saved to: {csv_path}\n")
# Surface failures prominently: a run that crashed produces no .pkl,
# and downstream analysis would otherwise fail far from the cause.
failed = [r for r in timing_records if r['Status'] != 'Success']
if failed:
print(f"⚠️ {len(failed)}/{total_remaining} runs FAILED:")
for r in failed[:5]:
first_line = str(r['Error']).strip().splitlines()
first_line = first_line[-1] if first_line else ''
print(f" - {r['Filename']}: {first_line}")
if len(failed) > 5:
print(f" ... and {len(failed) - 5} more (see {csv_path})")
print()
print(f"=" * 60)
if failed:
print(f"⚠️ Experiments finished with {len(failed)} failure(s)")
else:
print(f"🎉🎉🎉 All Experiments Completed! 🎉🎉🎉")
print(f"=" * 60)
print("\n")
return timing_records
def _save_timing_summary(self, timing_records: List[Dict], csv_path: str):
"""
Save timing summary to CSV file
Args:
timing_records: List of timing records from all experiments
csv_path: Path to save the CSV file
"""
if not timing_records:
print("Warning: No timing records to save.")
return
try:
# utf-8-sig so Excel on Windows opens the file with correct encoding
with open(csv_path, 'w', newline='', encoding='utf-8-sig') as csvfile:
fieldnames = ['Algorithm', 'Problem', 'Run', 'Filename', 'Time(s)', 'Status', 'Error', 'Seed']
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(timing_records)
except Exception as e:
print(f"Error saving timing summary: {str(e)}")
# Usage example and demonstration
if __name__ == "__main__":
from ddmtolab.Algorithms.STSO.GA import GA
from ddmtolab.Algorithms.STSO.PSO import PSO
from ddmtolab.Algorithms.STSO.DE import DE
from ddmtolab.Problems.MTSO.cec17_mtso import CEC17MTSO
# ========== Method 1: Create and run experiments normally (auto-save config) ==========
# Create batch experiment instance with folder clearing enabled
# batch_exp = BatchExperiment(base_path='./Data', clear_folder=True)
# Add benchmark problems using creator functions
# cec17mtso = CEC17MTSO()
# batch_exp.add_problem(cec17mtso.P1, 'P1')
# batch_exp.add_problem(cec17mtso.P2, 'P2')
# Configure algorithm parameters
# n = 100
# max_nfes = 10000
# disable_tqdm = True
# Add optimization algorithm classes with their parameters
# batch_exp.add_algorithm(GA, 'GA', n=n, max_nfes=max_nfes, disable_tqdm=disable_tqdm)
# batch_exp.add_algorithm(DE, 'DE', n=n, max_nfes=max_nfes, disable_tqdm=disable_tqdm)
# batch_exp.add_algorithm(PSO, 'PSO', n=n, max_nfes=max_nfes, disable_tqdm=disable_tqdm)
# Execute experiments with parallel processing
# batch_exp.run(n_runs=5, verbose=True, max_workers=6)
# ========== Method 2: Load configuration from file and run experiments ==========
# Load configuration and run
# batch_exp2 = BatchExperiment.from_config('./Data/experiment_config.yaml')
# batch_exp2.run()
# Or override specific settings
# batch_exp2.run(n_runs=10, max_workers=8)