API 参考 - 主程序仿真入口 (Simulation)

主程序仿真入口 (Simulation)

主程序仿真入口 (Simulation) 包含了 TRICYS 的主要执行脚本，负责启动标准仿真和分析工作流。 请在下方的标签页中选择您感兴趣的特定模块。

Standard Simulation (标准仿真)Simulation Analysis (仿真分析)

main(config_path)

Main entry point for a standard simulation run.

This function prepares the configuration, sets up logging, and calls the main run_simulation orchestrator.

Parameters:

Name	Type	Description	Default
config_path	str	The path to the JSON configuration file.	required

Source code in tricys/simulation/simulation.py

def main(config_path: str) -> None:
    """Main entry point for a standard simulation run.

    This function prepares the configuration, sets up logging, and calls
    the main `run_simulation` orchestrator.

    Args:
        config_path (str): The path to the JSON configuration file.
    """
    config, original_config = basic_prepare_config(config_path)
    setup_logging(config, original_config)
    logger.info(
        "Loading configuration",
        extra={
            "config_path": os.path.abspath(config_path),
        },
    )
    try:
        run_simulation(config)
        logger.info("Main execution completed successfully")
    except Exception as e:
        logger.error(
            "Main execution failed", exc_info=True, extra={"exception": str(e)}
        )
        sys.exit(1)

run_simulation(config)

Orchestrates the main simulation workflow.

This function serves as the primary orchestrator for running simulations. It generates jobs from parameters, executes them (concurrently or sequentially, as standard or co-simulations), merges the results into a single DataFrame, and triggers any configured post-processing steps.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	The main configuration dictionary for the run.	required

Note

Supports concurrent (ThreadPoolExecutor) and sequential execution modes. Co-simulations use ProcessPoolExecutor for better isolation. Merges all job results into single CSV with parameter-labeled columns. Triggers post-processing tasks if configured. Results saved to paths.results_dir.

Source code in tricys/simulation/simulation.py

def run_simulation(config: Dict[str, Any]) -> None:
    """Orchestrates the main simulation workflow.

    This function serves as the primary orchestrator for running simulations.
    It generates jobs from parameters, executes them (concurrently or sequentially,
    as standard or co-simulations), merges the results into a single DataFrame,
    and triggers any configured post-processing steps.

    Args:
        config: The main configuration dictionary for the run.

    Note:
        Supports concurrent (ThreadPoolExecutor) and sequential execution modes.
        Co-simulations use ProcessPoolExecutor for better isolation. Merges all job
        results into single CSV with parameter-labeled columns. Triggers post-processing
        tasks if configured. Results saved to paths.results_dir.
    """
    jobs = generate_simulation_jobs(config.get("simulation_parameters", {}))

    try:
        results_dir = os.path.abspath(config["paths"]["results_dir"])
    except KeyError as e:
        logger.error(f"Missing required path key in configuration file: {e}")
        sys.exit(1)

    simulation_results = {}
    use_concurrent = config["simulation"].get("concurrent", False)

    try:
        max_workers = config["simulation"].get("max_workers", os.cpu_count())
        if config.get("co_simulation") is None:
            if use_concurrent:
                logger.info(
                    "Starting simulation",
                    extra={
                        "mode": "CONCURRENT",
                        "max_workers": max_workers,
                    },
                )
                with concurrent.futures.ThreadPoolExecutor(
                    max_workers=max_workers
                ) as executor:
                    future_to_job = {
                        executor.submit(
                            _run_single_job, config, job_params, i + 1
                        ): job_params
                        for i, job_params in enumerate(jobs)
                    }
                    for future in concurrent.futures.as_completed(future_to_job):
                        job_params = future_to_job[future]
                        try:
                            result_path = future.result()
                            if result_path:
                                simulation_results[
                                    tuple(sorted(job_params.items()))
                                ] = result_path
                        except Exception as exc:
                            logger.error(
                                f"Job for {job_params} generated an exception: {exc}",
                                exc_info=True,
                            )
            else:
                logger.info("Starting simulation", extra={"mode": "SEQUENTIAL"})
                result_paths = _run_sequential_sweep(config, jobs)
                for i, result_path in enumerate(result_paths):
                    if result_path:
                        simulation_results[tuple(sorted(jobs[i].items()))] = result_path
        else:
            if use_concurrent:
                logger.info(
                    "Starting co-simulation",
                    extra={
                        "mode": "CONCURRENT",
                        "max_workers": max_workers,
                    },
                )

                with concurrent.futures.ProcessPoolExecutor(
                    max_workers=max_workers
                ) as executor:
                    future_to_job = {
                        executor.submit(
                            _run_co_simulation, config, job_params, job_id=i + 1
                        ): job_params
                        for i, job_params in enumerate(jobs)
                    }

                    for future in concurrent.futures.as_completed(future_to_job):
                        job_params = future_to_job[future]
                        try:
                            result_path = future.result()
                            if result_path:
                                simulation_results[
                                    tuple(sorted(job_params.items()))
                                ] = result_path
                                logger.info(
                                    "Successfully finished co-simulation job",
                                    extra={
                                        "job_params": job_params,
                                    },
                                )
                            else:
                                logger.warning(
                                    "Co-simulation job did not return a result path",
                                    extra={
                                        "job_params": job_params,
                                    },
                                )
                        except Exception as exc:
                            logger.error(
                                "Co-simulation job generated an exception",
                                exc_info=True,
                                extra={
                                    "job_params": job_params,
                                    "exception": str(exc),
                                },
                            )
            else:
                logger.info("Starting co-simulation", extra={"mode": "SEQUENTIAL"})
                for i, job_params in enumerate(jobs):
                    job_id = i + 1
                    logger.info(
                        "Starting Sequential Co-simulation Job",
                        extra={
                            "job_index": f"{job_id}/{len(jobs)}",
                        },
                    )
                    try:
                        result_path = _run_co_simulation(
                            config, job_params, job_id=job_id
                        )
                        if result_path:
                            simulation_results[tuple(sorted(job_params.items()))] = (
                                result_path
                            )
                            logger.info(
                                "Successfully finished co-simulation job",
                                extra={
                                    "job_params": job_params,
                                },
                            )
                        else:
                            logger.warning(
                                "Co-simulation job did not return a result path",
                                extra={
                                    "job_params": job_params,
                                },
                            )
                    except Exception as exc:
                        logger.error(
                            "Co-simulation job generated an exception",
                            exc_info=True,
                            extra={
                                "job_params": job_params,
                                "exception": str(exc),
                            },
                        )
                    logger.info(
                        "Finished Sequential Co-simulation Job",
                        extra={
                            "job_index": f"{job_id}/{len(jobs)}",
                        },
                    )
    except Exception as e:
        raise RuntimeError("Failed to run simulation", e)

    # --- Result Handling ---
    # The simulation_results dictionary now contains paths to results inside temporary job workspaces.
    # The results_dir from the config is now the self-contained workspace's results folder.
    run_results_dir = results_dir
    os.makedirs(run_results_dir, exist_ok=True)

    # Unified result processing for both single and multiple jobs
    logger.info(
        "Processing jobs and combining results",
        extra={
            "num_jobs": len(jobs),
        },
    )
    combined_df = None

    all_dfs = []
    time_df_added = False

    for job_params in jobs:
        job_key = tuple(sorted(job_params.items()))
        result_path = simulation_results.get(job_key)

        if not result_path or not os.path.exists(result_path):
            logger.warning(
                "Job produced no result file",
                extra={
                    "job_params": job_params,
                },
            )
            continue

        # Read the current job's result file
        df = pd.read_csv(result_path)

        # From the very first valid DataFrame, grab the 'time' column
        if not time_df_added and "time" in df.columns:
            all_dfs.append(df[["time"]])
            time_df_added = True

        # Prepare the parameter string for column renaming
        param_string = "&".join([f"{k}={v}" for k, v in job_params.items()])

        # Isolate the data columns (everything except 'time')
        data_columns = df.drop(columns=["time"], errors="ignore")

        # Create a dictionary to map old column names to new ones
        # e.g., {'voltage': 'voltage&param1=A&param2=B'}
        rename_mapping = {
            col: f"{col}&{param_string}" if param_string else col
            for col in data_columns.columns
        }

        # Rename the columns and add the resulting DataFrame to our list
        all_dfs.append(data_columns.rename(columns=rename_mapping))

    # Concatenate all the DataFrames in the list along the columns axis (axis=1)
    if all_dfs:
        combined_df = pd.concat(all_dfs, axis=1)
    else:
        combined_df = pd.DataFrame()  # Or None, as you had before

    if combined_df is not None and not combined_df.empty:
        if len(jobs) == 1:
            # For single job, save as simulation_result.csv
            combined_csv_path = get_unique_filename(
                run_results_dir, "simulation_result.csv"
            )
        else:
            # For multiple jobs, save as sweep_results.csv
            combined_csv_path = get_unique_filename(
                run_results_dir, "sweep_results.csv"
            )

        combined_df.to_csv(combined_csv_path, index=False)
        logger.info(
            "Combined results saved",
            extra={
                "file_path": combined_csv_path,
            },
        )
    else:
        logger.warning("No valid results found to combine")

    # --- Post-Processing ---
    if combined_df is not None:
        # Calculate the top-level post-processing directory
        top_level_run_workspace = os.path.abspath(config["run_timestamp"])
        top_level_post_processing_dir = os.path.join(
            top_level_run_workspace, "post_processing"
        )
        _run_post_processing(config, combined_df, top_level_post_processing_dir)
    else:
        logger.warning("No simulation results generated, skipping post-processing")

    # --- Final Cleanup ---
    # The primary cleanup of job workspaces is handled by the `finally` block in `_run_co_simulation`.
    # This is an additional safeguard.
    if not config["simulation"].get("keep_temp_files", True):
        temp_dir_path = os.path.abspath(config["paths"].get("temp_dir", "temp"))
        logger.info(
            "Cleaning up temporary directory",
            extra={
                "directory": temp_dir_path,
            },
        )
        if os.path.exists(temp_dir_path):
            try:
                shutil.rmtree(temp_dir_path)
                os.makedirs(temp_dir_path)  # Recreate for next run
            except OSError as e:
                logger.error(
                    "Error cleaning up temporary directory",
                    extra={
                        "directory": temp_dir_path,
                        "error": str(e),
                    },
                )

main(config_path)

Main entry point for a simulation analysis run.

This function prepares the configuration for an analysis run, sets up logging, and calls the main run_simulation orchestrator for analysis.

Parameters:

Name	Type	Description	Default
config_path	str	The path to the JSON configuration file.	required

Source code in tricys/simulation/simulation_analysis.py

def main(config_path: str) -> None:
    """Main entry point for a simulation analysis run.

    This function prepares the configuration for an analysis run, sets up
    logging, and calls the main `run_simulation` orchestrator for analysis.

    Args:
        config_path (str): The path to the JSON configuration file.
    """
    config, original_config = analysis_prepare_config(config_path)
    setup_logging(config, original_config)
    logger.info(
        "Loading configuration",
        extra={
            "config_path": os.path.abspath(config_path),
        },
    )
    try:
        run_simulation(config)
        logger.info("Main execution completed successfully")
    except Exception as e:
        logger.error(
            "Main execution failed", exc_info=True, extra={"exception": str(e)}
        )
        sys.exit(1)

retry_analysis(timestamp)

Retries a failed AI analysis for a given run timestamp.

This function restores the configuration from the log file of a previous run and re-triggers the AI-dependent parts of the analysis, including report generation and consolidation.

Parameters:

Name	Type	Description	Default
timestamp	str	The timestamp of the run to retry (e.g., "20230101_120000").	required

Source code in tricys/simulation/simulation_analysis.py

def retry_analysis(timestamp: str) -> None:
    """Retries a failed AI analysis for a given run timestamp.

    This function restores the configuration from the log file of a previous
    run and re-triggers the AI-dependent parts of the analysis, including
    report generation and consolidation.

    Args:
        timestamp (str): The timestamp of the run to retry (e.g., "20230101_120000").
    """
    config, original_config = restore_configs_from_log(timestamp)
    if not config or not original_config:
        # Error is printed inside the helper function
        sys.exit(1)

    config["run_timestamp"] = timestamp

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s - %(levelname)s - %(message)s",
        stream=sys.stdout,
    )
    logger = logging.getLogger(__name__)
    logger.info(
        f"Successfully restored configuration for timestamp {timestamp} for retry."
    )

    logger.info("Starting in AI analysis retry mode...")
    if not analysis_validate_config(config):
        sys.exit(1)

    case_configs = analysis_setup_analysis_cases_workspaces(config)
    if not case_configs:
        logger.error("Could not set up case workspaces for retry. Aborting.")
        sys.exit(1)

    retry_ai_analysis(case_configs, config)
    consolidate_reports(case_configs, config)

    logger.info("AI analysis retry and consolidation complete.")

run_simulation(config)

Orchestrates the simulation analysis workflow.

This is the main orchestrator for a simulation analysis run. It handles different execution paths based on the configuration: - If 'analysis_cases' are defined, it sets up and executes each case, potentially in parallel. - If a SALib analysis is defined, it delegates to the SALib workflow. - Otherwise, it runs a standard parameter sweep, merges results, generates plots, and triggers sensitivity analysis and post-processing.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	The main configuration dictionary for the run.	required

Note

Supports three modes: multi-case analysis (analysis_cases), SALib analysis (independent_variable as list with analyzer), or standard parameter sweep. For multi-case mode, creates isolated workspaces and can run cases in parallel with ProcessPoolExecutor. Generates summary reports and handles AI analysis retries if configured.

Source code in tricys/simulation/simulation_analysis.py

def run_simulation(config: Dict[str, Any]) -> None:
    """Orchestrates the simulation analysis workflow.

    This is the main orchestrator for a simulation analysis run. It handles
    different execution paths based on the configuration:
    - If 'analysis_cases' are defined, it sets up and executes each case,
      potentially in parallel.
    - If a SALib analysis is defined, it delegates to the SALib workflow.
    - Otherwise, it runs a standard parameter sweep, merges results,
      generates plots, and triggers sensitivity analysis and post-processing.

    Args:
        config: The main configuration dictionary for the run.

    Note:
        Supports three modes: multi-case analysis (analysis_cases), SALib analysis
        (independent_variable as list with analyzer), or standard parameter sweep.
        For multi-case mode, creates isolated workspaces and can run cases in parallel
        with ProcessPoolExecutor. Generates summary reports and handles AI analysis retries
        if configured.
    """

    # 1. Split analysis_cases and determine salib_analysis_case
    has_analysis_cases = (
        "sensitivity_analysis" in config
        and "analysis_cases" in config["sensitivity_analysis"]
        and (
            # Support list format
            (
                isinstance(config["sensitivity_analysis"]["analysis_cases"], list)
                and len(config["sensitivity_analysis"]["analysis_cases"]) > 0
            )
            or
            # Support single object format
            isinstance(config["sensitivity_analysis"]["analysis_cases"], dict)
        )
    )

    # Check if it's a SALib analysis case (and not a multi-case analysis)
    sa_config = config.get("sensitivity_analysis", {})
    analysis_case = sa_config.get("analysis_case")

    has_salib_analysis_case = (
        not has_analysis_cases
        and isinstance(analysis_case, dict)
        and isinstance(analysis_case.get("independent_variable"), list)
        and isinstance(analysis_case.get("independent_variable_sampling"), dict)
        and "analyzer" in analysis_case
    )

    if has_analysis_cases and not has_salib_analysis_case:
        logger.info(
            "Detected analysis_cases field, starting to create independent working directories for each analysis case..."
        )

        # Create independent working directories and configuration files for each analysis_case
        case_configs = analysis_setup_analysis_cases_workspaces(config)

        if not case_configs:
            logger.error(
                "Unable to create analysis_cases working directories, stopping execution"
            )
            return

        logger.info(f"Starting execution of {len(case_configs)} analysis cases...")

        sa_config = config.get("sensitivity_analysis", {})
        run_cases_concurrently = sa_config.get("concurrent_cases", False)
        successful_cases = 0

        if run_cases_concurrently:
            logger.info(
                f"Starting execution of {len(case_configs)} analysis cases in PARALLEL."
            )
            max_workers = sa_config.get("max_case_workers", os.cpu_count())
            logger.info(
                f"Using up to {max_workers} parallel processes for analysis cases."
            )

            with concurrent.futures.ProcessPoolExecutor(
                max_workers=max_workers
            ) as executor:
                future_to_case = {
                    executor.submit(_execute_analysis_case, case_info): case_info
                    for case_info in case_configs
                }
                for future in concurrent.futures.as_completed(future_to_case):
                    case_info = future_to_case[future]
                    case_name = case_info["case_data"].get("name", case_info["index"])
                    try:
                        if future.result():
                            successful_cases += 1
                            logger.info(
                                f"Parallel case '{case_name}' completed successfully."
                            )
                        else:
                            logger.warning(
                                f"Parallel case '{case_name}' completed with errors."
                            )
                    except Exception as exc:
                        logger.error(
                            f"Parallel case '{case_name}' failed in executor with: {exc}",
                            exc_info=True,
                        )
        else:
            logger.info(
                f"Starting execution of {len(case_configs)} analysis cases SEQUENTIALLY."
            )
            for case_info in case_configs:
                try:
                    case_index = case_info["index"]
                    case_workspace = case_info["workspace"]
                    case_config = case_info["config"]
                    case_data = case_info["case_data"]

                    logger.info(
                        f"\n=== Starting execution of analysis case {case_index + 1}/{len(case_configs)} ==="
                    )
                    logger.info(
                        f"Case name: {case_data.get('name', f'Case{case_index+1}')}"
                    )
                    logger.info(
                        f"Independent variable: {case_data['independent_variable']}"
                    )
                    logger.info(f"Working directory: {case_workspace}")

                    original_cwd = os.getcwd()
                    os.chdir(case_workspace)

                    try:
                        setup_logging(case_config)
                        run_simulation(case_config)
                        successful_cases += 1
                        logger.info(
                            f"✓ Analysis case {case_index + 1} executed successfully"
                        )
                    except Exception as case_e:
                        logger.error(
                            f"✗ Analysis case {case_index + 1} execution failed: {case_e}",
                            exc_info=True,
                        )
                    finally:
                        os.chdir(original_cwd)
                        setup_logging(config)

                except Exception as e:
                    logger.error(
                        f"✗ Error processing analysis case {case_index + 1}: {e}",
                        exc_info=True,
                    )

        logger.info("\n=== Analysis Cases Execution Completed ===")
        logger.info(
            f"Successfully executed: {successful_cases}/{len(case_configs)} cases"
        )

        generate_analysis_cases_summary(case_configs, config)

        return  # End analysis_cases processing
    elif has_salib_analysis_case:
        logger.info("Detected SALib analysis case, diverting to SALib workflow...")
        run_salib_analysis(config)
        return  # SALib workflow is self-contained, so we exit here.

    # 2. Core operational logic
    jobs = generate_simulation_jobs(config.get("simulation_parameters", {}))

    # --- START: Add baseline jobs based on default parameter values ---
    analysis_case = config.get("sensitivity_analysis", {}).get("analysis_case", {})
    default_values = analysis_case.get("default_simulation_values")

    if default_values:
        logger.info(
            "Found default_simulation_values, generating additional baseline jobs."
        )

        # Prepare simulation parameters for the baseline run, starting with default values
        baseline_params = default_values.copy()

        # Add the main independent variable sweep to the baseline parameters
        independent_var = analysis_case.get("independent_variable")
        independent_sampling = analysis_case.get("independent_variable_sampling")

        if independent_var and independent_sampling:
            baseline_params[independent_var] = independent_sampling

            # Generate the additional jobs using the baseline config
            default_jobs = generate_simulation_jobs(baseline_params)

            # Combine with existing jobs and deduplicate
            combined_jobs = jobs + default_jobs

            # Deduplicate the list of job dictionaries
            seen = set()
            unique_jobs = []
            for job in combined_jobs:
                job_tuple = tuple(sorted(job.items()))
                if job_tuple not in seen:
                    seen.add(job_tuple)
                    unique_jobs.append(job)

            logger.info(
                f"Original jobs: {len(jobs)}, Combined jobs: {len(combined_jobs)}, Unique jobs after deduplication: {len(unique_jobs)}"
            )
            jobs = unique_jobs
    # --- END: Add baseline jobs ---

    try:
        results_dir = os.path.abspath(config["paths"]["results_dir"])
    except KeyError as e:
        logger.error(f"Missing required path key in configuration file: {e}")
        sys.exit(1)

    simulation_results = {}
    use_concurrent = config["simulation"].get("concurrent", False)

    try:
        if config.get("co_simulation") is None:
            if use_concurrent:
                logger.info("Starting simulation in CONCURRENT mode.")
                max_workers = config["simulation"].get("max_workers", os.cpu_count())
                logger.info(f"Using up to {max_workers} parallel workers.")
                final_results = []
                with concurrent.futures.ThreadPoolExecutor(
                    max_workers=max_workers
                ) as executor:
                    future_to_job = {
                        executor.submit(
                            _run_single_job, config, job_params, i + 1
                        ): job_params
                        for i, job_params in enumerate(jobs)
                    }
                    for future in concurrent.futures.as_completed(future_to_job):
                        job_params = future_to_job[future]
                        try:
                            (
                                optimal_params,
                                optimal_values,
                                result_path,
                            ) = future.result()
                            if result_path:
                                simulation_results[
                                    tuple(sorted(job_params.items()))
                                ] = result_path
                        except Exception as exc:
                            logger.error(
                                f"Job for {job_params} generated an exception: {exc}",
                                exc_info=True,
                            )
                        final_result_entry = job_params.copy()
                        final_result_entry.update(optimal_params)
                        final_result_entry.update(optimal_values)
                        final_results.append(final_result_entry)

                if _get_optimization_tasks(config):
                    results_dir = os.path.abspath(config["paths"]["results_dir"])
                    os.makedirs(results_dir, exist_ok=True)
                    if final_results:
                        final_df = pd.DataFrame(final_results)
                        output_path = os.path.join(
                            results_dir, "requierd_tbr_summary.csv"
                        )
                        final_df.to_csv(output_path, index=False, encoding="utf-8-sig")
                        logger.info(
                            f"Sweep optimization summary saved to: {output_path}"
                        )
            else:
                logger.info("Starting simulation in SEQUENTIAL mode.")
                result_paths = _run_sequential_sweep(config, jobs)
                for i, result_path in enumerate(result_paths):
                    if result_path:
                        simulation_results[tuple(sorted(jobs[i].items()))] = result_path
        else:
            if use_concurrent:
                logger.info("Starting co-simulation in CONCURRENT mode.")
                max_workers = config["simulation"].get("max_workers", 4)
                logger.info(f"Using up to {max_workers} parallel processes.")

                final_results = []

                with concurrent.futures.ProcessPoolExecutor(
                    max_workers=max_workers
                ) as executor:
                    future_to_job = {
                        executor.submit(
                            _run_co_simulation, config, job_params, job_id=i + 1
                        ): job_params
                        for i, job_params in enumerate(jobs)
                    }

                    for future in concurrent.futures.as_completed(future_to_job):
                        job_params = future_to_job[future]
                        try:
                            (
                                optimal_params,
                                optimal_values,
                                result_path,
                            ) = future.result()
                            if result_path:
                                simulation_results[
                                    tuple(sorted(job_params.items()))
                                ] = result_path
                                logger.info(
                                    f"Successfully finished job for params: {job_params}"
                                )
                            else:
                                logger.warning(
                                    f"Job for params {job_params} did not return a result path."
                                )
                        except Exception as exc:
                            logger.error(
                                f"Job for params {job_params} generated an exception: {exc}",
                                exc_info=True,
                            )
                        final_result_entry = job_params.copy()
                        final_result_entry.update(optimal_params)
                        final_result_entry.update(optimal_values)
                        final_results.append(final_result_entry)
            else:
                logger.info("Starting co-simulation in SEQUENTIAL mode.")
                final_results = []
                for i, job_params in enumerate(jobs):
                    job_id = i + 1
                    logger.info(f"--- Starting Sequential Job {job_id}/{len(jobs)} ---")
                    try:
                        (
                            optimal_params,
                            optimal_values,
                            result_path,
                        ) = _run_co_simulation(config, job_params, job_id=job_id)
                        if result_path:
                            simulation_results[tuple(sorted(job_params.items()))] = (
                                result_path
                            )
                            logger.info(
                                f"Successfully finished job for params: {job_params}"
                            )
                        else:
                            logger.warning(
                                f"Job for params {job_params} did not return a result path."
                            )
                        final_result_entry = job_params.copy()
                        final_result_entry.update(optimal_params)
                        final_result_entry.update(optimal_values)
                        final_results.append(final_result_entry)
                    except Exception as exc:
                        logger.error(
                            f"Job for params {job_params} generated an exception: {exc}",
                            exc_info=True,
                        )
                    logger.info(f"--- Finished Sequential Job {job_id}/{len(jobs)} ---")

            if _get_optimization_tasks(config):
                results_dir = os.path.abspath(config["paths"]["results_dir"])
                os.makedirs(results_dir, exist_ok=True)
                if final_results:
                    final_df = pd.DataFrame(final_results)
                    output_path = os.path.join(results_dir, "requierd_tbr_summary.csv")
                    final_df.to_csv(output_path, index=False, encoding="utf-8-sig")
                    logger.info(f"Sweep optimization summary saved to: {output_path}")
    except Exception as e:
        raise RuntimeError(f"Failed to run simualtion: {e}")

    # 3. Data merging and processing
    run_results_dir = results_dir
    os.makedirs(run_results_dir, exist_ok=True)

    # Unified result processing for both single and multiple jobs
    logger.info(f"Processing {len(jobs)} job(s). Combining results.")
    combined_df = None

    all_dfs = []
    time_df_added = False

    for job_params in jobs:
        job_key = tuple(sorted(job_params.items()))
        result_path = simulation_results.get(job_key)

        if not result_path or not os.path.exists(result_path):
            logger.warning(f"Job {job_params} produced no result file. Skipping.")
            continue

        # Read the current job's result file
        df = pd.read_csv(result_path)

        # From the very first valid DataFrame, grab the 'time' column
        if not time_df_added and "time" in df.columns:
            all_dfs.append(df[["time"]])
            time_df_added = True

        # Prepare the parameter string for column renaming
        param_string = "&".join([f"{k}={v}" for k, v in job_params.items()])

        # Isolate the data columns (everything except 'time')
        data_columns = df.drop(columns=["time"], errors="ignore")

        # Create a dictionary to map old column names to new ones
        # e.g., {'voltage': 'voltage&param1=A&param2=B'}
        rename_mapping = {
            col: f"{col}&{param_string}" if param_string else col
            for col in data_columns.columns
        }

        # Rename the columns and add the resulting DataFrame to our list
        all_dfs.append(data_columns.rename(columns=rename_mapping))

    # Concatenate all the DataFrames in the list along the columns axis (axis=1)
    if all_dfs:
        combined_df = pd.concat(all_dfs, axis=1)
    else:
        combined_df = pd.DataFrame()  # Or None, as you had before

    if combined_df is not None and not combined_df.empty:
        if len(jobs) == 1:
            # For single job, save as simulation_result.csv
            combined_csv_path = get_unique_filename(
                run_results_dir, "simulation_result.csv"
            )
        else:
            # For multiple jobs, save as sweep_results.csv
            combined_csv_path = get_unique_filename(
                run_results_dir, "sweep_results.csv"
            )

        # Clean up rows where the 'time' column is blank, which often occurs as redundant rows at the end of the file.
        combined_df.dropna(subset=["time"], inplace=True)
        combined_df.to_csv(combined_csv_path, index=False)
        logger.info(f"Combined results saved to: {combined_csv_path}")
    else:
        logger.warning("No valid results found to combine.")

    # Check if sweep_time plotting is enabled
    analysis_case = config["sensitivity_analysis"].get("analysis_case", {})
    sweep_time_list = analysis_case.get("sweep_time", None)
    if sweep_time_list and len(sweep_time_list) >= 1:
        # Get parameters for plot_sweep_time_series
        independent_var = analysis_case.get("independent_variable")
        dependent_vars = analysis_case.get("dependent_variables", [])
        independent_var_alias = analysis_case.get("independent_variable_alias")

        if (
            independent_var
            and dependent_vars
            and combined_csv_path
            and os.path.exists(combined_csv_path)
        ):

            try:
                # Get default values if they exist to filter the plot
                default_values = analysis_case.get("default_simulation_values")

                plot_path = plot_sweep_time_series(
                    csv_path=combined_csv_path,
                    save_dir=run_results_dir,
                    y_var_name=sweep_time_list,
                    independent_var_name=independent_var,
                    independent_var_alias=independent_var_alias,
                    default_params=default_values,  # Pass default values to the plot function
                    glossary_path=config["sensitivity_analysis"].get(
                        "glossary_path", None
                    ),
                )
                if plot_path:
                    logger.info(f"Sweep time series plot generated: {plot_path}")
                else:
                    logger.warning("Failed to generate sweep time series plot")
            except Exception as e:
                logger.error(f"Error generating sweep time series plot: {e}")

    # 4. Sensitivity analysis
    _run_sensitivity_analysis(config, run_results_dir, jobs)

    # 5. Post-processing
    if combined_df is not None:
        # Calculate the top-level post-processing directory
        top_level_run_workspace = os.path.abspath("post_processing")
        _run_post_processing(config, combined_df, top_level_run_workspace)
    else:
        logger.warning(
            "No simulation results were generated, skipping post-processing."
        )

    # 6. Intermediate data cleaning
    if not config["simulation"].get("keep_temp_files", True):
        logger.info("Cleaning up temporary directory...")
        temp_dir_path = os.path.abspath(config["paths"].get("temp_dir", "temp"))
        if os.path.exists(temp_dir_path):
            try:
                shutil.rmtree(temp_dir_path)
                os.makedirs(temp_dir_path)  # Recreate for next run
            except OSError as e:
                logger.error(f"Error cleaning up temp directory: {e}")