API 参考 - 分析模块 (Analysis)

分析模块 (Analysis)

分析模块 (Analysis) 提供了用于处理、可视化和报告 TRICYS 仿真结果的工具。 请在下方的标签页中选择您感兴趣的特定模块。

Metric (指标)Plot (绘图)Report (报告)SALib (敏感性)

calculate_doubling_time(series, time_series)

Calculates the time it takes for the inventory to double its initial value.

This function finds the first time point, after the inventory's minimum (turning point), where the inventory level reaches or exceeds twice its initial value.

Parameters:

Name	Type	Description	Default
series	Series	The inventory time series data.	required
time_series	Series	The corresponding time data.	required

Returns:

Type	Description
float	The doubling time, or NaN if the inventory never doubles.

Raises:

Type	Description
ValueError	If time_series is None.

Note

Only considers the portion of the series after the turning point (minimum). Returns NaN if the inventory never reaches twice the initial value in the post-turning-point region.

Source code in tricys/analysis/metric.py

def calculate_doubling_time(series: pd.Series, time_series: pd.Series) -> float:
    """Calculates the time it takes for the inventory to double its initial value.

    This function finds the first time point, after the inventory's minimum
    (turning point), where the inventory level reaches or exceeds twice its
    initial value.

    Args:
        series: The inventory time series data.
        time_series: The corresponding time data.

    Returns:
        The doubling time, or NaN if the inventory never doubles.

    Raises:
        ValueError: If time_series is None.

    Note:
        Only considers the portion of the series after the turning point (minimum).
        Returns NaN if the inventory never reaches twice the initial value in the
        post-turning-point region.
    """
    if time_series is None:
        raise ValueError("time_series must be provided for calculate_doubling_time")
    initial_inventory = series.iloc[0]
    doubled_inventory = 2 * initial_inventory

    # Find the first index where the inventory is >= doubled_inventory
    # We should only consider the part of the series after the turning point
    min_index = series.idxmin()
    after_turning_point_series = series.loc[min_index:]

    doubling_indices = after_turning_point_series[
        after_turning_point_series >= doubled_inventory
    ].index

    if not doubling_indices.empty:
        doubling_index = doubling_indices[0]
        return time_series.loc[doubling_index]
    else:
        # If it never doubles, return NaN
        return np.nan

calculate_startup_inventory(series, time_series=None)

Calculates the startup inventory.

The startup inventory is calculated as the difference between the initial inventory and the minimum inventory (the turning point).

Parameters:

Name	Type	Description	Default
series	Series	The inventory time series data.	required
time_series	Optional[Series]	The corresponding time data (unused).	None

Returns:

Type	Description
float	The calculated startup inventory.

Note

The time_series parameter is provided for interface consistency but is not used in the calculation. The startup inventory represents the amount of inventory consumed before reaching the minimum point.

Source code in tricys/analysis/metric.py

def calculate_startup_inventory(
    series: pd.Series, time_series: Optional[pd.Series] = None
) -> float:
    """Calculates the startup inventory.

    The startup inventory is calculated as the difference between the initial
    inventory and the minimum inventory (the turning point).

    Args:
        series: The inventory time series data.
        time_series: The corresponding time data (unused).

    Returns:
        The calculated startup inventory.

    Note:
        The time_series parameter is provided for interface consistency but is not
        used in the calculation. The startup inventory represents the amount of
        inventory consumed before reaching the minimum point.
    """
    initial_inventory = series.iloc[0]
    minimum_inventory = series.min()
    return initial_inventory - minimum_inventory

extract_metrics(results_df, metrics_definition, analysis_case)

Extracts summary metrics from detailed simulation results.

This function processes a DataFrame from a parameter sweep, calculates various metrics for each run based on a definitions dictionary, and pivots the results into a summary DataFrame where each row corresponds to a unique parameter combination.

Parameters:

Name	Type	Description	Default
results_df	DataFrame	DataFrame from the combined sweep results.	required
metrics_definition	Dict[str, Any]	Dictionary defining how to calculate each metric (e.g., source column, method).	required
analysis_case	Dict[str, Any]	The analysis case configuration, used to identify dependent variables.	required

Returns:

Type	Description
DataFrame	A pivoted DataFrame with parameters as the index and metric names as columns.

Note

Parses column names in format "variable&param1=value1&param2=value2" to extract parameter values. Skips metrics with "bisection_search" method. Returns empty DataFrame if no valid metrics are found or if pivoting fails.

Source code in tricys/analysis/metric.py

def extract_metrics(
    results_df: pd.DataFrame,
    metrics_definition: Dict[str, Any],
    analysis_case: Dict[str, Any],
) -> pd.DataFrame:
    """Extracts summary metrics from detailed simulation results.

    This function processes a DataFrame from a parameter sweep, calculates
    various metrics for each run based on a definitions dictionary, and
    pivots the results into a summary DataFrame where each row corresponds
    to a unique parameter combination.

    Args:
        results_df: DataFrame from the combined sweep results.
        metrics_definition: Dictionary defining how to calculate each metric
            (e.g., source column, method).
        analysis_case: The analysis case configuration, used to identify
            dependent variables.

    Returns:
        A pivoted DataFrame with parameters as the index and metric names as columns.

    Note:
        Parses column names in format "variable&param1=value1&param2=value2" to extract
        parameter values. Skips metrics with "bisection_search" method. Returns empty
        DataFrame if no valid metrics are found or if pivoting fails.
    """

    analysis_results = []

    source_to_metric = {}
    dependent_vars = analysis_case.get("dependent_variables", [])

    for metric_name in dependent_vars:
        definition = metrics_definition.get(metric_name)

        # If the metric is not in the definition or is calculated via optimization, skip it.
        if not definition or definition.get("method") == "bisection_search":
            continue

        source = definition["source_column"]
        if source not in source_to_metric:
            source_to_metric[source] = []
        source_to_metric[source].append(
            {
                "metric_name": metric_name,
                "method": definition["method"],
            }
        )

    for col_name in results_df.columns:
        if col_name.lower() == "time":
            continue

        source_var = None
        for var in source_to_metric.keys():
            if col_name.startswith(var):
                source_var = var
                break

        if not source_var:
            continue

        param_str = col_name[len(source_var) :].lstrip("&")

        try:
            params = dict(item.split("=") for item in param_str.split("&"))
        except ValueError:
            print(
                f"Warning: Could not parse parameters from column '{col_name}'. Skipping."
            )
            continue

        for k, v in params.items():
            try:
                params[k] = float(v)
            except ValueError:
                params[k] = v

        for metric_info in source_to_metric[source_var]:
            method_name = metric_info["method"]
            metric_name = metric_info["metric_name"]

            if method_name == "final_value":
                calculation_func = get_final_value
            elif method_name == "calculate_startup_inventory":
                calculation_func = calculate_startup_inventory
            elif method_name == "time_of_turning_point":
                calculation_func = time_of_turning_point
            elif method_name == "calculate_doubling_time":
                calculation_func = calculate_doubling_time
            else:
                print(
                    f"Warning: Calculation method '{method_name}' not implemented. Skipping."
                )
                continue

            metric_value = calculation_func(results_df[col_name], results_df["time"])

            result_row = params.copy()
            result_row["metric_name"] = metric_name
            result_row["metric_value"] = metric_value
            analysis_results.append(result_row)

    if not analysis_results:
        return pd.DataFrame()

    summary_df = pd.DataFrame(analysis_results)

    # Dynamically identify all parameter columns from the dataframe
    param_cols = [
        col for col in summary_df.columns if col not in ["metric_name", "metric_value"]
    ]

    if not param_cols:
        return pd.DataFrame()

    try:
        pivot_df = summary_df.pivot_table(
            index=param_cols, columns="metric_name", values="metric_value"
        ).reset_index()
        return pivot_df
    except Exception as e:
        print(f"Error during pivoting: {e}")
        return pd.DataFrame()

get_final_value(series, time_series=None)

Gets the final value of a time series.

Parameters:

Name	Type	Description	Default
series	Series	The time series data.	required
time_series	Optional[Series]	The corresponding time data (unused).	None

Returns:

Type	Description
float	The last value in the series.

Note

The time_series parameter is kept for interface consistency but is not used in the calculation. Only the series data is required.

Source code in tricys/analysis/metric.py

def get_final_value(
    series: pd.Series, time_series: Optional[pd.Series] = None
) -> float:
    """Gets the final value of a time series.

    Args:
        series: The time series data.
        time_series: The corresponding time data (unused).

    Returns:
        The last value in the series.

    Note:
        The time_series parameter is kept for interface consistency but is not used
        in the calculation. Only the series data is required.
    """
    return series.iloc[-1]

time_of_turning_point(series, time_series)

Finds the time of the turning point (minimum value) in a series.

This function identifies the time corresponding to the minimum value in the series, which often represents the self-sufficiency time in tritium inventory simulations. To handle noisy data, it first smooths the series to find the general trend's minimum. If the smoothed minimum is not at the boundaries, it returns the time of the absolute minimum from the original data.

Parameters:

Name	Type	Description	Default
series	Series	The time series data to analyze.	required
time_series	Series	The corresponding time data.	required

Returns:

Type	Description
float	The time of the turning point, or NaN if the trend is monotonic.

Raises:

Type	Description
ValueError	If time_series is None.

Note

Uses a rolling window (0.1% of data length) for smoothing to identify the general trend. If the smoothed minimum is within the last 30% of the series, the trend is considered monotonic and NaN is returned. Otherwise, returns the time of the absolute minimum in the original data.

Source code in tricys/analysis/metric.py

def time_of_turning_point(series: pd.Series, time_series: pd.Series) -> float:
    """Finds the time of the turning point (minimum value) in a series.

    This function identifies the time corresponding to the minimum value in the
    series, which often represents the self-sufficiency time in tritium inventory
    simulations. To handle noisy data, it first smooths the series to find the
    general trend's minimum. If the smoothed minimum is not at the boundaries,
    it returns the time of the absolute minimum from the original data.

    Args:
        series: The time series data to analyze.
        time_series: The corresponding time data.

    Returns:
        The time of the turning point, or NaN if the trend is monotonic.

    Raises:
        ValueError: If time_series is None.

    Note:
        Uses a rolling window (0.1% of data length) for smoothing to identify the
        general trend. If the smoothed minimum is within the last 30% of the series,
        the trend is considered monotonic and NaN is returned. Otherwise, returns
        the time of the absolute minimum in the original data.
    """

    print(
        f"Calculating time_of_turning_point for series with length {len(series)} and {len(time_series)} "
    )
    if time_series is None:
        raise ValueError("time_series must be provided for time_of_turning_point")

    # Define a window size for the rolling average, e.g., 5% of the data length
    # with a minimum size of 1. This helps in smoothing out local fluctuations.
    window_size = max(1, int(len(series) * 0.001))
    smoothed_series = series.rolling(
        window=window_size, center=True, min_periods=1
    ).mean()

    # Find the index label of the minimum value in the smoothed series.
    smooth_min_index = smoothed_series.idxmin()
    min_index = series.idxmin()

    # Check if the minimum of the smoothed data is within the first or last 5%
    # of the series. If so, the trend is considered monotonic.
    smooth_min_pos = series.index.get_loc(smooth_min_index)
    five_percent_threshold = int(len(series) * 0.3)

    if smooth_min_pos >= len(series) - five_percent_threshold:
        return np.nan
    else:
        # A clear turning point is identified in the overall trend.
        # Now, find the precise turning point in the original, noisy data.
        min_index = series.idxmin()
        return time_series.loc[min_index]

Utility functions for plotting simulation results.

This module provides functions to generate plots from the simulation output CSV files, such as visualizing startup tritium inventory or time-series data.

generate_analysis_plots(summary_df, analysis_case, save_dir, unit_map=None, glossary_path=None)

Generates and saves plots based on the sensitivity analysis summary.

This function first generates dedicated plots for all 'Required_***' metrics, then handles plotting for all other standard metrics.

Parameters:

Name	Type	Description	Default
summary_df	DataFrame	DataFrame containing the summarized analysis results.	required
analysis_case	dict	Configuration for the analysis cases.	required
save_dir	str	Directory to save the plot images.	required
unit_map	dict	Optional dictionary for unit conversion and labeling. Defaults to None.	None
glossary_path	str	Optional path to the glossary CSV file for professional labels.	None

Returns:

Type	Description
list[str]	A list of paths to the saved plot images.

Note

Handles Required_*** metrics separately with multi-subplot layouts. Standard metrics can be combined or plotted individually based on case configuration. Returns empty list if summary_df is empty. Loads glossary if path provided.

Source code in tricys/analysis/plot.py

def generate_analysis_plots(
    summary_df: pd.DataFrame,
    analysis_case: dict,
    save_dir: str,
    unit_map: dict = None,
    glossary_path: str = None,
) -> list[str]:
    """Generates and saves plots based on the sensitivity analysis summary.

    This function first generates dedicated plots for all 'Required_***' metrics,
    then handles plotting for all other standard metrics.

    Args:
        summary_df: DataFrame containing the summarized analysis results.
        analysis_case: Configuration for the analysis cases.
        save_dir: Directory to save the plot images.
        unit_map: Optional dictionary for unit conversion and labeling. Defaults to None.
        glossary_path: Optional path to the glossary CSV file for professional labels.

    Returns:
        A list of paths to the saved plot images.

    Note:
        Handles Required_*** metrics separately with multi-subplot layouts. Standard
        metrics can be combined or plotted individually based on case configuration.
        Returns empty list if summary_df is empty. Loads glossary if path provided.
    """
    glossary_maps = load_glossary(glossary_path) if glossary_path else ({}, {})

    if summary_df.empty:
        return []

    analysis_cases = [analysis_case]  # Keep as a list for consistency
    sns.set_theme(style="whitegrid")
    line_colors = sns.color_palette("viridis", 10)

    plot_paths = []

    # If unit_map is not provided, initialize as empty dict
    if unit_map is None:
        unit_map = {}

    # --- 1. Handle ALL 'Required_***' plots first and separately ---
    all_required_vars_from_config = {
        var
        for case in analysis_cases
        for var in case.get("dependent_variables", [])
        if var.startswith("Required_")
    }

    for req_var in all_required_vars_from_config:
        # Find all actual columns in the dataframe for this base name
        matching_cols = sorted(
            [
                c
                for c in summary_df.columns
                if c == req_var or c.startswith(req_var + "(")
            ]
        )

        if not matching_cols:
            continue

        case_for_plot = analysis_cases[0]

        if len(matching_cols) > 1:
            # Multi-value case -> generate a multi-subplot figure
            multi_plot_paths = _generate_multi_required_plot(
                summary_df,
                case_for_plot,
                matching_cols,
                req_var,
                save_dir,
                unit_map=unit_map,
                glossary_maps=glossary_maps,
            )
            plot_paths.extend(multi_plot_paths)
        elif len(matching_cols) == 1:
            # Single-value case -> generate a single, individual plot
            plot_config = {
                "case_name": case_for_plot["name"],
                "x_var": case_for_plot["independent_variable"],
                "y_var": matching_cols[0],
                "plot_type": "line",
                "hue_vars": sorted(
                    list(case_for_plot.get("default_simulation_values", {}).keys())
                ),
            }
            single_plot_path = _generate_individual_plots(
                summary_df,
                [plot_config],
                save_dir,
                line_colors,
                unit_map=unit_map,
                glossary_maps=glossary_maps,
            )
            plot_paths.extend(single_plot_path)

    # --- 2. Handle all other (non-Required) plots ---

    # Collect plot configurations for remaining standard variables
    valid_plots_for_combine = []
    for case in analysis_cases:
        case_name = case["name"]
        x_var = case["independent_variable"]

        # Filter out the Required_*** vars that we just plotted
        y_vars = [
            v
            for v in case.get("dependent_variables", [])
            if not v.startswith("Required_")
        ]

        if x_var not in summary_df.columns:
            print(
                f"Warning: Independent variable '{x_var}' not found in summary data for case '{case_name}'. Skipping."
            )
            continue

        case_sim_params = case.get("default_simulation_values", {})
        hue_vars = sorted(list(case_sim_params.keys()))

        for y_var in y_vars:
            if y_var not in summary_df.columns:
                print(
                    f"Warning: Dependent variable '{y_var}' not found in summary data for case '{case_name}'. Skipping."
                )
                continue

            # We assume standard metrics have a 1-to-1 name match in the dataframe
            valid_plots_for_combine.append(
                {
                    "case_name": case_name,
                    "x_var": x_var,
                    "y_var": y_var,
                    "plot_type": "line",
                    "hue_vars": hue_vars,
                }
            )

    if valid_plots_for_combine:
        combine_plots = any(case.get("combine_plots", False) for case in analysis_cases)
        generated_paths = []
        if combine_plots:
            generated_paths = _generate_combined_plots(
                summary_df,
                valid_plots_for_combine,
                save_dir,
                line_colors,
                unit_map=unit_map,
                glossary_maps=glossary_maps,
            )
        else:
            # If not combining, plot them individually anyway
            generated_paths = _generate_individual_plots(
                summary_df,
                valid_plots_for_combine,
                save_dir,
                line_colors,
                unit_map=unit_map,
                glossary_maps=glossary_maps,
            )
        plot_paths.extend(generated_paths)

    return plot_paths

load_glossary(glossary_path)

Loads glossary data from a CSV file.

The CSV file should contain columns for the model parameter, the English term, and the Chinese translation. This data is used to format plot labels.

Parameters:

Name	Type	Description	Default
glossary_path	str	The path to the glossary CSV file.	required

Returns:

Type	Description
tuple[dict, dict]	A tuple containing two dictionaries: (english_glossary_map, chinese_glossary_map)

Source code in tricys/analysis/plot.py

def load_glossary(glossary_path: str) -> tuple[dict, dict]:
    """Loads glossary data from a CSV file.

    The CSV file should contain columns for the model parameter, the English
    term, and the Chinese translation. This data is used to format plot labels.

    Args:
        glossary_path: The path to the glossary CSV file.

    Returns:
        A tuple containing two dictionaries: (english_glossary_map, chinese_glossary_map)
    """
    english_glossary_map = {}
    chinese_glossary_map = {}

    if not glossary_path or not os.path.exists(glossary_path):
        print(
            f"Warning: Glossary file not found at {glossary_path}. No labels will be loaded."
        )
        return english_glossary_map, chinese_glossary_map

    try:
        df = pd.read_csv(glossary_path)
        if (
            "模型参数 (Model Parameter)" in df.columns
            and "英文术语 (English Term)" in df.columns
            and "中文翻译 (Chinese Translation)" in df.columns
        ):
            df.dropna(subset=["模型参数 (Model Parameter)"], inplace=True)
            english_glossary_map = pd.Series(
                df["英文术语 (English Term)"].values,
                index=df["模型参数 (Model Parameter)"],
            ).to_dict()
            chinese_glossary_map = pd.Series(
                df["中文翻译 (Chinese Translation)"].values,
                index=df["模型参数 (Model Parameter)"],
            ).to_dict()
            print(f"Successfully loaded glossary from {glossary_path}.")
        else:
            print("Warning: Glossary CSV does not contain expected columns.")
    except Exception as e:
        print(f"Warning: Failed to load or parse glossary file. Error: {e}")

    return english_glossary_map, chinese_glossary_map

plot_sweep_time_series(csv_path, save_dir, y_var_name, independent_var_name, independent_var_alias=None, default_params=None, glossary_path=None)

Generates a single figure with two subplots: an overall time-series view and a zoomed-in view.

The time axis is in days. The overall view hides data points for a curve if they exceed twice its initial value.

Parameters:

Name	Type	Description	Default
csv_path	str	Path to the scan result CSV file.	required
save_dir	str	Directory to save the image.	required
y_var_name	Union[str, List[str]]	Name(s) of the Y-axis variable(s).	required
independent_var_name	str	Full name of the scan parameter.	required
independent_var_alias	str	Alias for the scan parameter for cleaner plot titles.	None
default_params	Dict[str, Any]	A dictionary of default parameters. If provided, only curves matching these parameters will be plotted.	None
glossary_path	str	Path to the glossary file for professional labels.	None

Returns:

Type	Description
List[str]	A list of paths to the saved plot images, or an empty list on failure.

Note

Converts time from hours to days. Generates bilingual plots (English and Chinese). Overall view masks data exceeding 2x initial value. Zoomed view shows region from t=0 to 2 days past minimum, with red rectangle indicator on overall view. Data is converted from grams to kilograms for display.

Source code in tricys/analysis/plot.py

def plot_sweep_time_series(
    csv_path: str,
    save_dir: str,
    y_var_name: Union[str, List[str]],
    independent_var_name: str,
    independent_var_alias: str = None,
    default_params: Dict[str, Any] = None,
    glossary_path: str = None,
) -> List[str]:
    """Generates a single figure with two subplots: an overall time-series view and a zoomed-in view.

    The time axis is in days. The overall view hides data points for a curve if
    they exceed twice its initial value.

    Args:
        csv_path: Path to the scan result CSV file.
        save_dir: Directory to save the image.
        y_var_name: Name(s) of the Y-axis variable(s).
        independent_var_name: Full name of the scan parameter.
        independent_var_alias: Alias for the scan parameter for cleaner plot titles.
        default_params: A dictionary of default parameters. If provided, only curves
            matching these parameters will be plotted.
        glossary_path: Path to the glossary file for professional labels.

    Returns:
        A list of paths to the saved plot images, or an empty list on failure.

    Note:
        Converts time from hours to days. Generates bilingual plots (English and Chinese).
        Overall view masks data exceeding 2x initial value. Zoomed view shows region from
        t=0 to 2 days past minimum, with red rectangle indicator on overall view. Data is
        converted from grams to kilograms for display.
    """
    glossary_maps = load_glossary(glossary_path) if glossary_path else ({}, {})

    try:
        df = pd.read_csv(csv_path)
    except FileNotFoundError:
        print(f"Error: Could not find results file at {csv_path}")
        return []

    if "time" not in df.columns:
        print(f"Error: 'time' column not found in {csv_path}")
        return []

    # Convert time from hours to days
    time_days = df["time"] / 24

    # Use alias if provided, otherwise format the original name
    raw_plot_alias = (
        independent_var_alias if independent_var_alias else independent_var_name
    )

    if isinstance(y_var_name, str):
        y_var_names = [y_var_name]
    else:
        y_var_names = y_var_name

    y_var_columns = []
    for y_var in y_var_names:
        y_var_columns.extend(
            [col for col in df.columns if col != "time" and y_var in col]
        )
    y_var_columns = list(dict.fromkeys(y_var_columns))

    # If default_params are provided, filter columns to only plot baseline curves
    if default_params:
        filtered_columns = []
        for col in y_var_columns:
            try:
                param_str = col.split("&", 1)[1]
                col_params = dict(p.split("=", 1) for p in param_str.split("&"))

                # Check if all default_params match the parameters in the column name
                is_match = all(
                    col_params.get(key) == str(val)
                    for key, val in default_params.items()
                )

                if is_match:
                    filtered_columns.append(col)
            except IndexError:
                # This column does not have parameters in its name, so it can't be a match
                continue
        y_var_columns = filtered_columns

    if not y_var_columns:
        print(
            f"Warning: No columns found containing any of {y_var_names} in {csv_path} that match the criteria."
        )
        return []

    # Convert y-axis data from grams to kilograms
    for col in y_var_columns:
        df[col] = df[col] / 1000.0

    # Generate clean labels (values only) for the legend
    plot_labels = []
    for col in y_var_columns:
        label = col
        try:
            param_parts = col.split("&")[1:]
            for part in param_parts:
                if part.startswith(independent_var_name + "="):
                    label = part.split("=", 1)[1]  # Extract just the value
                    break
        except IndexError:
            pass  # No parameters in name, use full column name
        plot_labels.append(label)

    print(
        f"Found {len(y_var_columns)} columns to plot containing {y_var_names}: {y_var_columns}"
    )

    sns.set_theme(style="whitegrid")
    plot_paths = []
    original_lang_is_chinese = _use_chinese_labels

    for lang in ["en", "cn"]:
        set_plot_language(lang)

        colors = sns.color_palette("plasma", len(y_var_columns))

        # Create a figure with two subplots (overall and zoom)
        fig, (ax1, ax2) = plt.subplots(
            2, 1, figsize=(12, 16), sharex=False, gridspec_kw={"height_ratios": [2, 1]}
        )
        y_var_names_formatted = [_format_label(y, glossary_maps) for y in y_var_names]

        min_y_global = float("inf")
        min_x_global = float("inf")

        # Define the y-axis label with units
        y_label = f"{', '.join(y_var_names_formatted)} ({_get_text('kg')})"

        # --- Subplot 1: Overall View ---
        for i, column in enumerate(y_var_columns):
            y_data = df[column]

            # For the global view, mask data that is more than 2x the initial value
            if not y_data.empty:
                initial_value = y_data.iloc[0]
                threshold = 2 * initial_value
                y_masked = y_data.where(y_data <= threshold)
            else:
                y_masked = y_data

            ax1.plot(
                time_days,
                y_masked,
                label=plot_labels[i],
                color=colors[i],
                linewidth=1.2,
                alpha=0.85,
            )

            # Calculations for zoom window should use the original, unmasked data
            if not y_data.empty:
                min_idx = y_data.idxmin()
                current_min_y = y_data.loc[min_idx]
                if current_min_y < min_y_global:
                    min_y_global = current_min_y
                    min_x_global = time_days.loc[min_idx]

        ax1.set_ylabel(y_label, fontsize=12)
        ax1.set_title(_get_text("overall_view"), fontsize=12)
        ax1.legend(loc="best", title=_format_label(independent_var_name, glossary_maps))
        ax1.grid(True)

        # --- Subplot 2: Zoomed-in View (uses original data) ---
        if min_y_global != float("inf") and np.isfinite(min_y_global):
            for i, column in enumerate(y_var_columns):
                # Plot original, unmasked data in the zoom plot
                ax2.plot(
                    time_days,
                    df[column],
                    label=plot_labels[i],
                    color=colors[i],
                    linewidth=1.8,
                    alpha=0.9,
                )

            # Define the zoom window from t=0 to a bit after the minimum
            x1 = 0
            x2 = min_x_global + 2  # Show 2 days past the minimum

            # Filter the DataFrame to the new x-range to find the y-range
            zoom_mask = (time_days >= x1) & (time_days <= x2)
            df_zoom_range = df[zoom_mask]

            # Find y-min and y-max within this specific range
            y_min_in_range = df_zoom_range[y_var_columns].min().min()
            y_max_in_range = df_zoom_range[y_var_columns].max().max()

            # Add padding to the y-axis
            y_padding = (y_max_in_range - y_min_in_range) * 0.05
            y1 = y_min_in_range - y_padding
            y2 = y_max_in_range + y_padding

            ax2.set_xlim(x1, x2)
            ax2.set_ylim(y1, y2)

            ax2.set_xlabel(_get_text("time_days"), fontsize=12)
            ax2.set_ylabel(y_label, fontsize=12)
            ax2.set_title(_get_text("detailed_view"), fontsize=12)
            ax2.grid(True, linestyle="--")

            # Add a rectangle to the main plot to indicate the new zoom area
            rect = patches.Rectangle(
                (x1, y1),
                (x2 - x1),
                (y2 - y1),
                linewidth=1,
                edgecolor="r",
                facecolor="none",
                linestyle="--",
                alpha=0.7,
            )
            ax1.add_patch(rect)
        else:
            # If no zoom, hide the second subplot
            ax2.set_visible(False)

        ax1.set_xlabel(_get_text("time_days"), fontsize=12)
        fig.tight_layout(rect=[0, 0.03, 1, 0.95])  # Adjust for suptitle

        # --- Save Figure ---
        safe_y_vars = "_".join(
            [
                var.replace(".", "_").replace("[", "").replace("]", "")
                for var in y_var_names
            ]
        )
        safe_param = raw_plot_alias.replace(".", "_").replace("[", "").replace("]", "")
        suffix = "_zh" if lang == "cn" else ""
        svg_path = os.path.join(
            save_dir, f"sweep_{safe_y_vars}_vs_{safe_param}{suffix}.svg"
        )

        try:
            # Force text to be rendered as paths in SVG.
            plt.rcParams["svg.fonttype"] = "path"
            plt.savefig(svg_path, format="svg", bbox_inches="tight")
            print(f"Successfully generated combined sweep plot: {svg_path}")
            plot_paths.append(svg_path)
        except Exception as e:
            print(f"Error saving plot: {e}")
        finally:
            plt.close(fig)

    set_plot_language("cn" if original_lang_is_chinese else "en")
    return plot_paths

set_plot_language(lang='en')

Sets the preferred language for plot labels and text.

Parameters:

Name	Type	Description	Default
lang	str	The language to set. 'en' for English (default), 'cn' for Chinese.	'en'

Note

For Chinese language, sets font to SimHei and adjusts unicode_minus handling. For English, restores matplotlib default settings. Changes apply to all subsequent plots until called again.

Source code in tricys/analysis/plot.py

def set_plot_language(lang: str = "en") -> None:
    """Sets the preferred language for plot labels and text.

    Args:
        lang: The language to set. 'en' for English (default), 'cn' for Chinese.

    Note:
        For Chinese language, sets font to SimHei and adjusts unicode_minus handling.
        For English, restores matplotlib default settings. Changes apply to all
        subsequent plots until called again.
    """
    global _use_chinese_labels
    _use_chinese_labels = lang.lower() == "cn"

    if _use_chinese_labels:
        # To display Chinese characters correctly, specify a list of fallback fonts.
        plt.rcParams["font.sans-serif"] = ["SimHei"]  # 替换成你电脑上有的字体
        plt.rcParams["axes.unicode_minus"] = False  # To display minus sign correctly.
        plt.rcParams["font.family"] = "sans-serif"  # 确保字体家族设置生效
    else:
        # Restore default settings
        plt.rcParams["font.sans-serif"] = plt.rcParamsDefault["font.sans-serif"]
        plt.rcParams["axes.unicode_minus"] = plt.rcParamsDefault["axes.unicode_minus"]

call_openai_analysis_api(case_name, df, api_key, base_url, ai_model, independent_variable, report_content, original_config, case_data, reference_col_for_turning_point=None)

Constructs a text-only prompt, calls the OpenAI API for analysis, and returns the result string.

Parameters:

Name	Type	Description	Default
case_name	str	Name of the analysis case.	required
df	DataFrame	DataFrame containing summary data.	required
api_key	str	OpenAI API key.	required
base_url	str	Base URL for the OpenAI API.	required
ai_model	str	Model name to use for analysis.	required
independent_variable	str	Name of the independent variable.	required
report_content	str	The report content to analyze.	required
original_config	dict	Original configuration dictionary.	required
case_data	dict	Case-specific data dictionary.	required
reference_col_for_turning_point	str	Optional reference column for turning point analysis.	None

Returns:

Type	Description
Optional[str]	The combined prompt and LLM analysis result, or None if failed.

Note

Constructs dynamic prompts based on case configuration. Includes sections for global sensitivity analysis, interaction effects (if simulation parameters present), and dynamic process analysis (if reference column provided). Retries up to 3 times on failure with 5-second delays.

Source code in tricys/analysis/report.py

def call_openai_analysis_api(
    case_name: str,
    df: pd.DataFrame,
    api_key: str,
    base_url: str,
    ai_model: str,
    independent_variable: str,
    report_content: str,
    original_config: dict,
    case_data: dict,
    reference_col_for_turning_point: str = None,
) -> Optional[str]:
    """Constructs a text-only prompt, calls the OpenAI API for analysis, and returns the result string.

    Args:
        case_name: Name of the analysis case.
        df: DataFrame containing summary data.
        api_key: OpenAI API key.
        base_url: Base URL for the OpenAI API.
        ai_model: Model name to use for analysis.
        independent_variable: Name of the independent variable.
        report_content: The report content to analyze.
        original_config: Original configuration dictionary.
        case_data: Case-specific data dictionary.
        reference_col_for_turning_point: Optional reference column for turning point analysis.

    Returns:
        The combined prompt and LLM analysis result, or None if failed.

    Note:
        Constructs dynamic prompts based on case configuration. Includes sections for
        global sensitivity analysis, interaction effects (if simulation parameters present),
        and dynamic process analysis (if reference column provided). Retries up to 3 times
        on failure with 5-second delays.
    """
    try:
        logger.info(f"Proceeding with LLM analysis for case {case_name}.")

        # 1. Construct the prompt for the API
        role_prompt = """**角色：** 你是一名聚变反应堆氚燃料循环领域的专家。

**任务：** 请**完全基于**下方提供的**两类数据表格**，对聚变堆燃料循环模型的**敏感性分析**结果进行深度解读。
"""

        analysis_prompt = f"""
**分析数据：**(注意：分析中不可使用任何图表信息，所有结论必须源于数据表格。)

{report_content}
"""

        # --- Dynamic Prompt Construction ---

        # 1. Detect analysis scenario
        has_sim_params = bool(case_data.get("simulation_parameters"))

        # 2. Build prompt sections dynamically
        prompt_sections = []

        # Section 1: Global Sensitivity Analysis
        global_sensitivity_points = [
            "1.  **全局敏感性分析 (参考“性能指标总表”) :**",
            "    *   分析性能指标总表（ `Startup_Inventory`, `Doubling_Time` 以及以 `Required_` 开头的求解指标等）呈现出怎样的**总体趋势**？请进行量化描述。",
            f"    *   如果存在多个性能指标，分析哪个性能指标对独立变量 `{independent_variable}` 的变化最为敏感？哪个最不敏感？\n",
        ]

        # Interaction effect analysis, with refined description
        if has_sim_params:
            param_names_list = []
            for p in case_data["simulation_parameters"].keys():
                if p == "Required_TBR":
                    label = "`Required_TBR约束值 (hour)`"
                else:
                    label = f"`{p}`"
                param_names_list.append(label)
            param_names = ", ".join(param_names_list)

            interaction_text = (
                f"2.  **交互效应分析：** 本次分析包含了多变量的交互效应。请分析独立变量 `{independent_variable}` "
                f"与背景扫描参数 ({param_names}) 之间的交互作用对各项性能指标的影响。"
                "请注意，独立变量或背景扫描参数中，可能包含常规的模型参数，也可能包含为满足特定性能目标（限制倍增时间Double_Time达到倍增）而求解出的特殊变量（约束限制变量Double_Time）。"
                "请讨论在不同的变量组合下，性能指标的敏感性有何不同？是否存在显著的交互效应？"
            )
            global_sensitivity_points.append(interaction_text)

        prompt_sections.append("\n".join(global_sensitivity_points))

        # Section 2: Dynamic Process Analysis
        if reference_col_for_turning_point:
            dynamic_process_points = [
                "3.  **动态过程分析 (参考“关键动态数据切片：过程数据”) :**",
                "    *   观察过程数据切片：系统在“初始阶段”和“结束阶段”的行为有何不同？",
                f"    *   以 `{reference_col_for_turning_point}` 为参考，其“转折点阶段”的数据揭示了什么物理过程？（例如，它是否是氚库存由消耗转为净增长的关键时刻？）",
            ]
            prompt_sections.append("\n".join(dynamic_process_points))

        # Section 3: Overall Conclusion (renumbered from 4)
        conclusion_points = ["3.  **综合结论：**"]
        conclusion_intro = "结合所有分析（包括主趋势"
        if has_sim_params:
            conclusion_intro += "、背景参数交互效应"
        conclusion_intro += "），"

        conclusion_points.append(
            conclusion_intro
            + f"总结在不同的运行场景下，调整 `{independent_variable}` 对整个氚燃料循环系统的综合影响和潜在的利弊权衡。"
        )
        conclusion_points.append(
            "    *   基于这些发现，可以得出哪些关于系统设计或运行优化的初步建议？"
        )
        prompt_sections.append("\n".join(conclusion_points))

        # Assemble the final prompt
        points_prompt = "\n\n".join(prompt_sections)
        points_prompt = (
            "\n**分析要点 (必须严格依据数据表格作答)：**\n\n" + points_prompt
        )

        # 2. Call API with retry logic
        max_retries = 3
        for attempt in range(max_retries):
            try:
                client = openai.OpenAI(api_key=api_key, base_url=base_url)
                logger.info(
                    f"Sending request to OpenAI API for case {case_name} (Attempt {attempt + 1}/{max_retries})..."
                )

                full_text_prompt = "\n\n".join(
                    [role_prompt, analysis_prompt, points_prompt]
                )

                response = client.chat.completions.create(
                    model=ai_model,
                    messages=[{"role": "user", "content": full_text_prompt}],
                    max_tokens=4000,
                )
                analysis_result = response.choices[0].message.content

                logger.info(f"LLM analysis successful for case {case_name}.")
                return (
                    role_prompt
                    + points_prompt
                    + "\n```\n\n"
                    + "\n\n---\n\n# AI模型分析结果\n\n"
                    + analysis_result
                )  # Return the result string

            except Exception as e:
                logger.error(f"Error calling OpenAI API on attempt {attempt + 1}: {e}")
                if attempt < max_retries - 1:
                    time.sleep(5)
                else:
                    logger.error(
                        f"Failed to call OpenAI API after {max_retries} attempts."
                    )
                    return None  # Return None on failure

    except Exception as e:
        logger.error(
            f"Error in call_openai_analysis_api for case {case_name}: {e}",
            exc_info=True,
        )
        return None

consolidate_reports(case_configs, original_config)

Consolidates generated reports and their images into a 'report' directory for each case.

Parameters:

Name	Type	Description	Default
case_configs	List[Dict[str, Any]]	List of case configuration dictionaries.	required
original_config	Dict[str, Any]	Original configuration dictionary.	required

Note

Moves analysis reports, academic reports, and plot images from results directory to report directory. Uses move operation (not copy). Creates report directory if it doesn't exist. Skips cases where source directory not found.

Source code in tricys/analysis/report.py

def consolidate_reports(
    case_configs: List[Dict[str, Any]], original_config: Dict[str, Any]
) -> None:
    """Consolidates generated reports and their images into a 'report' directory for each case.

    Args:
        case_configs: List of case configuration dictionaries.
        original_config: Original configuration dictionary.

    Note:
        Moves analysis reports, academic reports, and plot images from results directory
        to report directory. Uses move operation (not copy). Creates report directory
        if it doesn't exist. Skips cases where source directory not found.
    """
    logger.info("Consolidating analysis reports...")
    try:
        for case_info in case_configs:
            case_workspace = case_info["workspace"]
            source_dir = os.path.join(case_workspace, "results")
            dest_dir = os.path.join(case_workspace, "report")

            if not os.path.isdir(source_dir):
                logger.warning(
                    f"Source directory not found, skipping consolidation for case: {case_workspace}"
                )
                continue

            # Find files to copy
            files_to_copy = []
            for filename in os.listdir(source_dir):
                if (
                    filename.startswith("analysis_report")
                    or filename.startswith("academic_report")
                ) and filename.endswith(".md"):
                    files_to_copy.append(filename)
                elif filename.endswith((".svg", ".png")):
                    files_to_copy.append(filename)

            if not files_to_copy:
                logger.info(
                    f"No reports or images found in {source_dir}, skipping consolidation."
                )
                continue

            # Create destination directory and copy files
            os.makedirs(dest_dir, exist_ok=True)
            logger.info(f"Consolidating reports into: {dest_dir}")

            for filename in files_to_copy:
                source_path = os.path.join(source_dir, filename)
                shutil.move(source_path, dest_dir)
                logger.info(f"Moved {filename} to {dest_dir}")

    except Exception as e:
        logger.error(f"Error during report consolidation: {e}", exc_info=True)

generate_analysis_cases_summary(case_configs, original_config)

Generate summary report for analysis_cases.

Parameters:

Name	Type	Description	Default
case_configs	List[Dict[str, Any]]	List of case configuration dictionaries.	required
original_config	Dict[str, Any]	Original configuration dictionary containing run timestamp.	required

Note

Creates an execution report with basic information, case details, and status. Saves report to {run_timestamp}/execution_report_{run_timestamp}.md in current working directory. Also triggers generate_prompt_templates and consolidate_reports. Logs summary of successfully executed cases.

Source code in tricys/analysis/report.py

def generate_analysis_cases_summary(
    case_configs: List[Dict[str, Any]], original_config: Dict[str, Any]
) -> None:
    """Generate summary report for analysis_cases.

    Args:
        case_configs: List of case configuration dictionaries.
        original_config: Original configuration dictionary containing run timestamp.

    Note:
        Creates an execution report with basic information, case details, and status.
        Saves report to {run_timestamp}/execution_report_{run_timestamp}.md in current
        working directory. Also triggers generate_prompt_templates and consolidate_reports.
        Logs summary of successfully executed cases.
    """
    try:
        run_timestamp = original_config["run_timestamp"]
        # Generate report in current working directory
        current_dir = os.getcwd()

        # Create summary report
        summary_data = []
        for case_info in case_configs:
            case_data = case_info["case_data"]
            case_workspace = case_info["workspace"]

            # Check if case results exist
            case_results_dir = os.path.join(case_workspace, "results")
            has_results = (
                os.path.exists(case_results_dir)
                and len(os.listdir(case_results_dir)) > 0
            )

            summary_entry = {
                "case_name": case_data.get("name", f"Case{case_info['index']+1}"),
                "independent_variable": case_data["independent_variable"],
                "independent_variable_sampling": case_data[
                    "independent_variable_sampling"
                ],
                "workspace_path": case_workspace,
                "has_results": has_results,
                "config_file": case_info["config_path"],
            }
            summary_data.append(summary_entry)

        # Generate text report
        report_lines = [
            "# Analysis Cases Execution Report",
            "\n## Basic Information",
            f"- Execution time: {run_timestamp}",
            f"- Total cases: {len(case_configs)}",
            f"- Successfully executed: {sum(1 for entry in summary_data if entry['has_results'])}",
            f"- Working directory: {current_dir}",
            "\n## Case Details",
        ]

        for i, entry in enumerate(summary_data, 1):
            status = "✓ Success" if entry["has_results"] else "✗ Failed"
            report_lines.extend(
                [
                    f"\n### {i}. {entry['case_name']}",
                    f"- Status: {status}",
                    f"- Independent variable: {entry['independent_variable']}",
                    f"- Sampling method: {entry['independent_variable_sampling']}",
                    f"- Working directory: {entry['workspace_path']}",
                    f"- Configuration file: {entry['config_file']}",
                ]
            )

        # Save report to current directory
        report_path = os.path.join(
            current_dir,
            run_timestamp,
            f"execution_report_{run_timestamp}.md",
        )
        os.makedirs(os.path.dirname(report_path), exist_ok=True)
        with open(report_path, "w", encoding="utf-8") as f:
            f.write("\n".join(report_lines))

        logger.info("Summary report generated:")
        logger.info(f"  - Detailed report: {report_path}")

        # Generate prompt engineering template for each case
        generate_prompt_templates(case_configs, original_config)

        # Consolidate all generated reports
        consolidate_reports(case_configs, original_config)

    except Exception as e:
        logger.error(f"Error generating summary report: {e}", exc_info=True)

generate_prompt_templates(case_configs, original_config)

Generate detailed Markdown analysis reports for each analysis case.

Parameters:

Name	Type	Description	Default
case_configs	List[Dict[str, Any]]	List of case configuration dictionaries containing case data and workspace info.	required
original_config	Dict[str, Any]	Original configuration dictionary with sensitivity analysis settings.	required

Note

Skips SALib cases (those with analyzer.method defined). For each case, generates a detailed Markdown report including configuration details, optimization configs, time-series plots, performance metric plots, and data tables. Supports AI-enhanced reporting if API credentials are available. Creates bilingual plots prioritizing Chinese versions (_zh suffix).

Source code in tricys/analysis/report.py

def generate_prompt_templates(
    case_configs: List[Dict[str, Any]], original_config: Dict[str, Any]
) -> None:
    """Generate detailed Markdown analysis reports for each analysis case.

    Args:
        case_configs: List of case configuration dictionaries containing case data and workspace info.
        original_config: Original configuration dictionary with sensitivity analysis settings.

    Note:
        Skips SALib cases (those with analyzer.method defined). For each case, generates
        a detailed Markdown report including configuration details, optimization configs,
        time-series plots, performance metric plots, and data tables. Supports AI-enhanced
        reporting if API credentials are available. Creates bilingual plots prioritizing
        Chinese versions (_zh suffix).
    """

    def _find_unit_config(var_name: str, unit_map: dict) -> dict | None:
        """
        Finds the unit configuration for a variable name from the unit_map.
        1. Checks for an exact match.
        2. Checks if the last part of a dot-separated name matches.
        3. Checks for a simple substring containment as a fallback, matching longest keys first.
        """
        if not unit_map or not var_name:
            return None
        if var_name in unit_map:
            return unit_map[var_name]
        components = var_name.split(".")
        if len(components) > 1 and components[-1] in unit_map:
            return unit_map[components[-1]]
        for key in sorted(unit_map.keys(), key=len, reverse=True):
            if key in var_name:
                return unit_map[key]
        return None

    def _format_label(label: str) -> str:
        """Formats a label for display, replacing underscores/dots with spaces and capitalizing each word."""
        if not isinstance(label, str):
            return label
        label = label.replace("_", " ")
        label = re.sub(r"(?<!\d)\.|\.(?!\d)", " ", label)
        return label  # .title()

    try:
        sensitivity_analysis_config = original_config.get("sensitivity_analysis", {})
        unit_map = sensitivity_analysis_config.get("unit_map", {})

        for case_info in case_configs:
            case_data = case_info["case_data"]

            if "analyzer" in case_data and case_data.get("analyzer", {}).get("method"):
                logger.info(
                    f"Skipping default report generation for SALib case: {case_data.get('name', 'Unknown')}"
                )
                continue

            case_workspace = case_info["workspace"]
            case_name = case_data.get("name", f"Case{case_info['index']+1}")

            case_results_dir = os.path.join(case_workspace, "results")
            if not os.path.exists(case_results_dir):
                continue

            summary_csv_path = os.path.join(
                case_results_dir, "sensitivity_analysis_summary.csv"
            )
            sweep_csv_path = os.path.join(case_results_dir, "sweep_results.csv")

            if not os.path.exists(summary_csv_path):
                logger.warning(
                    f"summary_csv not found for case {case_name}, skipping report generation."
                )
                continue

            summary_df = pd.read_csv(summary_csv_path)
            independent_variable = case_data.get("independent_variable", "燃烧率")

            # Use a dictionary to ensure we only get one version of each plot, prioritizing Chinese
            all_plots_all_langs = [
                f for f in os.listdir(case_results_dir) if f.endswith(".svg")
            ]
            plot_map = {}
            for plot in sorted(
                all_plots_all_langs, reverse=True
            ):  # Process _zh.svg first
                base_name = plot.replace("_zh.svg", ".svg")
                if base_name not in plot_map:
                    plot_map[base_name] = plot

            all_plots = list(plot_map.values())
            sweep_plots = [f for f in all_plots if f.startswith("sweep_")]
            combined_plots = [f for f in all_plots if f.startswith("combined_")]
            multi_metric_plots = [
                f
                for f in all_plots
                if f.startswith("multi_") and f.endswith("_analysis_by_param.svg")
            ]
            all_individual_plots = [
                f
                for f in all_plots
                if not f.startswith("sweep_")
                and not f.startswith("combined_")
                and not f.startswith("multi_")
            ]
            required_individual_plots = [
                f for f in all_individual_plots if f.startswith("line_Required_")
            ]
            standard_individual_plots = [
                f for f in all_individual_plots if not f.startswith("line_Required_")
            ]

            # --- Markdown Generation (with dynamic title) ---
            sim_params = case_data.get("simulation_parameters")
            if sim_params:
                main_var_label = _format_label(independent_variable)

                other_vars_labels_list = []
                for p in sim_params.keys():
                    if p == "Required_TBR":
                        label = "Required_TBR约束值"
                    else:
                        label = _format_label(p)
                    other_vars_labels_list.append(label)
                other_vars_labels = "、".join(other_vars_labels_list)

                report_title = (
                    f"# {main_var_label} 与 {other_vars_labels} 交互敏感性分析报告\n\n"
                )
            else:
                report_title = (
                    f"# {_format_label(independent_variable)} 敏感性分析报告\n\n"
                )

            prompt_lines = [
                report_title,
                f"生成时间: {pd.Timestamp.now()}\n\n",
            ]

            config_details_lines = [
                "## 分析案例配置详情\n\n",
                "本分析案例的具体配置如下，这决定了仿真的扫描方式和分析的重点：\n\n",
                "| 配置项 | 值 | 说明 |",
                "| :--- | :--- | :--- |",
            ]

            def format_for_md(value):
                return f"`{json.dumps(value, ensure_ascii=False)}`".replace("|", "\\|")

            config_details_lines.extend(
                [
                    f"| **`name`** | {format_for_md(case_name)} | 本次分析案例的名称。 |",
                    f"| **`independent_variable`** | {format_for_md(independent_variable)} | 独立扫描变量，即本次分析中主要改变的参数。 |",
                    f"| **`independent_variable_sampling`** | {format_for_md(case_data.get('independent_variable_sampling'))} | 独立变量的采样方法和范围。 |",
                ]
            )
            if "default_independent_values" in case_data:
                config_details_lines.append(
                    f"| **`default_independent_values`** | {format_for_md(case_data['default_independent_values'])} | 独立扫描变量在模型中的原始默认值。 |"
                )
            if (
                "simulation_parameters" in case_data
                and case_data["simulation_parameters"]
            ):
                config_details_lines.append(
                    f"| **`simulation_parameters`** | {format_for_md(case_data['simulation_parameters'])} | 背景扫描参数，与独立变量组合形成多维扫描。 |"
                )
            if (
                "default_simulation_values" in case_data
                and case_data["default_simulation_values"]
            ):
                config_details_lines.append(
                    f"| **`default_simulation_values`** | {format_for_md(case_data['default_simulation_values'])} | 背景扫描参数在模型中的原始默认值。 |"
                )
            config_details_lines.append(
                f"| **`dependent_variables`** | {format_for_md(case_data.get('dependent_variables'))} | 因变量，即我们关心的、随自变量变化的性能指标。 |"
            )
            config_details_lines.append("\n")
            prompt_lines.extend(config_details_lines)

            optimization_metrics = [
                v
                for v in case_data.get("dependent_variables", [])
                if v.startswith("Required_")
            ]
            if optimization_metrics:
                for metric_name in optimization_metrics:
                    metric_config = (
                        original_config.get("sensitivity_analysis", {})
                        .get("metrics_definition", {})
                        .get(metric_name)
                    )
                    if metric_config:
                        details_lines = [
                            f"## “{metric_name}”优化配置\n",
                            f"当“{metric_name}”作为因变量时，系统会启用一个二分查找算法来寻找满足特定性能指标的最小`{metric_config.get('parameter_to_optimize', 'N/A')}`值。以下是本次优化任务的具体配置：\n\n",
                            "| 配置项 | 值 | 说明 |",
                            "| :--- | :--- | :--- |",
                        ]
                        config_map = {
                            "source_column": "限制条件的数据源列。",
                            "parameter_to_optimize": "优化的目标参数。",
                            "search_range": "参数的搜索范围。",
                            "tolerance": "搜索的收敛精度。",
                            "max_iterations": "最大迭代次数。",
                            "metric_name": "限制条件的性能指标。",
                            "metric_max_value": "限制条件满足的上限值。（hour）",
                        }
                        for key, description in config_map.items():
                            if key in metric_config:
                                value = metric_config[key]
                                details_lines.append(
                                    f"| **`{key}`** | {format_for_md(value)} | {description} |"
                                )

                            metric_config_sim = case_data.get(
                                "simulation_parameters", {}
                            ).get("Required_TBR", {})
                            if metric_config_sim and key in metric_config_sim:
                                value = metric_config_sim[key]
                                details_lines.append(
                                    f"| **`{key} (from simulation_parameters)`** | {format_for_md(value)} | {description} |"
                                )

                        details_lines.append("\n")
                        prompt_lines.extend(details_lines)

            for plot in sweep_plots:
                prompt_lines.extend(
                    [
                        "## SDS Inventory 的时间曲线图:\n\n",
                        f"![SDS Inventory 的时间曲线图]({plot})\n\n",
                    ]
                )
                if "default_simulation_values" in case_data and case_data.get(
                    "default_simulation_values"
                ):
                    default_values_str = json.dumps(
                        case_data["default_simulation_values"],
                        ensure_ascii=False,
                        indent=4,
                    )
                    note = (
                        "**筛选说明**：当存在多个背景扫描参数 (`simulation_parameters`) 时，为突出重点，上图默认仅显示与原始默认值 "
                        f"(`default_simulation_values`) 相匹配的基准情景曲线。本次分析中用于筛选的默认值为：\n\n"
                        f"```json\n{default_values_str}\n```\n\n"
                        "此方法有助于在固定的基准条件下，清晰地观察独立变量变化带来的影响。\n"
                    )
                    prompt_lines.append(note)

            if combined_plots:
                for plot in combined_plots:
                    title = "性能指标趋势曲线图"
                    prompt_lines.extend([f"## {title}\n\n", f"![{title}]({plot})\n"])
            elif standard_individual_plots:
                prompt_lines.append("## 性能指标分析图\n\n")
                for plot in standard_individual_plots:
                    title = _format_label(
                        os.path.splitext(plot)[0].replace("line_", "")
                    )
                    prompt_lines.extend([f"### {title}\n", f"![{title}]({plot})\n\n"])

            if multi_metric_plots or required_individual_plots:
                prompt_lines.append("## 约束求解性能指标分析图\n\n")
                for plot_file in multi_metric_plots:
                    try:
                        base_metric_name = plot_file.replace("multi_", "").replace(
                            "_analysis_by_param.svg", ""
                        )
                        friendly_name = _format_label(base_metric_name)
                    except Exception:
                        friendly_name = "Optimization"
                    prompt_lines.extend(
                        [
                            f"### 不同约束值下的“{friendly_name}”分析 (按参数分组)\n",
                            f"下图展示了“{friendly_name}”指标随独立变量变化的趋势。每个子图对应一组特定的背景扫描参数组合，子图内的每条曲线代表一个具体的约束值。\n\n",
                            f"![不同约束值下的{friendly_name}分析]({plot_file})\n\n",
                        ]
                    )
                for plot_file in required_individual_plots:
                    title = _format_label(
                        os.path.splitext(plot_file)[0].replace("line_", "")
                    )
                    prompt_lines.extend(
                        [f"### {title}\n", f"![{title}]({plot_file})\n\n"]
                    )

            def _format_df_to_md(
                sub_df: pd.DataFrame,
                ind_var: str,
                case_data: dict,
                current_unit_map: dict,
            ) -> str:
                if sub_df.empty:
                    return "无数据。"
                all_markdown_lines = []
                all_cols = sub_df.columns.tolist()
                if ind_var in all_cols:
                    all_cols.remove(ind_var)
                standard_cols = [
                    c
                    for c in all_cols
                    if not (c.startswith("Required_") or "_for_Required_" in c)
                ]
                required_groups = {}
                required_base_names = [
                    v
                    for v in case_data.get("dependent_variables", [])
                    if v.startswith("Required_")
                ]
                for base_name in required_base_names:
                    group_cols = []
                    # pattern = re.compile(f"_for_{re.escape(base_name)}(?:\\(.*\\))?$")
                    for col in all_cols:
                        if col == base_name or col.startswith(base_name + "("):
                            group_cols.append(col)
                    if group_cols:
                        required_groups[base_name] = group_cols

                def _format_slice_to_md(df_slice: pd.DataFrame, umap: dict) -> str:
                    if df_slice.empty:
                        return ""
                    df_formatted = df_slice.copy()
                    new_columns = {}
                    for col_name in df_formatted.columns:
                        unit_config = _find_unit_config(col_name, umap)
                        new_col_name = _format_label(col_name)
                        if unit_config:
                            unit = unit_config.get("unit")
                            factor = unit_config.get("conversion_factor")
                            if factor and pd.api.types.is_numeric_dtype(
                                df_formatted[col_name]
                            ):
                                df_formatted[col_name] = df_formatted[col_name] / float(
                                    factor
                                )
                            if unit:
                                new_col_name = f"{new_col_name} ({unit})"
                        new_columns[col_name] = new_col_name
                    df_formatted.rename(columns=new_columns, inplace=True)
                    format_map = {}
                    for original_col_name in df_slice.columns:
                        if original_col_name.startswith("Required_"):
                            format_map[new_columns[original_col_name]] = "{:.4f}"
                    default_format = "{:.2f}"
                    for col in df_formatted.columns:
                        if pd.api.types.is_numeric_dtype(df_formatted[col]):
                            formatter = format_map.get(col, default_format)
                            df_formatted[col] = df_formatted[col].apply(
                                lambda x: formatter.format(x) if pd.notnull(x) else x
                            )
                    return df_formatted.to_markdown(index=False)

                if standard_cols:
                    all_markdown_lines.append("##### 性能指标\n")
                    std_df_slice = sub_df[[ind_var] + sorted(standard_cols)]
                    all_markdown_lines.append(
                        _format_slice_to_md(std_df_slice, current_unit_map)
                    )
                    all_markdown_lines.append("\n")
                if required_groups:
                    for base_name, cols in required_groups.items():
                        existing_cols = [c for c in cols if c in sub_df.columns]
                        if not existing_cols:
                            continue

                        all_markdown_lines.append(
                            f"##### “{_format_label(base_name)}” 相关数据\n"
                        )
                        req_df_slice = sub_df[[ind_var] + sorted(existing_cols)]

                        try:
                            # --- PIVOT LOGIC to transform data from wide to long format ---
                            # e.g., from [A, B(v1), B(v2)] to [A, new_col, B]

                            # Columns to unpivot, e.g., ['Required_TBR(7.0)', 'Required_TBR(10.0)']
                            value_vars = [
                                c
                                for c in req_df_slice.columns
                                if c.startswith(base_name)
                                and "(" in c
                                and c.endswith(")")
                            ]

                            # If no columns are in the format B(v), pivot is not applicable.
                            if not value_vars:
                                all_markdown_lines.append(
                                    _format_slice_to_md(req_df_slice, current_unit_map)
                                )
                                all_markdown_lines.append("\n")
                                continue

                            # Melt the dataframe from wide to long format
                            melted_df = req_df_slice.melt(
                                id_vars=[ind_var],
                                value_vars=value_vars,
                                var_name="variable_col",
                                value_name=base_name,
                            )

                            # Determine the name for the new column from config (e.g., 'Doubling_Time')
                            new_col_name = "Constraint"  # Default name
                            metric_def = case_data.get("simulation_parameters").get(
                                "Required_TBR"
                            )
                            if metric_def and metric_def.get("metric_name"):
                                new_col_name = "Constraint " + metric_def["metric_name"]

                            # Extract constraint value from old column name, e.g., '7.0' from 'Required_TBR(7.0)'
                            pattern_str = f"{re.escape(base_name)}\\((.*)\\)"
                            melted_df[new_col_name] = melted_df[
                                "variable_col"
                            ].str.extract(pat=pattern_str)

                            # Create the final dataframe with the desired columns: [A, new_col, B]
                            final_df = melted_df[
                                [ind_var, new_col_name, base_name]
                            ].copy()
                            final_df.dropna(subset=[base_name], inplace=True)

                            all_markdown_lines.append(final_df.to_markdown(index=False))
                            all_markdown_lines.append("\n")

                        except Exception as e:
                            logger.warning(
                                f"Could not pivot data for '{base_name}', displaying in wide format. Error: {e}"
                            )
                            all_markdown_lines.append(
                                _format_slice_to_md(req_df_slice, current_unit_map)
                            )
                            all_markdown_lines.append("\n")
                return "\n".join(all_markdown_lines)

            reference_col_for_turning_point = None
            if case_data.get("sweep_time") and os.path.exists(sweep_csv_path):
                try:
                    logger.info("Loading sweep_results.csv for dynamic slicing.")
                    sweep_df = pd.read_csv(sweep_csv_path)
                    if "time" in sweep_df.columns and len(sweep_df.columns) > 1:
                        reference_col_for_turning_point = sweep_df.columns[
                            len(sweep_df.columns) // 2
                        ]
                    if reference_col_for_turning_point:
                        data_to_slice_df = sweep_df.copy()
                        data_to_slice_df.reset_index(drop=True, inplace=True)
                        if not data_to_slice_df.empty:
                            prompt_lines.append("## 关键动态数据切片：过程数据\n\n")
                            prompt_lines.append(
                                f"下表展示了过程数据中，以 `{reference_col_for_turning_point}` 为参考变量，在关键阶段的数据切片。**注意：下表中的默认单位为：时间(h), 库存(g), 功率(MW)。**\n\n"
                            )
                            base_var_name = reference_col_for_turning_point.split("&")[
                                0
                            ]
                            cols_to_rename = [
                                c for c in data_to_slice_df.columns if c != "time"
                            ]
                            rename_map = {
                                col: f"C{i+1}" for i, col in enumerate(cols_to_rename)
                            }
                            legend_lines = [
                                "**表格图例说明**：",
                                "| 简称 | 参数组合 |",
                                "| :--- | :--- |",
                            ]
                            for original_name, abbr in rename_map.items():
                                param_parts = original_name.split("&", 1)
                                param_str = (
                                    param_parts[1] if len(param_parts) > 1 else "无"
                                )
                                param_str_formatted = (
                                    "`" + "`, `".join(param_str.split("&")) + "`"
                                )
                                legend_lines.append(
                                    f"| **{abbr}** | {param_str_formatted} |"
                                )
                            base_var_info = f"**注**：表格中所有简称列（C1, C2, ...）的数据均代表变量 `{base_var_name}` 在不同参数组合下的值。\n"
                            legend_md = base_var_info + "\n".join(legend_lines) + "\n\n"
                            prompt_lines.append(legend_md)
                            primary_y_var = reference_col_for_turning_point
                            min_idx = -1
                            if primary_y_var in data_to_slice_df.columns:
                                y_data = data_to_slice_df[primary_y_var]
                                if not y_data.empty:
                                    min_idx = y_data.idxmin()
                            num_points, interval = 20, 2
                            window_size = (num_points - 1) * interval + 1
                            start_data = data_to_slice_df.iloc[:window_size:interval]
                            end_data = data_to_slice_df.iloc[-(window_size)::interval]
                            prompt_lines.append(
                                f"### 1. 初始阶段 (前 {num_points} 个数据点, 间隔 {interval})\n"
                            )
                            prompt_lines.append(
                                start_data.rename(columns=rename_map).to_markdown(
                                    index=False
                                )
                                + "\n\n"
                            )
                            if min_idx != -1:
                                window_radius_indices = (num_points // 2) * interval
                                start_idx = max(0, min_idx - window_radius_indices)
                                end_idx = min(
                                    len(data_to_slice_df),
                                    min_idx + window_radius_indices,
                                )
                                turning_point_data = data_to_slice_df.iloc[
                                    start_idx:end_idx:interval
                                ]
                                prompt_lines.append(
                                    f"### 2. 转折点阶段 (围绕 '{primary_y_var}' 最小值)\n"
                                )
                                prompt_lines.append(
                                    turning_point_data.rename(
                                        columns=rename_map
                                    ).to_markdown(index=False)
                                    + "\n\n"
                                )
                            prompt_lines.append(
                                f"### 3. 结束阶段 (后 {num_points} 个数据点, 间隔 {interval})\n"
                            )
                            prompt_lines.append(
                                end_data.rename(columns=rename_map).to_markdown(
                                    index=False
                                )
                                + "\n\n"
                            )
                except Exception as e:
                    logger.warning(
                        f"Could not generate dynamic data slices for case {case_name}: {e}"
                    )

            grouping_vars = list(case_data.get("default_simulation_values", {}).keys())
            if not grouping_vars:
                prompt_lines.append("## 性能指标总表\n\n")
                prompt_lines.append(
                    _format_df_to_md(
                        summary_df, independent_variable, case_data, unit_map
                    )
                )
            else:
                prompt_lines.append(
                    f"## 性能指标总表 (分组: `{'`, `'.join(grouping_vars)}`)\n\n"
                )
                groups = dict(list(summary_df.groupby(grouping_vars)))
                default_values = case_data.get("default_simulation_values")
                default_group_key = None
                if default_values:
                    try:
                        default_group_key = tuple(
                            default_values[key] for key in grouping_vars
                        )
                    except KeyError:
                        logger.warning(
                            "Mismatch between default_simulation_values and grouping_vars. Cannot find default group."
                        )
                        default_group_key = None
                if default_group_key and default_group_key in groups:
                    default_group_df = groups.pop(default_group_key)
                    header = " & ".join(
                        f"`{var}={val}`"
                        for var, val in zip(grouping_vars, default_group_key)
                    )
                    prompt_lines.append(f"#### 数据子表 (原始默认值: {header})\n")
                    sub_df_to_format = default_group_df.drop(
                        columns=grouping_vars, errors="ignore"
                    )
                    prompt_lines.append(
                        _format_df_to_md(
                            sub_df_to_format, independent_variable, case_data, unit_map
                        )
                    )
                    prompt_lines.append("\n---\n")
                if groups:
                    prompt_lines.append("> 其他参数组合下的数据子表：\n")
                for group_name, group_df in groups.items():
                    header = (
                        " & ".join(
                            f"`{var}={val}`"
                            for var, val in zip(grouping_vars, group_name)
                        )
                        if isinstance(group_name, tuple)
                        else f"`{grouping_vars[0]}={group_name}`"
                    )
                    prompt_lines.append(f"#### 数据子表 (当 {header} 时)\n")
                    sub_df_to_format = group_df.drop(
                        columns=grouping_vars, errors="ignore"
                    )
                    prompt_lines.append(
                        _format_df_to_md(
                            sub_df_to_format, independent_variable, case_data, unit_map
                        )
                    )
                    prompt_lines.append("\n")

            base_report_content = "\n".join(prompt_lines)

            # --- AI Analysis and Report Writing ---
            if not case_data.get("ai", False):
                # AI is off: write a single, simple report
                report_path = os.path.join(
                    case_results_dir, f"analysis_report_{case_name}.md"
                )
                with open(report_path, "w", encoding="utf-8") as f:
                    f.write(base_report_content)
                logger.info(
                    f"Detailed analysis report generated for {case_name}: {report_path}"
                )
                continue  # Go to next case

            # AI is ON: go into multi-model logic
            load_dotenv()
            api_key = os.environ.get("API_KEY")
            base_url = os.environ.get("BASE_URL")

            ai_models_str = os.environ.get("AI_MODELS")
            if not ai_models_str:
                ai_models_str = os.environ.get("AI_MODEL")

            if not all((api_key, base_url, ai_models_str)):
                logger.warning(
                    "API_KEY, BASE_URL, or AI_MODELS/AI_MODEL not found in environment variables. Skipping LLM analysis."
                )
                # Also write the base report here so something is generated
                report_path = os.path.join(
                    case_results_dir, f"analysis_report_{case_name}.md"
                )
                with open(report_path, "w", encoding="utf-8") as f:
                    f.write(base_report_content)
                logger.info(
                    f"Wrote base report for {case_name} because AI credentials were not found: {report_path}"
                )
                continue

            ai_models = [model.strip() for model in ai_models_str.split(",")]

            for ai_model in ai_models:
                logger.info(
                    f"Generating AI analysis for case '{case_name}' with model '{ai_model}'."
                )

                sanitized_model_name = "".join(
                    c for c in ai_model if c.isalnum() or c in ("-", "_")
                ).rstrip()
                model_report_filename = (
                    f"analysis_report_{case_name}_{sanitized_model_name}.md"
                )
                model_report_path = os.path.join(
                    case_results_dir, model_report_filename
                )

                with open(model_report_path, "w", encoding="utf-8") as f:
                    f.write(base_report_content)
                logger.info(
                    f"Generated base report for model {ai_model}: {model_report_path}"
                )

                llm_analysis = call_openai_analysis_api(
                    case_name=case_name,
                    df=summary_df,
                    api_key=api_key,
                    base_url=base_url,
                    ai_model=ai_model,
                    independent_variable=independent_variable,
                    report_content=base_report_content,
                    original_config=original_config,
                    case_data=case_data,
                    reference_col_for_turning_point=reference_col_for_turning_point,
                )

                if llm_analysis:
                    with open(model_report_path, "a", encoding="utf-8") as f:
                        f.write(
                            f"\n\n---\n\n# AI模型分析提示词 ({ai_model})\n\n```markdown\n"
                        )
                        f.write(llm_analysis)
                        f.write("\n```\n")
                    logger.info(f"Appended LLM analysis to {model_report_path}")

                    generate_sensitivity_academic_report(
                        case_name=case_name,
                        case_workspace=case_workspace,
                        independent_variable=independent_variable,
                        original_config=original_config,
                        case_data=case_data,
                        ai_model=ai_model,
                        report_path=model_report_path,
                    )

    except Exception as e:
        logger.error(f"Error generating detailed analysis reports: {e}", exc_info=True)

generate_sensitivity_academic_report(case_name, case_workspace, independent_variable, original_config, case_data, ai_model, report_path)

Generates a professional academic analysis summary for a sensitivity analysis case.

Sends the existing report and a glossary of terms to an LLM for academic formatting.

Parameters:

Name	Type	Description	Default
case_name	str	Name of the analysis case.	required
case_workspace	str	Path to the case workspace directory.	required
independent_variable	str	Name of the independent variable.	required
original_config	dict	Original configuration dictionary.	required
case_data	dict	Case-specific data dictionary.	required
ai_model	str	Model name to use for generating the report.	required
report_path	str	Path to the existing report file.	required

Note

Requires report file and glossary file to exist. Loads API credentials from environment variables. Generates academic report with proper structure including title, abstract, introduction, methodology, results & discussion, and conclusion. Retries up to 3 times on API failure. Saves result to academic_report_{case_name}_{model}.md.

Source code in tricys/analysis/report.py

def generate_sensitivity_academic_report(
    case_name: str,
    case_workspace: str,
    independent_variable: str,
    original_config: dict,
    case_data: dict,
    ai_model: str,
    report_path: str,
) -> None:
    """Generates a professional academic analysis summary for a sensitivity analysis case.

    Sends the existing report and a glossary of terms to an LLM for academic formatting.

    Args:
        case_name: Name of the analysis case.
        case_workspace: Path to the case workspace directory.
        independent_variable: Name of the independent variable.
        original_config: Original configuration dictionary.
        case_data: Case-specific data dictionary.
        ai_model: Model name to use for generating the report.
        report_path: Path to the existing report file.

    Note:
        Requires report file and glossary file to exist. Loads API credentials from
        environment variables. Generates academic report with proper structure including
        title, abstract, introduction, methodology, results & discussion, and conclusion.
        Retries up to 3 times on API failure. Saves result to academic_report_{case_name}_{model}.md.
    """
    try:
        logger.info(
            f"Starting generation of the academic analysis summary for case {case_name} with model {ai_model}."
        )

        # 1. Read the existing report
        results_dir = os.path.join(case_workspace, "results")
        report_filename = os.path.basename(report_path)

        if not os.path.exists(report_path):
            logger.error(
                f"Cannot generate academic summary: Original report '{report_path}' not found."
            )
            return
        with open(report_path, "r", encoding="utf-8") as f:
            original_report_content = f.read()

        # 2. Read the glossary
        glossary_path = original_config.get("sensitivity_analysis", {}).get(
            "glossary_path", "./sheets.csv"
        )
        if not os.path.exists(glossary_path):
            logger.error(
                f"Cannot generate academic summary: Glossary file '{glossary_path}' not found."
            )
            return
        with open(glossary_path, "r", encoding="utf-8") as f:
            glossary_content = f.read()

        # 3. Check for API credentials
        load_dotenv()
        api_key = os.environ.get("API_KEY")
        base_url = os.environ.get("BASE_URL")

        if not all([api_key, base_url, ai_model]):
            logger.warning(
                "API_KEY, BASE_URL, or AI_MODEL not found. Skipping academic summary generation."
            )
            return

        # 4. Construct the prompt
        role_prompt = """**角色：** 您是一位在核聚变工程，特别是氚燃料循环领域，具有深厚学术背景的资深科学家。

**任务：** 您收到了一个关于**敏感性分析**的程序自动生成的初步报告和一份专业术语表。请您基于这两份文件，撰写一份更加专业、正式、符合学术发表标准的深度分析总结报告。
"""

        # Extract relevant details from case_data for the prompt
        sampling_range = case_data.get("independent_variable_sampling", {})
        simulation_parameters = case_data.get("simulation_parameters", {})
        dependent_variables = case_data.get("dependent_variables", [])

        simulation_params_str = ""
        if simulation_parameters:
            params_list = []
            # Access metric definitions from the original config
            metrics_definitions = original_config.get("sensitivity_analysis", {}).get(
                "metrics_definition", {}
            )

            for k, v in simulation_parameters.items():
                # Check if the parameter is a 'Required_' metric and has configurations defined
                if (
                    k.startswith("Required_")
                    and k in metrics_definitions
                    and "configurations" in metrics_definitions[k]
                ):
                    try:
                        metric_configs = metrics_definitions[k]["configurations"]
                        # Look up the metric_max_value for each configuration key in the scan list `v`
                        actual_values = [
                            metric_configs.get(conf_name, {}).get(
                                "metric_max_value", conf_name
                            )
                            for conf_name in v
                        ]
                        params_list.append(f"`{k}` (约束扫描值): {actual_values}")
                    except Exception:
                        # If lookup fails for any reason, fall back to the original representation
                        params_list.append(f"`{k}`: {v}")
                else:
                    # For regular parameters
                    params_list.append(f"`{k}`: {v}")

            simulation_params_str = (
                "\n        *   **背景扫描参数 (Simulation Parameters):** "
                + ", ".join(params_list)
            )

        dependent_vars_str = ""
        if dependent_variables:
            dependent_vars_str = (
                "\n        *   **因变量 (Dependent Variables):** "
                + ", ".join([f"`{v}`" for v in dependent_variables])
            )

        # Find all plots to instruct the LLM to include them, prioritizing Chinese versions
        all_files = [f for f in os.listdir(results_dir) if f.endswith((".svg", ".png"))]

        plot_map = {}
        # Handle SVGs, prioritizing _zh versions
        svg_plots = sorted([f for f in all_files if f.endswith(".svg")], reverse=True)
        for plot in svg_plots:
            base_name = plot.replace("_zh.svg", ".svg")
            if base_name not in plot_map:
                plot_map[base_name] = plot

        # Add PNGs (which are not bilingual)
        png_plots = [f for f in all_files if f.endswith(".png")]
        for plot in png_plots:
            plot_map[plot] = plot  # Use plot name as key for uniqueness

        all_plots = list(plot_map.values())
        plot_list_str = "\n".join([f"    *   `{plot}`" for plot in all_plots])

        # Dynamically build the "Results and Discussion" section for the prompt
        results_and_discussion_points = []

        # 1. Main Effect Analysis (always included)
        main_effect_text = (
            f"           *   **主效应分析：** 详细分析独立变量 **`{independent_variable}`** 的变化对主要性能指标（如 `Startup_Inventory`, `Doubling_Time` 等）的总体影响趋势。"
            "评估不同指标对自变量变化的敏感度，并讨论指标间的**权衡关系 (Trade-offs)**。"
        )
        results_and_discussion_points.append(main_effect_text)

        # 2. Interaction Effect Analysis (conditional)
        if simulation_parameters:
            interaction_text = (
                f"           *   **交互效应分析：** 深入探讨独立变量与背景参数间的**交互效应**。"
                "背景参数可能包含常规的模型参数，也可能包含约束相关的变量（例如 `Required_TBR`）。"
                f"请阐述在不同的背景参数组合下，`{independent_variable}` 对性能指标的敏感性是否存在显著差异（例如，是被放大还是减弱）。"
                "请特别关注当独立变量与约束类背景参数交互时，对系统性能和达成工程目标的影响。"
            )
            results_and_discussion_points.append(interaction_text)

        # 3. Dynamic Behavior Analysis (conditional)
        if "关键动态数据切片" in original_report_content:
            dynamic_text = (
                f"           *   **动态行为分析：** 解读系统在“初始”、“转折点”和“结束”阶段的行为变化。"
                f"分析 **`{independent_variable}`** 的变化如何影响系统的动态过程，如达到平衡的时间、库存的转折点等。"
            )
            results_and_discussion_points.append(dynamic_text)

        results_and_discussion_str = "\n".join(results_and_discussion_points)

        # Dynamically create the title instruction
        if simulation_parameters:
            title_instruction_text = "请在标题中明确指出，本次分析是关于“独立变量”与“背景扫描参数”的【交互敏感性分析】。"
        else:
            title_instruction_text = (
                "请在标题中明确指出，本次分析是关于“独立变量”的【敏感性分析】。"
            )

        instructions_prompt = f"""**指令：**

1.  **专业化语言：** 将初步报告中的模型参数/缩写（例如 `sds.I[1]`, `Startup_Inventory`）替换为术语表中对应的“中文翻译”或“英文术语”。例如，应将“`sds`的库存”表述为“储存与输送系统 (SDS) 的氚库存量 (Tritium Inventory)”。
2.  **学术化重述：** 用严谨、客观的学术语言重新组织和阐述初步报告中的发现。避免使用“看起来”、“好像”等模糊词汇。
3.  **图表和表格的呈现与引用：**
    *   **显示图表：** 在报告的“结果与讨论”部分，您**必须**使用Markdown语法 `![图表标题](图表文件名)` 来**直接嵌入**和显示初步报告中包含的所有图表。可用的图表文件如下：
{plot_list_str}
    *   **引用图表：** 在正文中分析和讨论图表内容时，请使用“如图1所示...”等方式对图表进行编号和文字引用。
    *   **显示表格：** 当呈现数据时（例如，性能指标总表或关键动态数据切片），您**必须**使用Markdown的管道表格（pipe-table）格式来清晰地展示它们。您可以直接复用或重新格式化初步报告中的数据表格。
4.  **结构化报告：** 您的报告是关于一项**敏感性分析**。报告应包含以下部分：
    *   **标题 (Title):** {title_instruction_text}
    *   **摘要 (Abstract):** 简要概括本次敏感性研究的目的，明确指明独立变量是 **`{independent_variable}`** 以及背景扫描参数，总结其对哪些关键性能指标（如启动库存、增殖时间等）影响最显著，并陈述核心结论。
    *   **引言 (Introduction):** 描述进行这项关于 **`{independent_variable}`** 的敏感性分析的背景和重要性。阐述研究目标，即量化评估 **`{independent_variable}`** 的变化对氚燃料循环系统性能的影响。
        *   **独立变量采样 (Independent Variable Sampling):** 本次分析中，独立变量 `{independent_variable}` 扫描范围为 `{sampling_range}`。
{simulation_params_str}{dependent_vars_str}
    *   **方法 (Methodology):** 简要说明分析方法，包括提及 **`{independent_variable}`** 的扫描范围和被评估的关键性能指标。
    *   **结果与讨论 (Results and Discussion):** 这是报告的核心。请结合所有图表和数据表格，并根据分析内容，组织分点详细论述：
{results_and_discussion_str}
    *   **结论 (Conclusion):** 总结本次敏感性分析得出的主要学术结论，并对反应堆设计或未来运行策略提出具体建议。
5.  **输出格式：** 请直接输出完整的学术分析报告正文，确保所有内容（包括图表和表格）都遵循正确的Markdown语法。

**输入文件：**
"""

        analysis_prompt = f"""
---
### 1. 初步分析报告 (`{report_filename}`)
---
{original_report_content}

---
### 2. 专业术语表 (`sheets.csv`)
---
{glossary_content}
"""

        # 5. Call the API
        max_retries = 3
        for attempt in range(max_retries):
            try:
                client = openai.OpenAI(api_key=api_key, base_url=base_url)
                logger.info(
                    f"Sending request to OpenAI API for academic summary for case {case_name} with model {ai_model} (Attempt {attempt + 1}/{max_retries})..."
                )

                full_text_prompt = "\n\n".join(
                    [role_prompt, instructions_prompt, analysis_prompt]
                )

                response = client.chat.completions.create(
                    model=ai_model,
                    messages=[{"role": "user", "content": full_text_prompt}],
                    max_tokens=4000,
                )
                academic_summary = response.choices[0].message.content

                # 6. Save the result
                sanitized_model_name = "".join(
                    c for c in ai_model if c.isalnum() or c in ("-", "_")
                ).rstrip()
                summary_filename = (
                    f"academic_report_{case_name}_{sanitized_model_name}.md"
                )
                summary_path = os.path.join(results_dir, summary_filename)
                with open(summary_path, "w", encoding="utf-8") as f:
                    f.write(academic_summary)

                logger.info(
                    f"Successfully generated academic analysis summary: {summary_path}"
                )
                return  # Exit after success

            except Exception as e:
                logger.error(
                    f"Error calling OpenAI API for academic summary on attempt {attempt + 1}: {e}"
                )
                if attempt < max_retries - 1:
                    time.sleep(5)
                else:
                    logger.error(
                        f"Failed to generate academic summary for {case_name} after {max_retries} attempts."
                    )
                    return  # Exit after all retries failed

    except Exception as e:
        logger.error(
            f"Error in generate_sensitivity_academic_report for case {case_name}: {e}",
            exc_info=True,
        )

retry_ai_analysis(case_configs, original_config)

Retries AI analysis for cases where it might have failed due to network issues.

Checks for existing reports and re-runs only the AI-dependent parts if they are missing. This function can be triggered by setting an environment variable.

Parameters:

Name	Type	Description	Default
case_configs	List[Dict[str, Any]]	List of case configuration dictionaries.	required
original_config	Dict[str, Any]	Original configuration dictionary.	required

Note

Routes to _retry_salib_case for SALib cases or _retry_standard_case for standard cases. Only regenerates missing AI analysis and academic reports. Does not re-run simulations. Logs all retry attempts and failures.

Source code in tricys/analysis/report.py

def retry_ai_analysis(
    case_configs: List[Dict[str, Any]], original_config: Dict[str, Any]
) -> None:
    """Retries AI analysis for cases where it might have failed due to network issues.

    Checks for existing reports and re-runs only the AI-dependent parts if they are missing.
    This function can be triggered by setting an environment variable.

    Args:
        case_configs: List of case configuration dictionaries.
        original_config: Original configuration dictionary.

    Note:
        Routes to _retry_salib_case for SALib cases or _retry_standard_case for standard cases.
        Only regenerates missing AI analysis and academic reports. Does not re-run simulations.
        Logs all retry attempts and failures.
    """
    logger.info("Starting AI analysis retry process...")
    try:
        for case_info in case_configs:
            case_data = case_info["case_data"]
            if "analyzer" in case_data and case_data.get("analyzer", {}).get("method"):
                _retry_salib_case(case_info, original_config)
            else:
                _retry_standard_case(case_info, original_config)
    except Exception as e:
        logger.error(f"Error during AI analysis retry process: {e}", exc_info=True)

TricysSALibAnalyzer

Integrated SALib's Tricys Sensitivity Analyzer.

Supported Analysis Methods: - Sobol: Variance-based global sensitivity analysis - Morris: Screening-based sensitivity analysis - FAST: Fourier Amplitude Sensitivity Test - LHS: Latin Hypercube Sampling uncertainty analysis

Attributes:

Name	Type	Description
base_config		Copy of the Tricys base configuration.
problem		SALib problem definition dictionary.
parameter_samples		Generated parameter samples array.
simulation_results		Results from simulations.
sensitivity_results		Dictionary storing sensitivity analysis results by method.

Note

Automatically sets up Chinese font support and validates Tricys configuration on initialization. Supports multiple sensitivity analysis methods with appropriate sampling strategies.

Source code in tricys/analysis/salib.py

class TricysSALibAnalyzer:
    """Integrated SALib's Tricys Sensitivity Analyzer.

    Supported Analysis Methods:
    - Sobol: Variance-based global sensitivity analysis
    - Morris: Screening-based sensitivity analysis
    - FAST: Fourier Amplitude Sensitivity Test
    - LHS: Latin Hypercube Sampling uncertainty analysis

    Attributes:
        base_config: Copy of the Tricys base configuration.
        problem: SALib problem definition dictionary.
        parameter_samples: Generated parameter samples array.
        simulation_results: Results from simulations.
        sensitivity_results: Dictionary storing sensitivity analysis results by method.

    Note:
        Automatically sets up Chinese font support and validates Tricys configuration
        on initialization. Supports multiple sensitivity analysis methods with appropriate
        sampling strategies.
    """

    def __init__(self, base_config: Dict[str, Any]) -> None:
        """Initialize the analyzer.

        Args:
            base_config: Tricys base configuration dictionary.

        Note:
            Creates a deep copy of base_config. Initializes problem, samples, and results
            to None. Calls _setup_chinese_font() and _validate_tricys_config() automatically.
        """
        self.base_config = base_config.copy()
        self.problem = None
        self.parameter_samples = None
        self.simulation_results = None
        self.sensitivity_results = {}

        self._setup_chinese_font()
        self._validate_tricys_config()

    def _setup_chinese_font(self) -> None:
        """Set the Chinese font to ensure proper display of Chinese characters in charts.

        Note:
            Tries multiple Chinese fonts in order of preference. Falls back to default
            if no Chinese font found. Also sets axes.unicode_minus to False for proper
            minus sign display. Logs warnings if font setup fails.
        """
        try:
            import matplotlib.font_manager as fm

            chinese_fonts = [
                "SimHei",  # 黑体
                "Microsoft YaHei",  # 微软雅黑
                "KaiTi",  # 楷体
                "FangSong",  # 仿宋
                "STSong",  # 华文宋体
                "STKaiti",  # 华文楷体
                "STHeiti",  # 华文黑体
                "DejaVu Sans",  # 备用字体
                "Arial Unicode MS",  # 备用字体
            ]

            available_font = None
            system_fonts = [f.name for f in fm.fontManager.ttflist]

            for font in chinese_fonts:
                if font in system_fonts:
                    available_font = font
                    break

            if available_font:
                plt.rcParams["font.sans-serif"] = [available_font] + plt.rcParams[
                    "font.sans-serif"
                ]
                logger.info("Using Chinese font", extra={"font": available_font})
            else:
                logger.warning(
                    "No suitable Chinese font found, which may affect Chinese display"
                )

            plt.rcParams["axes.unicode_minus"] = False

        except Exception as e:
            logger.warning(
                "Failed to set Chinese font, using default font",
                extra={"error": str(e)},
            )

    def _handle_nan_values(
        self, Y: np.ndarray, method_name: str = "Sensitivity analysis"
    ) -> np.ndarray:
        """
        Handling NaN values with maximum interpolation

        Args:
            Y: Output array that may contain NaN values
            method_name: Analysis method name for logging

        Returns:
            Processed output array
        """
        nan_indices = np.isnan(Y)
        if np.any(nan_indices):
            n_nan = np.sum(nan_indices)
            logger.info(
                "Found NaN values, using maximum value for imputation",
                extra={
                    "method_name": method_name,
                    "nan_count": n_nan,
                },
            )

            valid_values = Y[~nan_indices]

            if len(valid_values) > 0:
                max_value = np.max(valid_values)
                Y_processed = Y.copy()
                Y_processed[nan_indices] = max_value
                return Y_processed
            else:
                logger.error(
                    "All values are NaN, analysis cannot be performed",
                    extra={
                        "method_name": method_name,
                    },
                )
                raise ValueError(
                    f"{method_name}: All simulation results are NaN, sensitivity analysis cannot be performed"
                )
        return Y

    def _validate_tricys_config(self) -> None:
        """Validate the Tricys configuration for required sections and keys."""
        required_keys = {
            "paths": ["package_path"],
            "simulation": ["model_name", "stop_time"],
        }

        for section, keys in required_keys.items():
            if section not in self.base_config:
                logger.warning(
                    "Missing configuration section, default values will be used",
                    extra={
                        "section": section,
                    },
                )
                continue

            for key in keys:
                if key not in self.base_config[section]:
                    logger.warning(
                        "Missing configuration item, using default value",
                        extra={
                            "section": section,
                            "key": key,
                        },
                    )

        package_path = self.base_config.get("paths", {}).get("package_path")
        if package_path and not os.path.exists(package_path):
            logger.warning(
                "Model file does not exist, which may cause simulation failure",
                extra={
                    "package_path": package_path,
                },
            )

    def _find_unit_config(self, var_name: str, unit_map: dict) -> dict | None:
        """
        Finds the unit configuration for a variable name from the unit_map.
        1. Checks for an exact match.
        2. Checks if the last part of a dot-separated name matches.
        3. Checks for a simple substring containment as a fallback, matching longest keys first.
        """
        if not unit_map or not var_name:
            return None
        if var_name in unit_map:
            return unit_map[var_name]
        components = var_name.split(".")
        if len(components) > 1 and components[-1] in unit_map:
            return unit_map[components[-1]]
        # Fallback to substring match, longest key first
        for key in sorted(unit_map.keys(), key=len, reverse=True):
            if key in var_name:
                return unit_map[key]
        return None

    def define_problem(
        self,
        param_bounds: Dict[str, Tuple[float, float]],
        param_distributions: Dict[str, str] = None,
    ) -> Dict[str, Any]:
        """Define SALib problem space.

        Args:
            param_bounds: Parameter bounds dictionary {'param_name': (min_val, max_val)}.
            param_distributions: Parameter distribution type dictionary {'param_name': 'unif'/'norm'/etc}.
                Valid distribution types: 'unif', 'triang', 'norm', 'truncnorm', 'lognorm'.

        Returns:
            SALib problem definition dictionary.

        Note:
            Defaults to 'unif' distribution if not specified. Validates distribution types
            and warns if invalid. Logs parameter definitions including bounds and distributions.
        """
        if param_distributions is None:
            param_distributions = {name: "unif" for name in param_bounds.keys()}

        valid_dists = ["unif", "triang", "norm", "truncnorm", "lognorm"]
        for name, dist in param_distributions.items():
            if dist not in valid_dists:
                logger.warning(
                    "Invalid distribution type, using 'unif' instead",
                    extra={
                        "parameter_name": name,
                        "invalid_distribution": dist,
                    },
                )
                param_distributions[name] = "unif"

        self.problem = {
            "num_vars": len(param_bounds),
            "names": list(param_bounds.keys()),
            "bounds": list(param_bounds.values()),
            "dists": [
                param_distributions.get(name, "unif") for name in param_bounds.keys()
            ],
        }

        logger.info(
            "Defined a problem space",
            extra={
                "num_parameters": self.problem["num_vars"],
            },
        )
        for i, name in enumerate(self.problem["names"]):
            logger.info(
                "Parameter definition",
                extra={
                    "parameter_name": name,
                    "bounds": self.problem["bounds"][i],
                    "distribution": self.problem["dists"][i],
                },
            )

        return self.problem

    def generate_samples(
        self, method: str = "sobol", N: int = 1024, **kwargs
    ) -> np.ndarray:
        """Generate parameter samples.

        Args:
            method: Sampling method ('sobol', 'morris', 'fast', 'latin').
            N: Number of samples (for Sobol this is the base sample count, actual count is N*(2*D+2)).
            **kwargs: Method-specific parameters.

        Returns:
            Parameter sample array (n_samples, n_params).

        Raises:
            ValueError: If problem not defined or unsupported method.

        Note:
            Sobol generates N*(2*D+2) samples. Morris generates N trajectories. Samples are
            rounded to 5 decimal places. Stores last sampling method for compatibility checking.
        """
        if self.problem is None:
            raise ValueError(
                "You must first call define_problem() to define the problem space."
            )

        logger.info(
            "Generating samples",
            extra={
                "method": method,
                "base_sample_count": N,
            },
        )

        if method.lower() == "sobol":
            # Sobol method: generate N*(2*D+2) samples
            self.parameter_samples = saltelli.sample(self.problem, N, **kwargs)
            actual_samples = N * (2 * self.problem["num_vars"] + 2)

        elif method.lower() == "morris":
            # Morris method: Generate N trajectories
            # Note: Different versions of SALib may have different parameter names
            morris_kwargs = {"num_levels": 4}
            # Check the SALib version and use the correct parameter names
            try:
                morris_kwargs.update(kwargs)
                self.parameter_samples = morris.sample(self.problem, N, **morris_kwargs)
            except TypeError as e:
                if "grid_jump" in str(e):
                    morris_kwargs = {
                        k: v for k, v in morris_kwargs.items() if k != "grid_jump"
                    }
                    morris_kwargs.update(
                        {k: v for k, v in kwargs.items() if k != "grid_jump"}
                    )
                    self.parameter_samples = morris.sample(
                        self.problem, N, **morris_kwargs
                    )
                else:
                    raise e

            actual_samples = len(self.parameter_samples)

        elif method.lower() == "fast":
            # FAST method
            fast_kwargs = {"M": 4}
            fast_kwargs.update(kwargs)
            self.parameter_samples = fast_sampler.sample(self.problem, N, **fast_kwargs)
            actual_samples = len(self.parameter_samples)

        elif method.lower() == "latin":
            # Latin Hypercube Sampling
            self.parameter_samples = latin.sample(self.problem, N, **kwargs)
            actual_samples = N

        else:
            raise ValueError(f"Unsupported sampling method: {method}")

        logger.info(
            "Successfully generated samples", extra={"actual_samples": actual_samples}
        )

        if self.parameter_samples is not None:
            self.parameter_samples = np.round(self.parameter_samples, decimals=5)
            logger.info("Parameter sample precision adjusted to 5 decimal places")

        self._last_sampling_method = method.lower()

        return self.parameter_samples

    def run_tricys_simulations(self, output_metrics: List[str] = None) -> str:
        """
        Generate sampling parameters and output them as a CSV file, which can be subsequently read by the Tricys simulation engine.

        Args:
            output_metrics: List of output metrics to be extracted (for recording but does not affect CSV generation)
            max_workers: Number of concurrent worker processes (reserved for compatibility, currently unused)

        Returns:
            Path to the generated CSV file
        """
        if self.parameter_samples is None:
            raise ValueError(
                "You must first call generate_samples() to generate samples."
            )

        if output_metrics is None:
            output_metrics = [
                "Startup_Inventory",
                "Self_Sufficiency_Time",
                "Doubling_Time",
            ]

        logger.info("Target output metrics", extra={"output_metrics": output_metrics})

        sampled_param_names = self.problem["names"]

        base_params = self.base_config.get("simulation_parameters", {}).copy()
        csv_output_path = (
            Path(self.base_config.get("paths", {}).get("temp_dir"))
            / "salib_sampling.csv"
        )

        os.makedirs(os.path.dirname(csv_output_path), exist_ok=True)

        param_data = []
        for i, sample in enumerate(self.parameter_samples):
            sampled_params = {
                sampled_param_names[j]: sample[j]
                for j in range(len(sampled_param_names))
            }

            job_params = base_params.copy()
            job_params.update(sampled_params)

            param_data.append(job_params)

        df = pd.DataFrame(param_data)

        for col in df.columns:
            if df[col].dtype in ["float64", "float32"]:
                df[col] = df[col].round(5)

        df.to_csv(csv_output_path, index=False, encoding="utf-8")

        logger.info(
            "Successfully generated parameter samples",
            extra={"num_samples": len(param_data)},
        )
        logger.info("Parameter file saved", extra={"file_path": csv_output_path})
        logger.info("Parameter file columns", extra={"columns": list(df.columns)})
        logger.info("Parameter precision set to 5 decimal places")
        logger.info("Sample statistics", extra={"statistics": df.describe().to_dict()})

        self.sampling_csv_path = csv_output_path

        return csv_output_path

    def generate_tricys_config(
        self, csv_file_path: str = None, output_metrics: List[str] = None
    ) -> Dict[str, Any]:
        """
        Generate Tricys configuration file for reading CSV parameter files and executing simulations
        This function reuses the base configuration and specifically modifies simulation_parameters and analysis_case for file-based SALib runs

        Args:
            csv_file_path: Path to the CSV parameter file. If None, the last generated file is used
            output_metrics: List of output metrics to be calculated

        Returns:
            Path of the generated configuration file
        """
        if csv_file_path is None:
            if hasattr(self, "sampling_csv_path"):
                csv_file_path = self.sampling_csv_path
            else:
                raise ValueError(
                    "CSV file path not found, please first call run_tricys_simulations() or specify csv_file_path"
                )

        if output_metrics is None:
            output_metrics = [
                "Startup_Inventory",
                "Self_Sufficiency_Time",
                "Doubling_Time",
            ]

        csv_abs_path = os.path.abspath(csv_file_path)

        import copy

        tricys_config = copy.deepcopy(self.base_config)
        tricys_config["simulation_parameters"] = {"file": csv_abs_path}

        if "sensitivity_analysis" not in tricys_config:
            tricys_config["sensitivity_analysis"] = {"enabled": True}

        tricys_config["sensitivity_analysis"]["analysis_case"] = {
            "name": "SALib_Analysis",
            "independent_variable": "file",
            "independent_variable_sampling": csv_abs_path,
            "dependent_variables": output_metrics,
        }

        return tricys_config

    def load_tricys_results(
        self, sensitivity_summary_csv: str, output_metrics: List[str] = None
    ) -> np.ndarray:
        """
        Read simulation results from the sensitivity_analysis_summary.csv file output by Tricys

        Args:
            sensitivity_summary_csv: Path to the sensitivity analysis summary CSV file output by Tricys
            output_metrics: List of output metrics to extract

        Returns:
            Simulation result array (n_samples, n_metrics)
        """
        if output_metrics is None:
            output_metrics = [
                "Startup_Inventory",
                "Self_Sufficiency_Time",
                "Doubling_Time",
            ]

        logger.info(f"Read data from the Tricys result file: {sensitivity_summary_csv}")

        df = pd.read_csv(sensitivity_summary_csv)

        logger.info(f"Read {len(df)} simulation results")
        logger.info(f"Result file columns: {list(df.columns)}")

        param_cols = []
        metric_cols = []

        for col in df.columns:
            if col in output_metrics:
                metric_cols.append(col)
            elif col in self.problem["names"] if self.problem else False:
                param_cols.append(col)

        logger.info(f"Recognized parameter columns: {param_cols}")
        logger.info(f"Identified metric columns: {metric_cols}")

        ordered_metric_cols = []
        for metric in output_metrics:
            if metric in metric_cols:
                ordered_metric_cols.append(metric)
            else:
                logger.warning(f"Metric column not found: {metric}")

        if not ordered_metric_cols:
            raise ValueError(f"No valid output metrics columns found: {output_metrics}")

        results_data = df[ordered_metric_cols].values

        self.simulation_results = results_data

        logger.info(f"Successfully loaded simulation results: {results_data.shape}")
        logger.info(
            f"Result Statistics:\n{pd.DataFrame(results_data, columns=ordered_metric_cols).describe()}"
        )
        logger.info(
            f"Result preview:\n{pd.DataFrame(results_data, columns=metric_cols).head()}"
        )
        return self.simulation_results

    def get_compatible_analysis_methods(self, sampling_method: str) -> List[str]:
        """
        Get analysis methods compatible with the specified sampling method.

        Args:
            sampling_method: Sampling method

        Returns:
            List of compatible analysis methods
        """
        compatibility_map = {
            "sobol": ["sobol"],
            "morris": ["morris"],
            "fast": ["fast"],
            "latin": ["latin"],
            "unknown": [],
        }

        return compatibility_map.get(sampling_method, [])

    def run_tricys_analysis(
        self, csv_file_path: str = None, output_metrics: List[str] = None
    ) -> str:
        """
        Run the Tricys simulation using the generated CSV parameter file and obtain the sensitivity analysis results

        Args:
            csv_file_path: Path to the CSV parameter file. If None, the last generated file will be used
            output_metrics: List of output metrics to be calculated
            config_output_path: Path for the configuration file output. If None, it will be automatically generated

        Returns:
            Path to the sensitivity_analysis_summary.csv file
        """
        # Generate Tricys configuration file
        tricys_config = self.generate_tricys_config(
            csv_file_path=csv_file_path, output_metrics=output_metrics
        )

        logger.info("Starting Tricys simulation analysis...")

        try:
            # Call the Tricys simulation engine
            from datetime import datetime

            from tricys.simulation.simulation_analysis import run_simulation

            tricys_config["run_timestamp"] = datetime.now().strftime("%Y%m%d_%H%M%S")

            run_simulation(tricys_config)

            results_dir = tricys_config["paths"]["results_dir"]

            return Path(results_dir) / "sensitivity_analysis_summary.csv"

        except Exception as e:
            logger.error(f"Tricys simulation execution failed: {e}")
            raise

    def analyze_sobol(self, output_index: int = 0, **kwargs) -> Dict[str, Any]:
        """
        Perform Sobol Sensitivity Analysis

        Args:
            output_index: Output variable index
            **kwargs: Sobol analysis parameters

        Returns:
            Sobol sensitivity analysis results

        Note:
            Sobol analysis requires samples generated using the Saltelli sampling method!
            Results from Morris or FAST sampling cannot be used.
        """
        if self.simulation_results is None:
            raise ValueError("The simulation must be run first to obtain the results.")

        # Check sampling method compatibility
        if (
            hasattr(self, "_last_sampling_method")
            and self._last_sampling_method != "sobol"
        ):
            logger.warning(
                f"⚠️ Currently using {self._last_sampling_method} sampling, but Sobol analysis requires Saltelli sampling!"
            )
            logger.warning(
                "Suggestion: Regenerate samples using generate_samples('sobol')"
            )

        Y = self.simulation_results[:, output_index]

        Y = self._handle_nan_values(Y, "Sobol分析")

        # Remove NaN values
        # valid_indices = ~np.isnan(Y)
        # if not np.all(valid_indices):
        #    logger.warning(f"发现{np.sum(~valid_indices)}个无效结果，将被排除")
        #    Y = Y[valid_indices]
        #    X = self.parameter_samples[valid_indices]
        # else:
        #    X = self.parameter_samples

        try:
            Si = sobol.analyze(self.problem, Y, **kwargs)

            if "sobol" not in self.sensitivity_results:
                self.sensitivity_results["sobol"] = {}

            metric_name = f"metric_{output_index}"
            self.sensitivity_results["sobol"][metric_name] = {
                "output_index": output_index,
                "Si": Si,
                "S1": Si["S1"],
                "ST": Si["ST"],
                "S2": Si.get("S2", None),
                "S1_conf": Si["S1_conf"],
                "ST_conf": Si["ST_conf"],
                "sampling_method": getattr(self, "_last_sampling_method", "unknown"),
            }

            logger.info(f"Sobol sensitivity analysis completed (index {output_index})")
            return self.sensitivity_results["sobol"][metric_name]

        except Exception as e:
            if "saltelli" in str(e).lower() or "sample" in str(e).lower():
                raise ValueError(
                    f"Sobol analysis failed, possibly due to incompatible sampling method: {e}\nPlease regenerate samples using generate_samples('sobol')"
                ) from e
            else:
                raise

    def analyze_morris(self, output_index: int = 0, **kwargs) -> Dict[str, Any]:
        """
        Perform Morris sensitivity analysis

        Args:
            output_index: Output variable index
            **kwargs: Morris analysis parameters

        Returns:
            Morris sensitivity analysis results
        """
        if self.simulation_results is None:
            raise ValueError("The simulation must be run first to obtain the results.")

        Y = self.simulation_results[:, output_index]

        Y = self._handle_nan_values(Y, "Morris分析")
        X = self.parameter_samples

        # Remove NaN values
        # valid_indices = ~np.isnan(Y)
        # if not np.all(valid_indices):
        #    logger.warning(f"发现{np.sum(~valid_indices)}个无效结果，将被排除")
        #    Y = Y[valid_indices]
        #    X = self.parameter_samples[valid_indices]
        # else:
        #    X = self.parameter_samples

        # Perform Morris analysis
        logger.info(
            f"Start Morris sensitivity analysis: X.shape={X.shape}, Y.shape={Y.shape}, X.dtype={X.dtype}"
        )

        try:
            Si = morris_analyze.analyze(self.problem, X, Y, **kwargs)
        except Exception as e:
            logger.error(f"Morris analysis execution failed: {e}")
            logger.error(f"problem: {self.problem}")
            logger.error(f"X shape: {X.shape}, type: {X.dtype}")
            logger.error(f"Yshape: {Y.shape}, type: {Y.dtype}")
            if hasattr(X, "dtype") and X.dtype == "object":
                logger.error(
                    "X contains non-numeric data, please check the sampled data"
                )
            raise

        if "morris" not in self.sensitivity_results:
            self.sensitivity_results["morris"] = {}

        metric_name = f"metric_{output_index}"
        self.sensitivity_results["morris"][metric_name] = {
            "output_index": output_index,
            "Si": Si,
            "mu": Si["mu"],
            "mu_star": Si["mu_star"],
            "sigma": Si["sigma"],
            "mu_star_conf": Si["mu_star_conf"],
        }

        logger.info(f"Morris sensitivity analysis completed (metric {output_index})")
        return self.sensitivity_results["morris"][metric_name]

    def analyze_fast(self, output_index: int = 0, **kwargs) -> Dict[str, Any]:
        """
        Perform FAST sensitivity analysis

        Args:
            output_index: Output variable index
            **kwargs: FAST analysis parameters

        Returns:
            FAST sensitivity analysis results

        Note:
            FAST analysis requires samples generated by the fast_sampler sampling method!
            Results from Morris or Sobol sampling cannot be used.
        """
        if self.simulation_results is None:
            raise ValueError("The simulation must be run first to obtain the results.")

        if (
            hasattr(self, "_last_sampling_method")
            and self._last_sampling_method != "fast"
        ):
            logger.warning(
                f"⚠️ The current sampling method is {self._last_sampling_method}, but FAST analysis requires FAST sampling!"
            )
            logger.warning(
                "Suggestion: Regenerate samples using generate_samples('fast')"
            )

        Y = self.simulation_results[:, output_index]

        Y = self._handle_nan_values(Y, "FAST分析")

        # Remove NaN values
        # valid_indices = ~np.isnan(Y)
        # if not np.all(valid_indices):
        #    logger.warning(f"发现{np.sum(~valid_indices)}个无效结果，将被排除")
        #    Y = Y[valid_indices]

        try:
            # Perform FAST analysis
            Si = fast.analyze(self.problem, Y, **kwargs)

            if "fast" not in self.sensitivity_results:
                self.sensitivity_results["fast"] = {}

            metric_name = f"metric_{output_index}"
            self.sensitivity_results["fast"][metric_name] = {
                "output_index": output_index,
                "Si": Si,
                "S1": Si["S1"],
                "ST": Si["ST"],
                "sampling_method": getattr(self, "_last_sampling_method", "unknown"),
            }

            logger.info(
                f"FAST sensitivity analysis completed (indicator {output_index})"
            )
            return self.sensitivity_results["fast"][metric_name]

        except Exception as e:
            if "fast" in str(e).lower() or "sample" in str(e).lower():
                raise ValueError(
                    f"FAST analysis failed, possibly due to incompatible sampling method: {e}\nPlease regenerate samples using generate_samples('fast')"
                ) from e
            else:
                raise

    def analyze_lhs(self, output_index: int = 0, **kwargs) -> Dict[str, Any]:
        """
        Perform LHS (Latin Hypercube Sampling) uncertainty analysis

        Note: This is a basic statistical analysis method for LHS samples,
        providing descriptive statistics and basic sensitivity indices.

        Args:
            output_index: Output variable index
            **kwargs: Analysis parameters (reserved for future use)

        Returns:
            LHS uncertainty analysis results
        """
        if self.simulation_results is None:
            raise ValueError("The simulation must be run first to obtain the results.")

        if (
            hasattr(self, "_last_sampling_method")
            and self._last_sampling_method != "latin"
        ):
            logger.warning(
                f"⚠️ The current sampling method is {self._last_sampling_method}, but LHS analysis is designed for Latin Hypercube Sampling!"
            )
            logger.warning(
                "Suggestion: Regenerate samples using generate_samples('latin')"
            )

        Y = self.simulation_results[:, output_index]

        # Handle NaN values
        Y = self._handle_nan_values(Y, "LHS分析")

        # Basic statistical analysis
        mean_val = np.mean(Y)
        std_val = np.std(Y)
        min_val = np.min(Y)
        max_val = np.max(Y)
        percentile_5 = np.percentile(Y, 5)
        percentile_95 = np.percentile(Y, 95)

        # Create results dictionary
        Si = {
            "mean": mean_val,
            "std": std_val,
            "min": min_val,
            "max": max_val,
            "percentile_5": percentile_5,
            "percentile_95": percentile_95,
        }

        if "latin" not in self.sensitivity_results:
            self.sensitivity_results["latin"] = {}

        metric_name = f"metric_{output_index}"
        self.sensitivity_results["latin"][metric_name] = {
            "output_index": output_index,
            "Si": Si,
            "mean": mean_val,
            "std": std_val,
            "min": min_val,
            "max": max_val,
            "percentile_5": percentile_5,
            "percentile_95": percentile_95,
            "sampling_method": getattr(self, "_last_sampling_method", "unknown"),
        }

        logger.info(f"LHS uncertainty analysis completed (指标 {output_index})")
        return self.sensitivity_results["latin"][metric_name]

    def run_salib_analysis_from_tricys_results(
        self,
        sensitivity_summary_csv: str,
        param_bounds: Dict[str, Tuple[float, float]] = None,
        output_metrics: List[str] = None,
        methods: List[str] = ["sobol", "morris", "fast"],
        save_dir: str = None,
    ) -> Dict[str, Any]:
        """
        Run a complete SALib sensitivity analysis from the sensitivity analysis results file output by Tricys

        Args:
            sensitivity_summary_csv: Path to the sensitivity summary CSV file output by Tricys
            param_bounds: Dictionary of parameter bounds, inferred from the CSV file if None
            output_metrics: List of output metrics to analyze
            methods: List of sensitivity analysis methods to execute
            save_dir: Directory to save the results

        Returns:
            Dictionary containing all analysis results
        """
        if output_metrics is None:
            output_metrics = [
                "Startup_Inventory",
                "Self_Sufficiency_Time",
                "Doubling_Time",
            ]

        if save_dir is None:
            save_dir = os.path.join(
                os.path.dirname(sensitivity_summary_csv), "salib_analysis"
            )
        os.makedirs(save_dir, exist_ok=True)

        df = pd.read_csv(sensitivity_summary_csv)

        if param_bounds is None:
            param_bounds = {}
            param_candidates = []
            for col in df.columns:
                if col not in output_metrics and "." in col:
                    param_candidates.append(col)

            for param in param_candidates:
                param_data = df[param].dropna()
                if len(param_data) > 0:
                    param_bounds[param] = (param_data.min(), param_data.max())

        if not param_bounds:
            raise ValueError(
                "Unable to determine parameter boundaries, please provide the param_bounds parameter"
            )

        self.define_problem(param_bounds)

        self.load_tricys_results(sensitivity_summary_csv, output_metrics)

        detected_method = self._last_sampling_method

        methods = self.get_compatible_analysis_methods(detected_method)

        all_results = {}

        for metric_idx, metric_name in enumerate(output_metrics):
            if metric_idx >= self.simulation_results.shape[1]:
                logger.warning(f"The metric {metric_name} is out of range, skipping")
                continue

            logger.info(f"\n=== Analysis indicators: {metric_name} ===")
            metric_results = {}

            # Check data validity
            Y = self.simulation_results[:, metric_idx]
            valid_ratio = np.sum(~np.isnan(Y)) / len(Y)
            logger.info(f"Valid data ratio: {valid_ratio:.2%}")

            if valid_ratio < 0.5:
                logger.warning(
                    f"The metric {metric_name} has less than 50% valid data, which may affect the analysis quality."
                )

            # Sobol analysis
            if "sobol" in methods:
                try:
                    logger.info("Performing Sobol sensitivity analysis...")
                    sobol_result = self.analyze_sobol(output_index=metric_idx)
                    metric_results["sobol"] = sobol_result

                    # Display Sobol results summary
                    logger.info("\nSobol sensitivity index:")
                    for i, param_name in enumerate(self.problem["names"]):
                        s1 = sobol_result["S1"][i]
                        st = sobol_result["ST"][i]
                        logger.info(f"  {param_name}: S1={s1:.4f}, ST={st:.4f}")

                except Exception as e:
                    logger.error(f"Sobol analysis failed: {e}")

            # Morris analysis
            if "morris" in methods:
                try:
                    logger.info("Performing Morris sensitivity analysis...")
                    morris_result = self.analyze_morris(output_index=metric_idx)
                    metric_results["morris"] = morris_result

                    # Display Morris results summary
                    logger.info("\nMorris sensitivity index:")
                    for i, param_name in enumerate(self.problem["names"]):
                        mu_star = morris_result["mu_star"][i]
                        sigma = morris_result["sigma"][i]
                        logger.info(f"  {param_name}: μ*={mu_star:.4f}, σ={sigma:.4f}")

                except Exception as e:
                    logger.error(f"Morris analysis failed: {e}")

            # FAST analysis
            if "fast" in methods:
                try:
                    logger.info("Performing FAST sensitivity analysis...")
                    fast_result = self.analyze_fast(output_index=metric_idx)
                    metric_results["fast"] = fast_result

                    # Display FAST results summary
                    logger.info("\nFAST sensitivity index:")
                    for i, param_name in enumerate(self.problem["names"]):
                        s1 = fast_result["S1"][i]
                        st = fast_result["ST"][i]
                        logger.info(f"  {param_name}: S1={s1:.4f}, ST={st:.4f}")

                except Exception as e:
                    logger.error(f"FAST analysis failed: {e}")

            # LHS analysis
            if "latin" in methods:
                try:
                    logger.info("Performing LHS uncertainty analysis...")
                    lhs_result = self.analyze_lhs(output_index=metric_idx)
                    metric_results["latin"] = lhs_result

                    # Display LHS results summary
                    logger.info("\nLHS分析结果:")
                    logger.info(f"  均值: {lhs_result['mean']:.4f}")
                    logger.info(f"  标准差: {lhs_result['std']:.4f}")
                    logger.info(f"  最小值: {lhs_result['min']:.4f}")
                    logger.info(f"  最大值: {lhs_result['max']:.4f}")
                    logger.info(f"  5%分位数: {lhs_result['percentile_5']:.4f}")
                    logger.info(f"  95%分位数: {lhs_result['percentile_95']:.4f}")

                except Exception as e:
                    logger.error(f"LHS分析失败: {e}")

            all_results[metric_name] = metric_results

        try:
            if "sobol" in methods and "sobol" in self.sensitivity_results:
                self.plot_sobol_results(save_dir=save_dir, metric_names=output_metrics)

            if "morris" in methods and "morris" in self.sensitivity_results:
                self.plot_morris_results(save_dir=save_dir, metric_names=output_metrics)

            if "fast" in methods and "fast" in self.sensitivity_results:
                self.plot_fast_results(save_dir=save_dir, metric_names=output_metrics)

            # Plot LHS results
            if "latin" in methods and "latin" in self.sensitivity_results:
                self.plot_lhs_results(save_dir=save_dir, metric_names=output_metrics)

        except Exception as e:
            logger.warning(f"Drawing failed: {e}")

        try:
            self.save_results(
                save_dir=save_dir, format="csv", metric_names=output_metrics
            )

            report_content = self._save_sensitivity_report(all_results, save_dir)
            report_path = os.path.join(save_dir, "analysis_report.md")

            load_dotenv()

            # --- LLM Calls for analysis ---
            api_key = os.environ.get("API_KEY")
            base_url = os.environ.get("BASE_URL")
            ai_model = os.environ.get("AI_MODEL")

            sa_config = self.base_config.get("sensitivity_analysis", {})
            case_config = sa_config.get("analysis_case", {})
            ai_config = case_config.get("ai")

            ai_enabled = False
            if isinstance(ai_config, bool):
                ai_enabled = ai_config
            elif isinstance(ai_config, dict):
                ai_enabled = ai_config.get("enabled", False)

            if api_key and base_url and ai_model and ai_enabled:
                # First LLM call for initial analysis
                wrapper_prompt, llm_summary = call_llm_for_salib_analysis(
                    report_content=report_content,
                    api_key=api_key,
                    base_url=base_url,
                    ai_model=ai_model,
                    method=detected_method,
                )
                if wrapper_prompt and llm_summary:
                    with open(report_path, "a", encoding="utf-8") as f:
                        f.write("\n\n---\n\n# AI模型分析提示词\n\n")
                        f.write("```markdown\n")
                        f.write(wrapper_prompt)
                        f.write("\n```\n\n")
                        f.write("\n\n---\n\n# AI模型分析结果\n\n")
                        f.write(llm_summary)
                    logger.info(f"Appended LLM prompt and summary to {report_path}")

                    # Second LLM call for academic report
                    glossary_path = None
                    if isinstance(case_config, dict):
                        glossary_path = sa_config.get("glossary_path")

                    if glossary_path and os.path.exists(glossary_path):
                        try:
                            with open(glossary_path, "r", encoding="utf-8") as f:
                                glossary_content = f.read()

                            (
                                academic_wrapper_prompt,
                                academic_report,
                            ) = call_llm_for_academic_report(
                                analysis_report=llm_summary,
                                glossary_content=glossary_content,
                                api_key=api_key,
                                base_url=base_url,
                                ai_model=ai_model,
                                problem_details=self.problem,
                                metric_names=output_metrics,
                                method=detected_method,
                                save_dir=save_dir,
                            )

                            if academic_wrapper_prompt and academic_report:
                                academic_report_path = os.path.join(
                                    save_dir, "academic_report.md"
                                )
                                with open(
                                    academic_report_path, "w", encoding="utf-8"
                                ) as f:
                                    f.write(academic_report)
                                logger.info(
                                    f"Generated academic report: {academic_report_path}"
                                )
                        except Exception as e:
                            logger.error(
                                f"Failed to generate or save academic report: {e}"
                            )
                    elif glossary_path:
                        logger.warning(
                            f"Glossary file not found at {glossary_path}, skipping academic report generation."
                        )

            else:
                logger.warning(
                    "API_KEY, BASE_URL, or AI_MODEL not set, or AI analysis is disabled. Skipping LLM summary generation."
                )

        except Exception as e:
            logger.warning(f"Failed to save result: {e}")

        logger.info("\n✅ SALib sensitivity analysis completed!")
        logger.info(f"📁 The result has been saved to: {save_dir}")

        return all_results

    def _save_sensitivity_report(
        self, all_results: Dict[str, Any], save_dir: str
    ) -> str:
        """The result has been saved to: {save_dir}"""
        report_file = os.path.join(save_dir, "analysis_report.md")
        # Determine analysis type based on sampling method
        is_uncertainty_analysis = (
            hasattr(self, "_last_sampling_method")
            and self._last_sampling_method == "latin"
        )
        report_title = (
            "# SALib 不确定性分析报告\n\n"
            if is_uncertainty_analysis
            else "# SALib 敏感性分析报告\n\n"
        )
        report_lines = []
        report_lines.append(report_title)
        report_lines.append(f"生成时间: {pd.Timestamp.now()}\n\n")
        # Get unit_map from config
        sensitivity_analysis_config = self.base_config.get("sensitivity_analysis", {})
        unit_map = sensitivity_analysis_config.get("unit_map", {})
        report_lines.append("## 分析参数\n\n")
        if self.problem:
            for i, param_name in enumerate(self.problem["names"]):
                bounds = self.problem["bounds"][i]
                # --- Unit Conversion Logic for Bounds ---
                unit_config = self._find_unit_config(param_name, unit_map)
                display_bounds = list(bounds)
                unit_str = ""
                if unit_config:
                    unit = unit_config.get("unit")
                    factor = unit_config.get("conversion_factor")
                    if factor:
                        display_bounds[0] /= float(factor)
                        display_bounds[1] /= float(factor)
                    if unit:
                        unit_str = f" ({unit})"
                # --- End Conversion Logic ---
                report_lines.append(
                    f"- **{param_name}**: [{display_bounds[0]:.4f}, {display_bounds[1]:.4f}]{unit_str}\n"
                )
        report_lines.append("\n")
        for metric_name, metric_results in all_results.items():
            metric_section_title = (
                f"## {metric_name} 不确定性分析结果\n\n"
                if is_uncertainty_analysis
                else f"## {metric_name} 敏感性分析结果\n\n"
            )
            report_lines.append(metric_section_title)
            if "sobol" in metric_results:
                report_lines.append("### Sobol敏感性指数\n\n")
                report_lines.append(
                    "| 参数 | S1 (一阶) | ST (总) | S1置信区间 | ST置信区间 |\n"
                )
                report_lines.append(
                    "|------|----------|---------|------------|------------|\n"
                )
                sobol_data = metric_results["sobol"]
                for i, param_name in enumerate(self.problem["names"]):
                    s1 = sobol_data["S1"][i]
                    st = sobol_data["ST"][i]
                    s1_conf = sobol_data["S1_conf"][i]
                    st_conf = sobol_data["ST_conf"][i]
                    report_lines.append(
                        f"| {param_name} | {s1:.4f} | {st:.4f} | ±{s1_conf:.4f} | ±{st_conf:.4f} |\n"
                    )
                report_lines.append("\n")
                plot_filename = (
                    f'sobol_sensitivity_indices_{metric_name.replace(" ", "_")}.png'
                )
                report_lines.append(
                    f"![Sobol Analysis for {metric_name}]({plot_filename})\n\n"
                )
            if "morris" in metric_results:
                report_lines.append("### Morris敏感性指数\n\n")
                report_lines.append(
                    "| 参数 | μ* (平均绝对效应) | σ (标准差) | μ*置信区间 |\n"
                )
                report_lines.append(
                    "|------|-------------------|------------|------------|\n"
                )
                morris_data = metric_results["morris"]
                for i, param_name in enumerate(self.problem["names"]):
                    mu_star = morris_data["mu_star"][i]
                    sigma = morris_data["sigma"][i]
                    mu_star_conf = morris_data["mu_star_conf"][i]
                    report_lines.append(
                        f"| {param_name} | {mu_star:.4f} | {sigma:.4f} | ±{mu_star_conf:.4f} |\n"
                    )
                report_lines.append("\n")
                plot_filename = (
                    f'morris_sensitivity_analysis_{metric_name.replace(" ", "_")}.png'
                )
                report_lines.append(
                    f"![Morris Analysis for {metric_name}]({plot_filename})\n\n"
                )
            if "fast" in metric_results:
                report_lines.append("### FAST敏感性指数\n\n")
                report_lines.append("| 参数 | S1 (一阶) | ST (总) |\n")
                report_lines.append("|------|----------|---------|\n")
                fast_data = metric_results["fast"]
                for i, param_name in enumerate(self.problem["names"]):
                    s1 = fast_data["S1"][i]
                    st = fast_data["ST"][i]
                    report_lines.append(f"| {param_name} | {s1:.4f} | {st:.4f} |\n")
                report_lines.append("\n")
                plot_filename = (
                    f'fast_sensitivity_indices_{metric_name.replace(" ", "_")}.png'
                )
                report_lines.append(
                    f"![FAST Analysis for {metric_name}]({plot_filename})\n\n"
                )
            if "latin" in metric_results:
                # --- Unit Conversion Logic for Metrics ---
                unit_config = self._find_unit_config(metric_name, unit_map)
                unit_str = ""
                factor = 1.0
                if unit_config:
                    unit = unit_config.get("unit")
                    conv_factor = unit_config.get("conversion_factor")
                    if conv_factor:
                        factor = float(conv_factor)
                    if unit:
                        unit_str = f" ({unit})"
                # --- End Conversion Logic ---
                # 1. Get raw data and clean it
                output_index = metric_results["latin"]["output_index"]
                Y = self.simulation_results[:, output_index]
                Y_clean = Y[~np.isnan(Y)]
                # --- Modify report generation ---
                report_lines.append("### 统计摘要\n\n")
                lhs_data = metric_results["latin"]
                report_lines.append(
                    f"- 均值: {lhs_data['mean']/factor:.4f}{unit_str}\n"
                )
                report_lines.append(
                    f"- 标准差: {lhs_data['std']/factor:.4f}{unit_str}\n"
                )
                report_lines.append(
                    f"- 最小值: {lhs_data['min']/factor:.4f}{unit_str}\n"
                )
                report_lines.append(
                    f"- 最大值: {lhs_data['max']/factor:.4f}{unit_str}\n\n"
                )
                # 2. Calculate more percentiles
                if len(Y_clean) > 0:
                    percentiles_to_calc = [5, 10, 25, 50, 75, 90, 95]
                    percentile_values = np.percentile(Y_clean, percentiles_to_calc)
                    report_lines.append("### 分布关键点 (CDF)\n\n")
                    report_lines.append(
                        f"- 5%分位数: {percentile_values[0]/factor:.4f}{unit_str}\n"
                    )
                    report_lines.append(
                        f"- 10%分位数: {percentile_values[1]/factor:.4f}{unit_str}\n"
                    )
                    report_lines.append(
                        f"- 25%分位数 (Q1): {percentile_values[2]/factor:.4f}{unit_str}\n"
                    )
                    report_lines.append(
                        f"- 50%分位数 (中位数): {percentile_values[3]/factor:.4f}{unit_str}\n"
                    )
                    report_lines.append(
                        f"- 75%分位数 (Q3): {percentile_values[4]/factor:.4f}{unit_str}\n"
                    )
                    report_lines.append(
                        f"- 90%分位数: {percentile_values[5]/factor:.4f}{unit_str}\n"
                    )
                    report_lines.append(
                        f"- 95%分位数: {percentile_values[6]/factor:.4f}{unit_str}\n\n"
                    )
                    # 3. Calculate histogram data
                    hist_freq, bin_edges = np.histogram(Y_clean, bins=10)
                    report_lines.append("### 输出分布 (直方图数据)\n\n")
                    report_lines.append("| 数值范围 | 频数 |\n")
                    report_lines.append("|:---|---:|\n")
                    for i in range(len(hist_freq)):
                        lower_bound = bin_edges[i] / factor
                        upper_bound = bin_edges[i + 1] / factor
                        freq = hist_freq[i]
                        report_lines.append(
                            f"| {lower_bound:.2f} - {upper_bound:.2f} | {freq} |\n"
                        )
                    report_lines.append("\n")
                plot_filename = f'lhs_analysis_{metric_name.replace(" ", "_")}.png'
                report_lines.append(
                    f"![LHS Analysis for {metric_name}]({plot_filename})\n\n"
                )
        report_content = "".join(report_lines)
        with open(report_file, "w", encoding="utf-8") as f:
            f.write(report_content)
        logger.info(f"The sensitivity analysis report has been saved.: {report_file}")
        return report_content

    def plot_sobol_results(
        self,
        save_dir: str = None,
        figsize: Tuple[int, int] = (12, 8),
        metric_names: List[str] = None,
    ) -> None:
        """Plot Sobol analysis results"""
        if "sobol" not in self.sensitivity_results:
            raise ValueError("No analysis results for the Sobol method were found.")

        # Ensure Chinese font settings
        self._setup_chinese_font()

        if save_dir is None:
            save_dir = "."
        os.makedirs(save_dir, exist_ok=True)

        # Get the results of all indicators
        sobol_results = self.sensitivity_results["sobol"]

        if not sobol_results:
            raise ValueError("Sobol analysis results not found")

        # Generate charts for each metric
        for metric_key, results in sobol_results.items():
            Si = results["Si"]
            output_index = results["output_index"]

            if metric_names and output_index < len(metric_names):
                metric_display_name = metric_names[output_index]
            else:
                metric_display_name = f"Metric_{output_index}"

            # Bar chart of first-order and total sensitivity indices
            fig, (ax1, ax2) = plt.subplots(1, 2, figsize=figsize)

            # First-order sensitivity index
            y_pos = np.arange(len(self.problem["names"]))
            ax1.barh(y_pos, Si["S1"], xerr=Si["S1_conf"], alpha=0.7, color="skyblue")
            ax1.set_yticks(y_pos)
            ax1.set_yticklabels(self.problem["names"], fontsize=10)
            ax1.set_xlabel("First-order sensitivity index (S1)", fontsize=12)
            ax1.set_title(
                f"First-order Sensitivity Indices\n{metric_display_name}",
                fontsize=14,
                pad=20,
            )
            ax1.grid(True, alpha=0.3)

            # # Total Sensitivity Index
            ax2.barh(y_pos, Si["ST"], xerr=Si["ST_conf"], alpha=0.7, color="orange")
            ax2.set_yticks(y_pos)
            ax2.set_yticklabels(self.problem["names"], fontsize=10)
            ax2.set_xlabel("Total Sensitivity Index (ST)", fontsize=12)
            ax2.set_title(
                f"Total Sensitivity Indices\n{metric_display_name}", fontsize=14, pad=20
            )
            ax2.grid(True, alpha=0.3)

            plt.tight_layout()
            filename = (
                f'sobol_sensitivity_indices_{metric_display_name.replace(" ", "_")}.png'
            )
            plt.savefig(os.path.join(save_dir, filename), dpi=300, bbox_inches="tight")

            logger.info(f"Sobol result chart has been saved: {filename}")

    def plot_morris_results(
        self,
        save_dir: str = None,
        figsize: Tuple[int, int] = (12, 8),
        metric_names: List[str] = None,
    ) -> None:
        """Plot the Morris analysis results"""
        if "morris" not in self.sensitivity_results:
            raise ValueError("No analysis results were found for the Morris method.")

        # Ensure Chinese font settings
        self._setup_chinese_font()

        if save_dir is None:
            save_dir = "."
        os.makedirs(save_dir, exist_ok=True)

        # Obtain the results of all indicators
        morris_results = self.sensitivity_results["morris"]

        if not morris_results:
            raise ValueError("No Morris analysis results found")

        for metric_key, results in morris_results.items():
            Si = results["Si"]
            output_index = results["output_index"]

            if metric_names and output_index < len(metric_names):
                metric_display_name = metric_names[output_index]
            else:
                metric_display_name = f"Metric_{output_index}"

            # Morris μ*-σ diagram
            fig, (ax1, ax2) = plt.subplots(1, 2, figsize=figsize)

            # μ*-σ scatter plot
            ax1.scatter(Si["mu_star"], Si["sigma"], s=100, alpha=0.7, color="red")
            for i, name in enumerate(self.problem["names"]):
                ax1.annotate(
                    name,
                    (Si["mu_star"][i], Si["sigma"][i]),
                    xytext=(5, 5),
                    textcoords="offset points",
                    fontsize=9,
                )

            ax1.set_xlabel("μ*(Average Absolute Effect)", fontsize=12)
            ax1.set_ylabel("σ (Standard Deviation)", fontsize=12)
            ax1.set_title(
                f"Morris μ*-σ Plot\n{metric_display_name}", fontsize=14, pad=20
            )
            ax1.grid(True, alpha=0.3)

            y_pos = np.arange(len(self.problem["names"]))
            ax2.barh(
                y_pos, Si["mu_star"], xerr=Si["mu_star_conf"], alpha=0.7, color="green"
            )
            ax2.set_yticks(y_pos)
            ax2.set_yticklabels(self.problem["names"], fontsize=10)
            ax2.set_xlabel("μ*(Average Absolute Effect)", fontsize=12)
            ax2.set_title(
                f"Morris Elementary Effects\n{metric_display_name}", fontsize=14, pad=20
            )
            ax2.grid(True, alpha=0.3)

            plt.tight_layout()
            filename = f'morris_sensitivity_analysis_{metric_display_name.replace(" ", "_")}.png'
            plt.savefig(os.path.join(save_dir, filename), dpi=300, bbox_inches="tight")

            logger.info(f"Morris result chart has been saved: {filename}")

    def plot_fast_results(
        self,
        save_dir: str = None,
        figsize: Tuple[int, int] = (12, 8),
        metric_names: List[str] = None,
    ) -> None:
        """Plot FAST analysis results"""
        if "fast" not in self.sensitivity_results:
            raise ValueError("No analysis results found for the FAST method")

        # No analysis results found for the FAST method
        self._setup_chinese_font()

        if save_dir is None:
            save_dir = "."
        os.makedirs(save_dir, exist_ok=True)

        # Get the results of all indicators
        fast_results = self.sensitivity_results["fast"]

        if not fast_results:
            raise ValueError("FAST analysis results not found")

        # Generate a chart for each metric
        for metric_key, results in fast_results.items():
            Si = results["Si"]
            output_index = results["output_index"]

            # Determine the indicator name
            if metric_names and output_index < len(metric_names):
                metric_display_name = metric_names[output_index]
            else:
                metric_display_name = f"Metric_{output_index}"

            # Bar charts of first-order and total sensitivity indices
            fig, (ax1, ax2) = plt.subplots(1, 2, figsize=figsize)

            # first-order sensitivity index
            y_pos = np.arange(len(self.problem["names"]))
            ax1.barh(y_pos, Si["S1"], alpha=0.7, color="purple")
            ax1.set_yticks(y_pos)
            ax1.set_yticklabels(self.problem["names"], fontsize=10)
            ax1.set_xlabel("一阶敏感性指数 (S1)", fontsize=12)
            ax1.set_title(
                f"FAST First-order Sensitivity Indices\n{metric_display_name}",
                fontsize=14,
                pad=20,
            )
            ax1.grid(True, alpha=0.3)

            # Total Sensitivity Index
            ax2.barh(y_pos, Si["ST"], alpha=0.7, color="darkgreen")
            ax2.set_yticks(y_pos)
            ax2.set_yticklabels(self.problem["names"], fontsize=10)
            ax2.set_xlabel("总敏感性指数 (ST)", fontsize=12)
            ax2.set_title(
                f"FAST Total Sensitivity Indices\n{metric_display_name}",
                fontsize=14,
                pad=20,
            )
            ax2.grid(True, alpha=0.3)

            plt.tight_layout()
            filename = (
                f'fast_sensitivity_indices_{metric_display_name.replace(" ", "_")}.png'
            )
            plt.savefig(os.path.join(save_dir, filename), dpi=300, bbox_inches="tight")

            logger.info(f"The FAST result chart has been saved: {filename}")

    def plot_lhs_results(
        self,
        save_dir: str = None,
        figsize: Tuple[int, int] = (12, 8),
        metric_names: List[str] = None,
    ) -> None:
        """Plot LHS (Latin Hypercube Sampling) uncertainty analysis results"""
        if "latin" not in self.sensitivity_results:
            raise ValueError("No analysis results found for the LHS method")

        # Ensure Chinese font settings
        self._setup_chinese_font()

        if save_dir is None:
            save_dir = "."
        os.makedirs(save_dir, exist_ok=True)

        # Get the results of all indicators
        lhs_results = self.sensitivity_results["latin"]

        if not lhs_results:
            raise ValueError("LHS analysis results not found")

        # Generate charts for each metric
        for metric_key, results in lhs_results.items():
            Si = results["Si"]
            output_index = results["output_index"]

            # Determine the indicator name
            if metric_names and output_index < len(metric_names):
                metric_display_name = metric_names[output_index]
            else:
                metric_display_name = f"Metric_{output_index}"

            # Get unit from config
            sensitivity_analysis_config = self.base_config.get(
                "sensitivity_analysis", {}
            )
            unit_map = sensitivity_analysis_config.get("unit_map", {})
            unit_config = self._find_unit_config(metric_display_name, unit_map)
            unit_str = ""
            if unit_config:
                unit = unit_config.get("unit")
                if unit:
                    unit_str = f" ({unit})"

            xlabel = f"{metric_display_name}{unit_str}"

            # Create a figure with two subplots
            fig, (ax1, ax2) = plt.subplots(1, 2, figsize=figsize)

            # Plot 1: Distribution histogram
            ax1.hist(
                self.simulation_results[:, output_index],
                bins=30,
                alpha=0.7,
                color="skyblue",
                edgecolor="black",
            )
            ax1.set_xlabel(xlabel, fontsize=12)
            ax1.set_ylabel("频率", fontsize=12)
            ax1.set_title("输出分布直方图", fontsize=14, pad=10)
            ax1.grid(True, alpha=0.3)

            # Add statistics text to the histogram plot
            stats_text = f"均值: {Si['mean']:.4f}\n标准差: {Si['std']:.4f}\n最小值: {Si['min']:.4f}\n最大值: {Si['max']:.4f}"
            ax1.text(
                0.05,
                0.95,
                stats_text,
                transform=ax1.transAxes,
                fontsize=10,
                verticalalignment="top",
                bbox=dict(boxstyle="round", facecolor="white", alpha=0.8),
            )

            # Plot 2: Cumulative distribution function
            sorted_data = np.sort(self.simulation_results[:, output_index])
            y_vals = np.arange(1, len(sorted_data) + 1) / len(sorted_data)
            ax2.plot(sorted_data, y_vals, linewidth=2, color="darkgreen")
            ax2.set_xlabel(xlabel, fontsize=12)
            ax2.set_ylabel("累积概率", fontsize=12)
            ax2.set_title("累积分布函数", fontsize=14, pad=10)
            ax2.grid(True, alpha=0.3)

            plt.tight_layout()
            filename = f'lhs_analysis_{metric_display_name.replace(" ", "_")}.png'
            plt.savefig(os.path.join(save_dir, filename), dpi=300, bbox_inches="tight")

            logger.info(f"LHS分析结果图表已保存: {filename}")

    def save_results(
        self, save_dir: str = None, format: str = "csv", metric_names: List[str] = None
    ) -> None:
        """
        Save sensitivity analysis results

        Args:
            save_dir: Save directory
            format: Save format ('csv
        """
        if save_dir is None:
            save_dir = "."
        os.makedirs(save_dir, exist_ok=True)

        for method, method_results in self.sensitivity_results.items():
            if not method_results:
                continue

            for metric_key, results in method_results.items():
                output_index = results["output_index"]

                if metric_names and output_index < len(metric_names):
                    metric_display_name = metric_names[output_index]
                else:
                    metric_display_name = f"Metric_{output_index}"

                if format == "csv":
                    if method == "sobol":
                        sobol_df = pd.DataFrame(
                            {
                                "Parameter": self.problem["names"],
                                "S1": results["S1"],
                                "ST": results["ST"],
                                "S1_conf": results["S1_conf"],
                                "ST_conf": results["ST_conf"],
                            }
                        )
                        filename = (
                            f'sobol_indices_{metric_display_name.replace(" ", "_")}.csv'
                        )
                        sobol_df.to_csv(os.path.join(save_dir, filename), index=False)
                        logger.info(f"Sobol results have been saved: {filename}")

                    elif method == "morris":
                        morris_df = pd.DataFrame(
                            {
                                "Parameter": self.problem["names"],
                                "mu": results["mu"],
                                "mu_star": results["mu_star"],
                                "sigma": results["sigma"],
                                "mu_star_conf": results["mu_star_conf"],
                            }
                        )
                        filename = f'morris_indices_{metric_display_name.replace(" ", "_")}.csv'
                        morris_df.to_csv(os.path.join(save_dir, filename), index=False)
                        logger.info(f"Morris results have been saved: {filename}")

                    elif method == "fast":
                        fast_df = pd.DataFrame(
                            {
                                "Parameter": self.problem["names"],
                                "S1": results["S1"],
                                "ST": results["ST"],
                            }
                        )
                        filename = (
                            f'fast_indices_{metric_display_name.replace(" ", "_")}.csv'
                        )
                        fast_df.to_csv(os.path.join(save_dir, filename), index=False)
                        logger.info(f"FAST results have been saved: {filename}")

                    elif method == "latin":
                        # Save LHS statistics
                        lhs_stats_df = pd.DataFrame(
                            {
                                "Metric": [metric_display_name],
                                "Mean": [results["mean"]],
                                "Std": [results["std"]],
                                "Min": [results["min"]],
                                "Max": [results["max"]],
                                "Percentile_5": [results["percentile_5"]],
                                "Percentile_95": [results["percentile_95"]],
                            }
                        )
                        filename_stats = (
                            f'lhs_stats_{metric_display_name.replace(" ", "_")}.csv'
                        )
                        lhs_stats_df.to_csv(
                            os.path.join(save_dir, filename_stats), index=False
                        )
                        logger.info(f"LHS统计结果已保存: {filename_stats}")

                        # Remove LHS sensitivity indices saving
                        # lhs_sens_df = pd.DataFrame({
                        #     "Parameter": self.problem["names"],
                        #     "Partial_Correlation": results["partial_correlations"]
                        # })
                        # filename_sens = f'lhs_sensitivity_{metric_display_name.replace(" ", "_")}.csv'
                        # lhs_sens_df.to_csv(os.path.join(save_dir, filename_sens), index=False)
                        # logger.info(f"LHS敏感性结果已保存: {filename_sens}")

        logger.info(f"The result has been saved to: {save_dir}")

__init__(base_config)

Initialize the analyzer.

Parameters:

Name	Type	Description	Default
base_config	Dict[str, Any]	Tricys base configuration dictionary.	required

Note

Creates a deep copy of base_config. Initializes problem, samples, and results to None. Calls _setup_chinese_font() and _validate_tricys_config() automatically.

Source code in tricys/analysis/salib.py

def __init__(self, base_config: Dict[str, Any]) -> None:
    """Initialize the analyzer.

    Args:
        base_config: Tricys base configuration dictionary.

    Note:
        Creates a deep copy of base_config. Initializes problem, samples, and results
        to None. Calls _setup_chinese_font() and _validate_tricys_config() automatically.
    """
    self.base_config = base_config.copy()
    self.problem = None
    self.parameter_samples = None
    self.simulation_results = None
    self.sensitivity_results = {}

    self._setup_chinese_font()
    self._validate_tricys_config()

analyze_fast(output_index=0, **kwargs)

Perform FAST sensitivity analysis

Parameters:

Name	Type	Description	Default
output_index	int	Output variable index	0
**kwargs		FAST analysis parameters	{}

Returns:

Type	Description
Dict[str, Any]	FAST sensitivity analysis results

Note

FAST analysis requires samples generated by the fast_sampler sampling method! Results from Morris or Sobol sampling cannot be used.

Source code in tricys/analysis/salib.py

def analyze_fast(self, output_index: int = 0, **kwargs) -> Dict[str, Any]:
    """
    Perform FAST sensitivity analysis

    Args:
        output_index: Output variable index
        **kwargs: FAST analysis parameters

    Returns:
        FAST sensitivity analysis results

    Note:
        FAST analysis requires samples generated by the fast_sampler sampling method!
        Results from Morris or Sobol sampling cannot be used.
    """
    if self.simulation_results is None:
        raise ValueError("The simulation must be run first to obtain the results.")

    if (
        hasattr(self, "_last_sampling_method")
        and self._last_sampling_method != "fast"
    ):
        logger.warning(
            f"⚠️ The current sampling method is {self._last_sampling_method}, but FAST analysis requires FAST sampling!"
        )
        logger.warning(
            "Suggestion: Regenerate samples using generate_samples('fast')"
        )

    Y = self.simulation_results[:, output_index]

    Y = self._handle_nan_values(Y, "FAST分析")

    # Remove NaN values
    # valid_indices = ~np.isnan(Y)
    # if not np.all(valid_indices):
    #    logger.warning(f"发现{np.sum(~valid_indices)}个无效结果，将被排除")
    #    Y = Y[valid_indices]

    try:
        # Perform FAST analysis
        Si = fast.analyze(self.problem, Y, **kwargs)

        if "fast" not in self.sensitivity_results:
            self.sensitivity_results["fast"] = {}

        metric_name = f"metric_{output_index}"
        self.sensitivity_results["fast"][metric_name] = {
            "output_index": output_index,
            "Si": Si,
            "S1": Si["S1"],
            "ST": Si["ST"],
            "sampling_method": getattr(self, "_last_sampling_method", "unknown"),
        }

        logger.info(
            f"FAST sensitivity analysis completed (indicator {output_index})"
        )
        return self.sensitivity_results["fast"][metric_name]

    except Exception as e:
        if "fast" in str(e).lower() or "sample" in str(e).lower():
            raise ValueError(
                f"FAST analysis failed, possibly due to incompatible sampling method: {e}\nPlease regenerate samples using generate_samples('fast')"
            ) from e
        else:
            raise

analyze_lhs(output_index=0, **kwargs)

Perform LHS (Latin Hypercube Sampling) uncertainty analysis

Note: This is a basic statistical analysis method for LHS samples, providing descriptive statistics and basic sensitivity indices.

Parameters:

Name	Type	Description	Default
output_index	int	Output variable index	0
**kwargs		Analysis parameters (reserved for future use)	{}

Returns:

Type	Description
Dict[str, Any]	LHS uncertainty analysis results

Source code in tricys/analysis/salib.py

def analyze_lhs(self, output_index: int = 0, **kwargs) -> Dict[str, Any]:
    """
    Perform LHS (Latin Hypercube Sampling) uncertainty analysis

    Note: This is a basic statistical analysis method for LHS samples,
    providing descriptive statistics and basic sensitivity indices.

    Args:
        output_index: Output variable index
        **kwargs: Analysis parameters (reserved for future use)

    Returns:
        LHS uncertainty analysis results
    """
    if self.simulation_results is None:
        raise ValueError("The simulation must be run first to obtain the results.")

    if (
        hasattr(self, "_last_sampling_method")
        and self._last_sampling_method != "latin"
    ):
        logger.warning(
            f"⚠️ The current sampling method is {self._last_sampling_method}, but LHS analysis is designed for Latin Hypercube Sampling!"
        )
        logger.warning(
            "Suggestion: Regenerate samples using generate_samples('latin')"
        )

    Y = self.simulation_results[:, output_index]

    # Handle NaN values
    Y = self._handle_nan_values(Y, "LHS分析")

    # Basic statistical analysis
    mean_val = np.mean(Y)
    std_val = np.std(Y)
    min_val = np.min(Y)
    max_val = np.max(Y)
    percentile_5 = np.percentile(Y, 5)
    percentile_95 = np.percentile(Y, 95)

    # Create results dictionary
    Si = {
        "mean": mean_val,
        "std": std_val,
        "min": min_val,
        "max": max_val,
        "percentile_5": percentile_5,
        "percentile_95": percentile_95,
    }

    if "latin" not in self.sensitivity_results:
        self.sensitivity_results["latin"] = {}

    metric_name = f"metric_{output_index}"
    self.sensitivity_results["latin"][metric_name] = {
        "output_index": output_index,
        "Si": Si,
        "mean": mean_val,
        "std": std_val,
        "min": min_val,
        "max": max_val,
        "percentile_5": percentile_5,
        "percentile_95": percentile_95,
        "sampling_method": getattr(self, "_last_sampling_method", "unknown"),
    }

    logger.info(f"LHS uncertainty analysis completed (指标 {output_index})")
    return self.sensitivity_results["latin"][metric_name]

analyze_morris(output_index=0, **kwargs)

Perform Morris sensitivity analysis

Parameters:

Name	Type	Description	Default
output_index	int	Output variable index	0
**kwargs		Morris analysis parameters	{}

Returns:

Type	Description
Dict[str, Any]	Morris sensitivity analysis results

Source code in tricys/analysis/salib.py

def analyze_morris(self, output_index: int = 0, **kwargs) -> Dict[str, Any]:
    """
    Perform Morris sensitivity analysis

    Args:
        output_index: Output variable index
        **kwargs: Morris analysis parameters

    Returns:
        Morris sensitivity analysis results
    """
    if self.simulation_results is None:
        raise ValueError("The simulation must be run first to obtain the results.")

    Y = self.simulation_results[:, output_index]

    Y = self._handle_nan_values(Y, "Morris分析")
    X = self.parameter_samples

    # Remove NaN values
    # valid_indices = ~np.isnan(Y)
    # if not np.all(valid_indices):
    #    logger.warning(f"发现{np.sum(~valid_indices)}个无效结果，将被排除")
    #    Y = Y[valid_indices]
    #    X = self.parameter_samples[valid_indices]
    # else:
    #    X = self.parameter_samples

    # Perform Morris analysis
    logger.info(
        f"Start Morris sensitivity analysis: X.shape={X.shape}, Y.shape={Y.shape}, X.dtype={X.dtype}"
    )

    try:
        Si = morris_analyze.analyze(self.problem, X, Y, **kwargs)
    except Exception as e:
        logger.error(f"Morris analysis execution failed: {e}")
        logger.error(f"problem: {self.problem}")
        logger.error(f"X shape: {X.shape}, type: {X.dtype}")
        logger.error(f"Yshape: {Y.shape}, type: {Y.dtype}")
        if hasattr(X, "dtype") and X.dtype == "object":
            logger.error(
                "X contains non-numeric data, please check the sampled data"
            )
        raise

    if "morris" not in self.sensitivity_results:
        self.sensitivity_results["morris"] = {}

    metric_name = f"metric_{output_index}"
    self.sensitivity_results["morris"][metric_name] = {
        "output_index": output_index,
        "Si": Si,
        "mu": Si["mu"],
        "mu_star": Si["mu_star"],
        "sigma": Si["sigma"],
        "mu_star_conf": Si["mu_star_conf"],
    }

    logger.info(f"Morris sensitivity analysis completed (metric {output_index})")
    return self.sensitivity_results["morris"][metric_name]

analyze_sobol(output_index=0, **kwargs)

Perform Sobol Sensitivity Analysis

Parameters:

Name	Type	Description	Default
output_index	int	Output variable index	0
**kwargs		Sobol analysis parameters	{}

Returns:

Type	Description
Dict[str, Any]	Sobol sensitivity analysis results

Note

Sobol analysis requires samples generated using the Saltelli sampling method! Results from Morris or FAST sampling cannot be used.

Source code in tricys/analysis/salib.py

def analyze_sobol(self, output_index: int = 0, **kwargs) -> Dict[str, Any]:
    """
    Perform Sobol Sensitivity Analysis

    Args:
        output_index: Output variable index
        **kwargs: Sobol analysis parameters

    Returns:
        Sobol sensitivity analysis results

    Note:
        Sobol analysis requires samples generated using the Saltelli sampling method!
        Results from Morris or FAST sampling cannot be used.
    """
    if self.simulation_results is None:
        raise ValueError("The simulation must be run first to obtain the results.")

    # Check sampling method compatibility
    if (
        hasattr(self, "_last_sampling_method")
        and self._last_sampling_method != "sobol"
    ):
        logger.warning(
            f"⚠️ Currently using {self._last_sampling_method} sampling, but Sobol analysis requires Saltelli sampling!"
        )
        logger.warning(
            "Suggestion: Regenerate samples using generate_samples('sobol')"
        )

    Y = self.simulation_results[:, output_index]

    Y = self._handle_nan_values(Y, "Sobol分析")

    # Remove NaN values
    # valid_indices = ~np.isnan(Y)
    # if not np.all(valid_indices):
    #    logger.warning(f"发现{np.sum(~valid_indices)}个无效结果，将被排除")
    #    Y = Y[valid_indices]
    #    X = self.parameter_samples[valid_indices]
    # else:
    #    X = self.parameter_samples

    try:
        Si = sobol.analyze(self.problem, Y, **kwargs)

        if "sobol" not in self.sensitivity_results:
            self.sensitivity_results["sobol"] = {}

        metric_name = f"metric_{output_index}"
        self.sensitivity_results["sobol"][metric_name] = {
            "output_index": output_index,
            "Si": Si,
            "S1": Si["S1"],
            "ST": Si["ST"],
            "S2": Si.get("S2", None),
            "S1_conf": Si["S1_conf"],
            "ST_conf": Si["ST_conf"],
            "sampling_method": getattr(self, "_last_sampling_method", "unknown"),
        }

        logger.info(f"Sobol sensitivity analysis completed (index {output_index})")
        return self.sensitivity_results["sobol"][metric_name]

    except Exception as e:
        if "saltelli" in str(e).lower() or "sample" in str(e).lower():
            raise ValueError(
                f"Sobol analysis failed, possibly due to incompatible sampling method: {e}\nPlease regenerate samples using generate_samples('sobol')"
            ) from e
        else:
            raise

define_problem(param_bounds, param_distributions=None)

Define SALib problem space.

Parameters:

Name	Type	Description	Default
param_bounds	Dict[str, Tuple[float, float]]	Parameter bounds dictionary {'param_name': (min_val, max_val)}.	required
param_distributions	Dict[str, str]	Parameter distribution type dictionary {'param_name': 'unif'/'norm'/etc}. Valid distribution types: 'unif', 'triang', 'norm', 'truncnorm', 'lognorm'.	None

Returns:

Type	Description
Dict[str, Any]	SALib problem definition dictionary.

Note

Defaults to 'unif' distribution if not specified. Validates distribution types and warns if invalid. Logs parameter definitions including bounds and distributions.

Source code in tricys/analysis/salib.py

def define_problem(
    self,
    param_bounds: Dict[str, Tuple[float, float]],
    param_distributions: Dict[str, str] = None,
) -> Dict[str, Any]:
    """Define SALib problem space.

    Args:
        param_bounds: Parameter bounds dictionary {'param_name': (min_val, max_val)}.
        param_distributions: Parameter distribution type dictionary {'param_name': 'unif'/'norm'/etc}.
            Valid distribution types: 'unif', 'triang', 'norm', 'truncnorm', 'lognorm'.

    Returns:
        SALib problem definition dictionary.

    Note:
        Defaults to 'unif' distribution if not specified. Validates distribution types
        and warns if invalid. Logs parameter definitions including bounds and distributions.
    """
    if param_distributions is None:
        param_distributions = {name: "unif" for name in param_bounds.keys()}

    valid_dists = ["unif", "triang", "norm", "truncnorm", "lognorm"]
    for name, dist in param_distributions.items():
        if dist not in valid_dists:
            logger.warning(
                "Invalid distribution type, using 'unif' instead",
                extra={
                    "parameter_name": name,
                    "invalid_distribution": dist,
                },
            )
            param_distributions[name] = "unif"

    self.problem = {
        "num_vars": len(param_bounds),
        "names": list(param_bounds.keys()),
        "bounds": list(param_bounds.values()),
        "dists": [
            param_distributions.get(name, "unif") for name in param_bounds.keys()
        ],
    }

    logger.info(
        "Defined a problem space",
        extra={
            "num_parameters": self.problem["num_vars"],
        },
    )
    for i, name in enumerate(self.problem["names"]):
        logger.info(
            "Parameter definition",
            extra={
                "parameter_name": name,
                "bounds": self.problem["bounds"][i],
                "distribution": self.problem["dists"][i],
            },
        )

    return self.problem

generate_samples(method='sobol', N=1024, **kwargs)

Generate parameter samples.

Parameters:

Name	Type	Description	Default
method	str	Sampling method ('sobol', 'morris', 'fast', 'latin').	'sobol'
N	int	Number of samples (for Sobol this is the base sample count, actual count is N(2D+2)).	1024
**kwargs		Method-specific parameters.	{}

Returns:

Type	Description
ndarray	Parameter sample array (n_samples, n_params).

Raises:

Type	Description
ValueError	If problem not defined or unsupported method.

Note

Sobol generates N(2D+2) samples. Morris generates N trajectories. Samples are rounded to 5 decimal places. Stores last sampling method for compatibility checking.

Source code in tricys/analysis/salib.py

def generate_samples(
    self, method: str = "sobol", N: int = 1024, **kwargs
) -> np.ndarray:
    """Generate parameter samples.

    Args:
        method: Sampling method ('sobol', 'morris', 'fast', 'latin').
        N: Number of samples (for Sobol this is the base sample count, actual count is N*(2*D+2)).
        **kwargs: Method-specific parameters.

    Returns:
        Parameter sample array (n_samples, n_params).

    Raises:
        ValueError: If problem not defined or unsupported method.

    Note:
        Sobol generates N*(2*D+2) samples. Morris generates N trajectories. Samples are
        rounded to 5 decimal places. Stores last sampling method for compatibility checking.
    """
    if self.problem is None:
        raise ValueError(
            "You must first call define_problem() to define the problem space."
        )

    logger.info(
        "Generating samples",
        extra={
            "method": method,
            "base_sample_count": N,
        },
    )

    if method.lower() == "sobol":
        # Sobol method: generate N*(2*D+2) samples
        self.parameter_samples = saltelli.sample(self.problem, N, **kwargs)
        actual_samples = N * (2 * self.problem["num_vars"] + 2)

    elif method.lower() == "morris":
        # Morris method: Generate N trajectories
        # Note: Different versions of SALib may have different parameter names
        morris_kwargs = {"num_levels": 4}
        # Check the SALib version and use the correct parameter names
        try:
            morris_kwargs.update(kwargs)
            self.parameter_samples = morris.sample(self.problem, N, **morris_kwargs)
        except TypeError as e:
            if "grid_jump" in str(e):
                morris_kwargs = {
                    k: v for k, v in morris_kwargs.items() if k != "grid_jump"
                }
                morris_kwargs.update(
                    {k: v for k, v in kwargs.items() if k != "grid_jump"}
                )
                self.parameter_samples = morris.sample(
                    self.problem, N, **morris_kwargs
                )
            else:
                raise e

        actual_samples = len(self.parameter_samples)

    elif method.lower() == "fast":
        # FAST method
        fast_kwargs = {"M": 4}
        fast_kwargs.update(kwargs)
        self.parameter_samples = fast_sampler.sample(self.problem, N, **fast_kwargs)
        actual_samples = len(self.parameter_samples)

    elif method.lower() == "latin":
        # Latin Hypercube Sampling
        self.parameter_samples = latin.sample(self.problem, N, **kwargs)
        actual_samples = N

    else:
        raise ValueError(f"Unsupported sampling method: {method}")

    logger.info(
        "Successfully generated samples", extra={"actual_samples": actual_samples}
    )

    if self.parameter_samples is not None:
        self.parameter_samples = np.round(self.parameter_samples, decimals=5)
        logger.info("Parameter sample precision adjusted to 5 decimal places")

    self._last_sampling_method = method.lower()

    return self.parameter_samples

generate_tricys_config(csv_file_path=None, output_metrics=None)

Generate Tricys configuration file for reading CSV parameter files and executing simulations This function reuses the base configuration and specifically modifies simulation_parameters and analysis_case for file-based SALib runs

Parameters:

Name	Type	Description	Default
csv_file_path	str	Path to the CSV parameter file. If None, the last generated file is used	None
output_metrics	List[str]	List of output metrics to be calculated	None

Returns:

Type	Description
Dict[str, Any]	Path of the generated configuration file

Source code in tricys/analysis/salib.py

def generate_tricys_config(
    self, csv_file_path: str = None, output_metrics: List[str] = None
) -> Dict[str, Any]:
    """
    Generate Tricys configuration file for reading CSV parameter files and executing simulations
    This function reuses the base configuration and specifically modifies simulation_parameters and analysis_case for file-based SALib runs

    Args:
        csv_file_path: Path to the CSV parameter file. If None, the last generated file is used
        output_metrics: List of output metrics to be calculated

    Returns:
        Path of the generated configuration file
    """
    if csv_file_path is None:
        if hasattr(self, "sampling_csv_path"):
            csv_file_path = self.sampling_csv_path
        else:
            raise ValueError(
                "CSV file path not found, please first call run_tricys_simulations() or specify csv_file_path"
            )

    if output_metrics is None:
        output_metrics = [
            "Startup_Inventory",
            "Self_Sufficiency_Time",
            "Doubling_Time",
        ]

    csv_abs_path = os.path.abspath(csv_file_path)

    import copy

    tricys_config = copy.deepcopy(self.base_config)
    tricys_config["simulation_parameters"] = {"file": csv_abs_path}

    if "sensitivity_analysis" not in tricys_config:
        tricys_config["sensitivity_analysis"] = {"enabled": True}

    tricys_config["sensitivity_analysis"]["analysis_case"] = {
        "name": "SALib_Analysis",
        "independent_variable": "file",
        "independent_variable_sampling": csv_abs_path,
        "dependent_variables": output_metrics,
    }

    return tricys_config

get_compatible_analysis_methods(sampling_method)

Get analysis methods compatible with the specified sampling method.

Parameters:

Name	Type	Description	Default
sampling_method	str	Sampling method	required

Returns:

Type	Description
List[str]	List of compatible analysis methods

Source code in tricys/analysis/salib.py

def get_compatible_analysis_methods(self, sampling_method: str) -> List[str]:
    """
    Get analysis methods compatible with the specified sampling method.

    Args:
        sampling_method: Sampling method

    Returns:
        List of compatible analysis methods
    """
    compatibility_map = {
        "sobol": ["sobol"],
        "morris": ["morris"],
        "fast": ["fast"],
        "latin": ["latin"],
        "unknown": [],
    }

    return compatibility_map.get(sampling_method, [])

load_tricys_results(sensitivity_summary_csv, output_metrics=None)

Read simulation results from the sensitivity_analysis_summary.csv file output by Tricys

Parameters:

Name	Type	Description	Default
sensitivity_summary_csv	str	Path to the sensitivity analysis summary CSV file output by Tricys	required
output_metrics	List[str]	List of output metrics to extract	None

Returns:

Type	Description
ndarray	Simulation result array (n_samples, n_metrics)

Source code in tricys/analysis/salib.py

def load_tricys_results(
    self, sensitivity_summary_csv: str, output_metrics: List[str] = None
) -> np.ndarray:
    """
    Read simulation results from the sensitivity_analysis_summary.csv file output by Tricys

    Args:
        sensitivity_summary_csv: Path to the sensitivity analysis summary CSV file output by Tricys
        output_metrics: List of output metrics to extract

    Returns:
        Simulation result array (n_samples, n_metrics)
    """
    if output_metrics is None:
        output_metrics = [
            "Startup_Inventory",
            "Self_Sufficiency_Time",
            "Doubling_Time",
        ]

    logger.info(f"Read data from the Tricys result file: {sensitivity_summary_csv}")

    df = pd.read_csv(sensitivity_summary_csv)

    logger.info(f"Read {len(df)} simulation results")
    logger.info(f"Result file columns: {list(df.columns)}")

    param_cols = []
    metric_cols = []

    for col in df.columns:
        if col in output_metrics:
            metric_cols.append(col)
        elif col in self.problem["names"] if self.problem else False:
            param_cols.append(col)

    logger.info(f"Recognized parameter columns: {param_cols}")
    logger.info(f"Identified metric columns: {metric_cols}")

    ordered_metric_cols = []
    for metric in output_metrics:
        if metric in metric_cols:
            ordered_metric_cols.append(metric)
        else:
            logger.warning(f"Metric column not found: {metric}")

    if not ordered_metric_cols:
        raise ValueError(f"No valid output metrics columns found: {output_metrics}")

    results_data = df[ordered_metric_cols].values

    self.simulation_results = results_data

    logger.info(f"Successfully loaded simulation results: {results_data.shape}")
    logger.info(
        f"Result Statistics:\n{pd.DataFrame(results_data, columns=ordered_metric_cols).describe()}"
    )
    logger.info(
        f"Result preview:\n{pd.DataFrame(results_data, columns=metric_cols).head()}"
    )
    return self.simulation_results

plot_fast_results(save_dir=None, figsize=(12, 8), metric_names=None)

Plot FAST analysis results

Source code in tricys/analysis/salib.py

def plot_fast_results(
    self,
    save_dir: str = None,
    figsize: Tuple[int, int] = (12, 8),
    metric_names: List[str] = None,
) -> None:
    """Plot FAST analysis results"""
    if "fast" not in self.sensitivity_results:
        raise ValueError("No analysis results found for the FAST method")

    # No analysis results found for the FAST method
    self._setup_chinese_font()

    if save_dir is None:
        save_dir = "."
    os.makedirs(save_dir, exist_ok=True)

    # Get the results of all indicators
    fast_results = self.sensitivity_results["fast"]

    if not fast_results:
        raise ValueError("FAST analysis results not found")

    # Generate a chart for each metric
    for metric_key, results in fast_results.items():
        Si = results["Si"]
        output_index = results["output_index"]

        # Determine the indicator name
        if metric_names and output_index < len(metric_names):
            metric_display_name = metric_names[output_index]
        else:
            metric_display_name = f"Metric_{output_index}"

        # Bar charts of first-order and total sensitivity indices
        fig, (ax1, ax2) = plt.subplots(1, 2, figsize=figsize)

        # first-order sensitivity index
        y_pos = np.arange(len(self.problem["names"]))
        ax1.barh(y_pos, Si["S1"], alpha=0.7, color="purple")
        ax1.set_yticks(y_pos)
        ax1.set_yticklabels(self.problem["names"], fontsize=10)
        ax1.set_xlabel("一阶敏感性指数 (S1)", fontsize=12)
        ax1.set_title(
            f"FAST First-order Sensitivity Indices\n{metric_display_name}",
            fontsize=14,
            pad=20,
        )
        ax1.grid(True, alpha=0.3)

        # Total Sensitivity Index
        ax2.barh(y_pos, Si["ST"], alpha=0.7, color="darkgreen")
        ax2.set_yticks(y_pos)
        ax2.set_yticklabels(self.problem["names"], fontsize=10)
        ax2.set_xlabel("总敏感性指数 (ST)", fontsize=12)
        ax2.set_title(
            f"FAST Total Sensitivity Indices\n{metric_display_name}",
            fontsize=14,
            pad=20,
        )
        ax2.grid(True, alpha=0.3)

        plt.tight_layout()
        filename = (
            f'fast_sensitivity_indices_{metric_display_name.replace(" ", "_")}.png'
        )
        plt.savefig(os.path.join(save_dir, filename), dpi=300, bbox_inches="tight")

        logger.info(f"The FAST result chart has been saved: {filename}")

plot_lhs_results(save_dir=None, figsize=(12, 8), metric_names=None)

Plot LHS (Latin Hypercube Sampling) uncertainty analysis results

Source code in tricys/analysis/salib.py

def plot_lhs_results(
    self,
    save_dir: str = None,
    figsize: Tuple[int, int] = (12, 8),
    metric_names: List[str] = None,
) -> None:
    """Plot LHS (Latin Hypercube Sampling) uncertainty analysis results"""
    if "latin" not in self.sensitivity_results:
        raise ValueError("No analysis results found for the LHS method")

    # Ensure Chinese font settings
    self._setup_chinese_font()

    if save_dir is None:
        save_dir = "."
    os.makedirs(save_dir, exist_ok=True)

    # Get the results of all indicators
    lhs_results = self.sensitivity_results["latin"]

    if not lhs_results:
        raise ValueError("LHS analysis results not found")

    # Generate charts for each metric
    for metric_key, results in lhs_results.items():
        Si = results["Si"]
        output_index = results["output_index"]

        # Determine the indicator name
        if metric_names and output_index < len(metric_names):
            metric_display_name = metric_names[output_index]
        else:
            metric_display_name = f"Metric_{output_index}"

        # Get unit from config
        sensitivity_analysis_config = self.base_config.get(
            "sensitivity_analysis", {}
        )
        unit_map = sensitivity_analysis_config.get("unit_map", {})
        unit_config = self._find_unit_config(metric_display_name, unit_map)
        unit_str = ""
        if unit_config:
            unit = unit_config.get("unit")
            if unit:
                unit_str = f" ({unit})"

        xlabel = f"{metric_display_name}{unit_str}"

        # Create a figure with two subplots
        fig, (ax1, ax2) = plt.subplots(1, 2, figsize=figsize)

        # Plot 1: Distribution histogram
        ax1.hist(
            self.simulation_results[:, output_index],
            bins=30,
            alpha=0.7,
            color="skyblue",
            edgecolor="black",
        )
        ax1.set_xlabel(xlabel, fontsize=12)
        ax1.set_ylabel("频率", fontsize=12)
        ax1.set_title("输出分布直方图", fontsize=14, pad=10)
        ax1.grid(True, alpha=0.3)

        # Add statistics text to the histogram plot
        stats_text = f"均值: {Si['mean']:.4f}\n标准差: {Si['std']:.4f}\n最小值: {Si['min']:.4f}\n最大值: {Si['max']:.4f}"
        ax1.text(
            0.05,
            0.95,
            stats_text,
            transform=ax1.transAxes,
            fontsize=10,
            verticalalignment="top",
            bbox=dict(boxstyle="round", facecolor="white", alpha=0.8),
        )

        # Plot 2: Cumulative distribution function
        sorted_data = np.sort(self.simulation_results[:, output_index])
        y_vals = np.arange(1, len(sorted_data) + 1) / len(sorted_data)
        ax2.plot(sorted_data, y_vals, linewidth=2, color="darkgreen")
        ax2.set_xlabel(xlabel, fontsize=12)
        ax2.set_ylabel("累积概率", fontsize=12)
        ax2.set_title("累积分布函数", fontsize=14, pad=10)
        ax2.grid(True, alpha=0.3)

        plt.tight_layout()
        filename = f'lhs_analysis_{metric_display_name.replace(" ", "_")}.png'
        plt.savefig(os.path.join(save_dir, filename), dpi=300, bbox_inches="tight")

        logger.info(f"LHS分析结果图表已保存: {filename}")

plot_morris_results(save_dir=None, figsize=(12, 8), metric_names=None)

Plot the Morris analysis results

Source code in tricys/analysis/salib.py

def plot_morris_results(
    self,
    save_dir: str = None,
    figsize: Tuple[int, int] = (12, 8),
    metric_names: List[str] = None,
) -> None:
    """Plot the Morris analysis results"""
    if "morris" not in self.sensitivity_results:
        raise ValueError("No analysis results were found for the Morris method.")

    # Ensure Chinese font settings
    self._setup_chinese_font()

    if save_dir is None:
        save_dir = "."
    os.makedirs(save_dir, exist_ok=True)

    # Obtain the results of all indicators
    morris_results = self.sensitivity_results["morris"]

    if not morris_results:
        raise ValueError("No Morris analysis results found")

    for metric_key, results in morris_results.items():
        Si = results["Si"]
        output_index = results["output_index"]

        if metric_names and output_index < len(metric_names):
            metric_display_name = metric_names[output_index]
        else:
            metric_display_name = f"Metric_{output_index}"

        # Morris μ*-σ diagram
        fig, (ax1, ax2) = plt.subplots(1, 2, figsize=figsize)

        # μ*-σ scatter plot
        ax1.scatter(Si["mu_star"], Si["sigma"], s=100, alpha=0.7, color="red")
        for i, name in enumerate(self.problem["names"]):
            ax1.annotate(
                name,
                (Si["mu_star"][i], Si["sigma"][i]),
                xytext=(5, 5),
                textcoords="offset points",
                fontsize=9,
            )

        ax1.set_xlabel("μ*(Average Absolute Effect)", fontsize=12)
        ax1.set_ylabel("σ (Standard Deviation)", fontsize=12)
        ax1.set_title(
            f"Morris μ*-σ Plot\n{metric_display_name}", fontsize=14, pad=20
        )
        ax1.grid(True, alpha=0.3)

        y_pos = np.arange(len(self.problem["names"]))
        ax2.barh(
            y_pos, Si["mu_star"], xerr=Si["mu_star_conf"], alpha=0.7, color="green"
        )
        ax2.set_yticks(y_pos)
        ax2.set_yticklabels(self.problem["names"], fontsize=10)
        ax2.set_xlabel("μ*(Average Absolute Effect)", fontsize=12)
        ax2.set_title(
            f"Morris Elementary Effects\n{metric_display_name}", fontsize=14, pad=20
        )
        ax2.grid(True, alpha=0.3)

        plt.tight_layout()
        filename = f'morris_sensitivity_analysis_{metric_display_name.replace(" ", "_")}.png'
        plt.savefig(os.path.join(save_dir, filename), dpi=300, bbox_inches="tight")

        logger.info(f"Morris result chart has been saved: {filename}")

plot_sobol_results(save_dir=None, figsize=(12, 8), metric_names=None)

Plot Sobol analysis results

Source code in tricys/analysis/salib.py

def plot_sobol_results(
    self,
    save_dir: str = None,
    figsize: Tuple[int, int] = (12, 8),
    metric_names: List[str] = None,
) -> None:
    """Plot Sobol analysis results"""
    if "sobol" not in self.sensitivity_results:
        raise ValueError("No analysis results for the Sobol method were found.")

    # Ensure Chinese font settings
    self._setup_chinese_font()

    if save_dir is None:
        save_dir = "."
    os.makedirs(save_dir, exist_ok=True)

    # Get the results of all indicators
    sobol_results = self.sensitivity_results["sobol"]

    if not sobol_results:
        raise ValueError("Sobol analysis results not found")

    # Generate charts for each metric
    for metric_key, results in sobol_results.items():
        Si = results["Si"]
        output_index = results["output_index"]

        if metric_names and output_index < len(metric_names):
            metric_display_name = metric_names[output_index]
        else:
            metric_display_name = f"Metric_{output_index}"

        # Bar chart of first-order and total sensitivity indices
        fig, (ax1, ax2) = plt.subplots(1, 2, figsize=figsize)

        # First-order sensitivity index
        y_pos = np.arange(len(self.problem["names"]))
        ax1.barh(y_pos, Si["S1"], xerr=Si["S1_conf"], alpha=0.7, color="skyblue")
        ax1.set_yticks(y_pos)
        ax1.set_yticklabels(self.problem["names"], fontsize=10)
        ax1.set_xlabel("First-order sensitivity index (S1)", fontsize=12)
        ax1.set_title(
            f"First-order Sensitivity Indices\n{metric_display_name}",
            fontsize=14,
            pad=20,
        )
        ax1.grid(True, alpha=0.3)

        # # Total Sensitivity Index
        ax2.barh(y_pos, Si["ST"], xerr=Si["ST_conf"], alpha=0.7, color="orange")
        ax2.set_yticks(y_pos)
        ax2.set_yticklabels(self.problem["names"], fontsize=10)
        ax2.set_xlabel("Total Sensitivity Index (ST)", fontsize=12)
        ax2.set_title(
            f"Total Sensitivity Indices\n{metric_display_name}", fontsize=14, pad=20
        )
        ax2.grid(True, alpha=0.3)

        plt.tight_layout()
        filename = (
            f'sobol_sensitivity_indices_{metric_display_name.replace(" ", "_")}.png'
        )
        plt.savefig(os.path.join(save_dir, filename), dpi=300, bbox_inches="tight")

        logger.info(f"Sobol result chart has been saved: {filename}")

run_salib_analysis_from_tricys_results(sensitivity_summary_csv, param_bounds=None, output_metrics=None, methods=['sobol', 'morris', 'fast'], save_dir=None)

Run a complete SALib sensitivity analysis from the sensitivity analysis results file output by Tricys

Parameters:

Name	Type	Description	Default
sensitivity_summary_csv	str	Path to the sensitivity summary CSV file output by Tricys	required
param_bounds	Dict[str, Tuple[float, float]]	Dictionary of parameter bounds, inferred from the CSV file if None	None
output_metrics	List[str]	List of output metrics to analyze	None
methods	List[str]	List of sensitivity analysis methods to execute	['sobol', 'morris', 'fast']
save_dir	str	Directory to save the results	None

Returns:

Type	Description
Dict[str, Any]	Dictionary containing all analysis results

Source code in tricys/analysis/salib.py

def run_salib_analysis_from_tricys_results(
    self,
    sensitivity_summary_csv: str,
    param_bounds: Dict[str, Tuple[float, float]] = None,
    output_metrics: List[str] = None,
    methods: List[str] = ["sobol", "morris", "fast"],
    save_dir: str = None,
) -> Dict[str, Any]:
    """
    Run a complete SALib sensitivity analysis from the sensitivity analysis results file output by Tricys

    Args:
        sensitivity_summary_csv: Path to the sensitivity summary CSV file output by Tricys
        param_bounds: Dictionary of parameter bounds, inferred from the CSV file if None
        output_metrics: List of output metrics to analyze
        methods: List of sensitivity analysis methods to execute
        save_dir: Directory to save the results

    Returns:
        Dictionary containing all analysis results
    """
    if output_metrics is None:
        output_metrics = [
            "Startup_Inventory",
            "Self_Sufficiency_Time",
            "Doubling_Time",
        ]

    if save_dir is None:
        save_dir = os.path.join(
            os.path.dirname(sensitivity_summary_csv), "salib_analysis"
        )
    os.makedirs(save_dir, exist_ok=True)

    df = pd.read_csv(sensitivity_summary_csv)

    if param_bounds is None:
        param_bounds = {}
        param_candidates = []
        for col in df.columns:
            if col not in output_metrics and "." in col:
                param_candidates.append(col)

        for param in param_candidates:
            param_data = df[param].dropna()
            if len(param_data) > 0:
                param_bounds[param] = (param_data.min(), param_data.max())

    if not param_bounds:
        raise ValueError(
            "Unable to determine parameter boundaries, please provide the param_bounds parameter"
        )

    self.define_problem(param_bounds)

    self.load_tricys_results(sensitivity_summary_csv, output_metrics)

    detected_method = self._last_sampling_method

    methods = self.get_compatible_analysis_methods(detected_method)

    all_results = {}

    for metric_idx, metric_name in enumerate(output_metrics):
        if metric_idx >= self.simulation_results.shape[1]:
            logger.warning(f"The metric {metric_name} is out of range, skipping")
            continue

        logger.info(f"\n=== Analysis indicators: {metric_name} ===")
        metric_results = {}

        # Check data validity
        Y = self.simulation_results[:, metric_idx]
        valid_ratio = np.sum(~np.isnan(Y)) / len(Y)
        logger.info(f"Valid data ratio: {valid_ratio:.2%}")

        if valid_ratio < 0.5:
            logger.warning(
                f"The metric {metric_name} has less than 50% valid data, which may affect the analysis quality."
            )

        # Sobol analysis
        if "sobol" in methods:
            try:
                logger.info("Performing Sobol sensitivity analysis...")
                sobol_result = self.analyze_sobol(output_index=metric_idx)
                metric_results["sobol"] = sobol_result

                # Display Sobol results summary
                logger.info("\nSobol sensitivity index:")
                for i, param_name in enumerate(self.problem["names"]):
                    s1 = sobol_result["S1"][i]
                    st = sobol_result["ST"][i]
                    logger.info(f"  {param_name}: S1={s1:.4f}, ST={st:.4f}")

            except Exception as e:
                logger.error(f"Sobol analysis failed: {e}")

        # Morris analysis
        if "morris" in methods:
            try:
                logger.info("Performing Morris sensitivity analysis...")
                morris_result = self.analyze_morris(output_index=metric_idx)
                metric_results["morris"] = morris_result

                # Display Morris results summary
                logger.info("\nMorris sensitivity index:")
                for i, param_name in enumerate(self.problem["names"]):
                    mu_star = morris_result["mu_star"][i]
                    sigma = morris_result["sigma"][i]
                    logger.info(f"  {param_name}: μ*={mu_star:.4f}, σ={sigma:.4f}")

            except Exception as e:
                logger.error(f"Morris analysis failed: {e}")

        # FAST analysis
        if "fast" in methods:
            try:
                logger.info("Performing FAST sensitivity analysis...")
                fast_result = self.analyze_fast(output_index=metric_idx)
                metric_results["fast"] = fast_result

                # Display FAST results summary
                logger.info("\nFAST sensitivity index:")
                for i, param_name in enumerate(self.problem["names"]):
                    s1 = fast_result["S1"][i]
                    st = fast_result["ST"][i]
                    logger.info(f"  {param_name}: S1={s1:.4f}, ST={st:.4f}")

            except Exception as e:
                logger.error(f"FAST analysis failed: {e}")

        # LHS analysis
        if "latin" in methods:
            try:
                logger.info("Performing LHS uncertainty analysis...")
                lhs_result = self.analyze_lhs(output_index=metric_idx)
                metric_results["latin"] = lhs_result

                # Display LHS results summary
                logger.info("\nLHS分析结果:")
                logger.info(f"  均值: {lhs_result['mean']:.4f}")
                logger.info(f"  标准差: {lhs_result['std']:.4f}")
                logger.info(f"  最小值: {lhs_result['min']:.4f}")
                logger.info(f"  最大值: {lhs_result['max']:.4f}")
                logger.info(f"  5%分位数: {lhs_result['percentile_5']:.4f}")
                logger.info(f"  95%分位数: {lhs_result['percentile_95']:.4f}")

            except Exception as e:
                logger.error(f"LHS分析失败: {e}")

        all_results[metric_name] = metric_results

    try:
        if "sobol" in methods and "sobol" in self.sensitivity_results:
            self.plot_sobol_results(save_dir=save_dir, metric_names=output_metrics)

        if "morris" in methods and "morris" in self.sensitivity_results:
            self.plot_morris_results(save_dir=save_dir, metric_names=output_metrics)

        if "fast" in methods and "fast" in self.sensitivity_results:
            self.plot_fast_results(save_dir=save_dir, metric_names=output_metrics)

        # Plot LHS results
        if "latin" in methods and "latin" in self.sensitivity_results:
            self.plot_lhs_results(save_dir=save_dir, metric_names=output_metrics)

    except Exception as e:
        logger.warning(f"Drawing failed: {e}")

    try:
        self.save_results(
            save_dir=save_dir, format="csv", metric_names=output_metrics
        )

        report_content = self._save_sensitivity_report(all_results, save_dir)
        report_path = os.path.join(save_dir, "analysis_report.md")

        load_dotenv()

        # --- LLM Calls for analysis ---
        api_key = os.environ.get("API_KEY")
        base_url = os.environ.get("BASE_URL")
        ai_model = os.environ.get("AI_MODEL")

        sa_config = self.base_config.get("sensitivity_analysis", {})
        case_config = sa_config.get("analysis_case", {})
        ai_config = case_config.get("ai")

        ai_enabled = False
        if isinstance(ai_config, bool):
            ai_enabled = ai_config
        elif isinstance(ai_config, dict):
            ai_enabled = ai_config.get("enabled", False)

        if api_key and base_url and ai_model and ai_enabled:
            # First LLM call for initial analysis
            wrapper_prompt, llm_summary = call_llm_for_salib_analysis(
                report_content=report_content,
                api_key=api_key,
                base_url=base_url,
                ai_model=ai_model,
                method=detected_method,
            )
            if wrapper_prompt and llm_summary:
                with open(report_path, "a", encoding="utf-8") as f:
                    f.write("\n\n---\n\n# AI模型分析提示词\n\n")
                    f.write("```markdown\n")
                    f.write(wrapper_prompt)
                    f.write("\n```\n\n")
                    f.write("\n\n---\n\n# AI模型分析结果\n\n")
                    f.write(llm_summary)
                logger.info(f"Appended LLM prompt and summary to {report_path}")

                # Second LLM call for academic report
                glossary_path = None
                if isinstance(case_config, dict):
                    glossary_path = sa_config.get("glossary_path")

                if glossary_path and os.path.exists(glossary_path):
                    try:
                        with open(glossary_path, "r", encoding="utf-8") as f:
                            glossary_content = f.read()

                        (
                            academic_wrapper_prompt,
                            academic_report,
                        ) = call_llm_for_academic_report(
                            analysis_report=llm_summary,
                            glossary_content=glossary_content,
                            api_key=api_key,
                            base_url=base_url,
                            ai_model=ai_model,
                            problem_details=self.problem,
                            metric_names=output_metrics,
                            method=detected_method,
                            save_dir=save_dir,
                        )

                        if academic_wrapper_prompt and academic_report:
                            academic_report_path = os.path.join(
                                save_dir, "academic_report.md"
                            )
                            with open(
                                academic_report_path, "w", encoding="utf-8"
                            ) as f:
                                f.write(academic_report)
                            logger.info(
                                f"Generated academic report: {academic_report_path}"
                            )
                    except Exception as e:
                        logger.error(
                            f"Failed to generate or save academic report: {e}"
                        )
                elif glossary_path:
                    logger.warning(
                        f"Glossary file not found at {glossary_path}, skipping academic report generation."
                    )

        else:
            logger.warning(
                "API_KEY, BASE_URL, or AI_MODEL not set, or AI analysis is disabled. Skipping LLM summary generation."
            )

    except Exception as e:
        logger.warning(f"Failed to save result: {e}")

    logger.info("\n✅ SALib sensitivity analysis completed!")
    logger.info(f"📁 The result has been saved to: {save_dir}")

    return all_results

run_tricys_analysis(csv_file_path=None, output_metrics=None)

Run the Tricys simulation using the generated CSV parameter file and obtain the sensitivity analysis results

Parameters:

Name	Type	Description	Default
csv_file_path	str	Path to the CSV parameter file. If None, the last generated file will be used	None
output_metrics	List[str]	List of output metrics to be calculated	None
config_output_path		Path for the configuration file output. If None, it will be automatically generated	required

Returns:

Type	Description
str	Path to the sensitivity_analysis_summary.csv file

Source code in tricys/analysis/salib.py

def run_tricys_analysis(
    self, csv_file_path: str = None, output_metrics: List[str] = None
) -> str:
    """
    Run the Tricys simulation using the generated CSV parameter file and obtain the sensitivity analysis results

    Args:
        csv_file_path: Path to the CSV parameter file. If None, the last generated file will be used
        output_metrics: List of output metrics to be calculated
        config_output_path: Path for the configuration file output. If None, it will be automatically generated

    Returns:
        Path to the sensitivity_analysis_summary.csv file
    """
    # Generate Tricys configuration file
    tricys_config = self.generate_tricys_config(
        csv_file_path=csv_file_path, output_metrics=output_metrics
    )

    logger.info("Starting Tricys simulation analysis...")

    try:
        # Call the Tricys simulation engine
        from datetime import datetime

        from tricys.simulation.simulation_analysis import run_simulation

        tricys_config["run_timestamp"] = datetime.now().strftime("%Y%m%d_%H%M%S")

        run_simulation(tricys_config)

        results_dir = tricys_config["paths"]["results_dir"]

        return Path(results_dir) / "sensitivity_analysis_summary.csv"

    except Exception as e:
        logger.error(f"Tricys simulation execution failed: {e}")
        raise

run_tricys_simulations(output_metrics=None)

Generate sampling parameters and output them as a CSV file, which can be subsequently read by the Tricys simulation engine.

Parameters:

Name	Type	Description	Default
output_metrics	List[str]	List of output metrics to be extracted (for recording but does not affect CSV generation)	None
max_workers		Number of concurrent worker processes (reserved for compatibility, currently unused)	required

Returns:

Type	Description
str	Path to the generated CSV file

Source code in tricys/analysis/salib.py

def run_tricys_simulations(self, output_metrics: List[str] = None) -> str:
    """
    Generate sampling parameters and output them as a CSV file, which can be subsequently read by the Tricys simulation engine.

    Args:
        output_metrics: List of output metrics to be extracted (for recording but does not affect CSV generation)
        max_workers: Number of concurrent worker processes (reserved for compatibility, currently unused)

    Returns:
        Path to the generated CSV file
    """
    if self.parameter_samples is None:
        raise ValueError(
            "You must first call generate_samples() to generate samples."
        )

    if output_metrics is None:
        output_metrics = [
            "Startup_Inventory",
            "Self_Sufficiency_Time",
            "Doubling_Time",
        ]

    logger.info("Target output metrics", extra={"output_metrics": output_metrics})

    sampled_param_names = self.problem["names"]

    base_params = self.base_config.get("simulation_parameters", {}).copy()
    csv_output_path = (
        Path(self.base_config.get("paths", {}).get("temp_dir"))
        / "salib_sampling.csv"
    )

    os.makedirs(os.path.dirname(csv_output_path), exist_ok=True)

    param_data = []
    for i, sample in enumerate(self.parameter_samples):
        sampled_params = {
            sampled_param_names[j]: sample[j]
            for j in range(len(sampled_param_names))
        }

        job_params = base_params.copy()
        job_params.update(sampled_params)

        param_data.append(job_params)

    df = pd.DataFrame(param_data)

    for col in df.columns:
        if df[col].dtype in ["float64", "float32"]:
            df[col] = df[col].round(5)

    df.to_csv(csv_output_path, index=False, encoding="utf-8")

    logger.info(
        "Successfully generated parameter samples",
        extra={"num_samples": len(param_data)},
    )
    logger.info("Parameter file saved", extra={"file_path": csv_output_path})
    logger.info("Parameter file columns", extra={"columns": list(df.columns)})
    logger.info("Parameter precision set to 5 decimal places")
    logger.info("Sample statistics", extra={"statistics": df.describe().to_dict()})

    self.sampling_csv_path = csv_output_path

    return csv_output_path

save_results(save_dir=None, format='csv', metric_names=None)

Save sensitivity analysis results

Parameters:

Name	Type	Description	Default
save_dir	str	Save directory	None
format	str	Save format ('csv	'csv'

Source code in tricys/analysis/salib.py

def save_results(
    self, save_dir: str = None, format: str = "csv", metric_names: List[str] = None
) -> None:
    """
    Save sensitivity analysis results

    Args:
        save_dir: Save directory
        format: Save format ('csv
    """
    if save_dir is None:
        save_dir = "."
    os.makedirs(save_dir, exist_ok=True)

    for method, method_results in self.sensitivity_results.items():
        if not method_results:
            continue

        for metric_key, results in method_results.items():
            output_index = results["output_index"]

            if metric_names and output_index < len(metric_names):
                metric_display_name = metric_names[output_index]
            else:
                metric_display_name = f"Metric_{output_index}"

            if format == "csv":
                if method == "sobol":
                    sobol_df = pd.DataFrame(
                        {
                            "Parameter": self.problem["names"],
                            "S1": results["S1"],
                            "ST": results["ST"],
                            "S1_conf": results["S1_conf"],
                            "ST_conf": results["ST_conf"],
                        }
                    )
                    filename = (
                        f'sobol_indices_{metric_display_name.replace(" ", "_")}.csv'
                    )
                    sobol_df.to_csv(os.path.join(save_dir, filename), index=False)
                    logger.info(f"Sobol results have been saved: {filename}")

                elif method == "morris":
                    morris_df = pd.DataFrame(
                        {
                            "Parameter": self.problem["names"],
                            "mu": results["mu"],
                            "mu_star": results["mu_star"],
                            "sigma": results["sigma"],
                            "mu_star_conf": results["mu_star_conf"],
                        }
                    )
                    filename = f'morris_indices_{metric_display_name.replace(" ", "_")}.csv'
                    morris_df.to_csv(os.path.join(save_dir, filename), index=False)
                    logger.info(f"Morris results have been saved: {filename}")

                elif method == "fast":
                    fast_df = pd.DataFrame(
                        {
                            "Parameter": self.problem["names"],
                            "S1": results["S1"],
                            "ST": results["ST"],
                        }
                    )
                    filename = (
                        f'fast_indices_{metric_display_name.replace(" ", "_")}.csv'
                    )
                    fast_df.to_csv(os.path.join(save_dir, filename), index=False)
                    logger.info(f"FAST results have been saved: {filename}")

                elif method == "latin":
                    # Save LHS statistics
                    lhs_stats_df = pd.DataFrame(
                        {
                            "Metric": [metric_display_name],
                            "Mean": [results["mean"]],
                            "Std": [results["std"]],
                            "Min": [results["min"]],
                            "Max": [results["max"]],
                            "Percentile_5": [results["percentile_5"]],
                            "Percentile_95": [results["percentile_95"]],
                        }
                    )
                    filename_stats = (
                        f'lhs_stats_{metric_display_name.replace(" ", "_")}.csv'
                    )
                    lhs_stats_df.to_csv(
                        os.path.join(save_dir, filename_stats), index=False
                    )
                    logger.info(f"LHS统计结果已保存: {filename_stats}")

                    # Remove LHS sensitivity indices saving
                    # lhs_sens_df = pd.DataFrame({
                    #     "Parameter": self.problem["names"],
                    #     "Partial_Correlation": results["partial_correlations"]
                    # })
                    # filename_sens = f'lhs_sensitivity_{metric_display_name.replace(" ", "_")}.csv'
                    # lhs_sens_df.to_csv(os.path.join(save_dir, filename_sens), index=False)
                    # logger.info(f"LHS敏感性结果已保存: {filename_sens}")

    logger.info(f"The result has been saved to: {save_dir}")

call_llm_for_academic_report(analysis_report, glossary_content, api_key, base_url, ai_model, problem_details, metric_names, method, save_dir)

Sends an analysis report and a glossary to an LLM to generate a professional academic report.

Source code in tricys/analysis/salib.py

def call_llm_for_academic_report(
    analysis_report: str,
    glossary_content: str,
    api_key: str,
    base_url: str,
    ai_model: str,
    problem_details: dict,
    metric_names: list,
    method: str,
    save_dir: str,
) -> Tuple[str, str]:
    """Sends an analysis report and a glossary to an LLM to generate a professional academic report."""
    try:
        logger.info("Proceeding with LLM for academic report generation.")

        param_names_str = ", ".join(
            [f"`{name}`" for name in problem_details.get("names", [])]
        )
        metric_names_str = ", ".join([f"`{name}`" for name in metric_names])

        all_plots = [f for f in os.listdir(save_dir) if f.endswith((".svg", ".png"))]
        plot_list_str = "\n".join([f"    *   `{plot}`" for plot in all_plots])

        method_details = {
            "sobol": {
                "name": "Sobol",
                "methodology": "指出本次分析采用了SALib库，并使用了**Sobol方法**。这是一种基于方差的全局敏感性分析技术，能够量化单个参数以及参数间交互作用对模型输出方差的贡献。",
                "results_discussion": """*   对于每个性能指标，哪些输入参数的一阶敏感性（S1）和总体敏感性（ST）最高？请结合图表（如条形图）进行解读。
        *   S1和ST指数之间的差异揭示了什么？（例如，ST显著大于S1意味着该参数与其他参数存在显著的交互作用或其影响是非线性的）。
        *   分析不同指标之间的**权衡关系 (Trade-offs)**。例如，某个参数对某个指标 (e.g., `Startup_Inventory`) 有正面影响，但可能对另一个指标 (e.g., `Doubling_Time`) 有负面影响。""",
            },
            "morris": {
                "name": "Morris",
                "methodology": "指出本次分析采用了SALib库，并使用了**Morris方法**。这是一种基于轨迹的“一次性”设计方法，常用于在高维参数空间中进行参数筛选，以识别出影响最大的少数几个参数。",
                "results_discussion": """*   对于每个性能指标，哪些参数的 `μ*` (mu_star) 值最高，表明其对输出的总体影响最重要？
        *   `σ` (sigma) 值的大小又说明了什么？较高的 `σ` 值通常表明参数具有非线性效应或与其他参数存在强烈的交互作用。
        *   请结合 `μ*-σ` 图进行分析，对参数进行分类（例如，高 `μ*`/高 `σ` vs. 高 `μ*`/低 `σ`），并解释其含义。
        *   分析不同指标之间的**权衡关系 (Trade-offs)**。例如，某个参数对某个指标 (e.g., `Startup_Inventory`) 有正面影响，但可能对另一个指标 (e.g., `Doubling_Time`) 有负面影响。""",
            },
            "fast": {
                "name": "FAST",
                "methodology": "指出本次分析采用了SALib库，并使用了**FAST（傅里叶幅度敏感性检验）方法**。这是一种基于频率的全局敏感性分析技术，通过将参数在傅里叶级数中展开来计算敏感性指数。",
                "results_discussion": """*   对于每个性能指标，哪些输入参数的一阶敏感性（S1）最高？
        *   （如果可用）总体敏感性（ST）与一阶敏感性（S1）的比较揭示了什么？较大的差异通常表明存在参数交互。
        *   分析不同指标之间的**权衡关系 (Trade-offs)**。例如，某个参数对某个指标 (e.g., `Startup_Inventory`) 有正面影响，但可能对另一个指标 (e.g., `Doubling_Time`) 有负面影响。""",
            },
        }

        if method == "latin":
            ACADEMIC_REPORT_PROMPT_WRAPPER = f"""**角色：** 您是一位在核聚变工程，特别是氚燃料循环领域，具有深厚学术背景的资深科学家，擅长进行**不确定性量化 (UQ)** 和风险评估。

**任务：** 您收到了一个基于**拉丁超立方采样 (LHS)** 的不确定性分析初步报告和一份专业术语表。请您基于这两份文件，撰写一份更加专业、正式、符合学术发表标准的深度分析总结报告。

**指令：**

1.  **专业化语言：** 将初步报告中的模型参数/缩写替换为术语表中对应的专业词汇。
2.  **学术化重述：** 用严谨、客观的学术语言重新组织和阐述初步报告中的发现，聚焦于**不确定性**的量化和解读。
3.  **图表和表格的呈现与引用：**
    *   **显示图表：** 在报告的“结果与讨论”部分，您**必须**使用Markdown语法 `![图表标题](图表文件名)` 来**直接嵌入**和显示初步报告中包含的所有图表。可用的图表文件如下：
{plot_list_str}
    *   **引用图表：** 在正文中分析和讨论图表内容时，请使用“如图1所示...”等方式对图表进行编号和文字引用。
    *   **显示表格：** 当呈现数据时（例如，统计摘要、分布数据等），您**必须**使用Markdown的管道表格（pipe-table）格式来清晰地展示它们。您可以直接复用或重新格式化初步报告中的数据表格。
4.  **结构化报告：** 您的报告是关于一项**不确定性分析**。报告应包含以下部分：
    *   **摘要 (Abstract):** 简要概括本次不确定性研究的目的，明确指出分析的输入参数是 {param_names_str}，总结这些参数的不确定性对关键性能指标 ({metric_names_str}) 的输出分布（如均值、标准差、置信区间）有何影响。
    *   **引言 (Introduction):** 描述进行这项不确定性分析的背景和重要性。阐述研究目标，即量化评估当输入参数 {param_names_str} 在其定义域内变化时，氚燃料循环系统关键性能指标的统计分布和稳定性。
    *   **方法 (Methodology):** 简要说明分析方法。指出本次分析采用了拉丁超立方采样（LHS）方法来对输入参数空间进行抽样。说明被评估的关键性能指标是 {metric_names_str}，以及输入参数的概率分布和范围。
    *   **结果与讨论 (Results and Discussion):** 这是报告的核心。请结合初步报告中的统计数据和您嵌入的图表（如直方图、累积分布图），分点详细论述：
        *   对于每个性能指标，其输出的**概率分布**是怎样的？（例如，是正态分布、偏态分布还是双峰分布？）
        *   输出指标的**不确定性范围**有多大？（参考标准差和5%-95%百分位数区间）。这个范围在工程实践中是否可以接受？
        *   是否存在某些指标的波动范围过大，可能导致系统性能低于设计要求或存在运行风险？
    *   **结论 (Conclusion):** 总结本次不确定性分析得出的主要学术结论（例如，模型的稳定性、输出指标的可靠性等），并对降低关键指标不确定性或未来的风险评估提出具体建议。
5.  **输出格式：** 请直接输出完整的学术分析报告正文，确保所有内容都遵循正确的Markdown语法。

**输入文件：**
"""
        else:
            selected_method = method_details.get(method)
            if not selected_method:
                # Fallback for unknown methods
                selected_method = {
                    "name": method.capitalize(),
                    "methodology": f"指出本次分析采用了SALib库，并提及具体的敏感性分析方法为**{method.capitalize()}**。",
                    "results_discussion": "*   对于每个性能指标，识别出最重要的输入参数。\n*   讨论这些发现的意义。",
                }

            ACADEMIC_REPORT_PROMPT_WRAPPER = f"""**角色：** 您是一位在核聚变工程，特别是氚燃料循环领域，具有深厚学术背景的资深科学家。

**任务：** 您收到了一个关于**SALib {selected_method['name']} 方法敏感性分析**的程序生成的初步报告和一份专业术语表。请您基于这两份文件，撰写一份更加专业、正式、符合学术发表标准的深度分析总结报告。

**指令：**

1.  **专业化语言：** 将初步报告中的模型参数/缩写（例如 `sds.I[1]`, `Startup_Inventory`）替换为术语表中对应的“中文翻译”或“英文术语”。
2.  **学术化重述：** 用严谨、客观的学术语言重新组织和阐述初步报告中的发现。
3.  **图表和表格的呈现与引用：**
    *   **显示图表：** 在报告的“结果与讨论”部分，您**必须**使用Markdown语法 `![图表标题](图表文件名)` 来**直接嵌入**和显示初步报告中包含的所有图表。可用的图表文件如下：
{plot_list_str}
    *   **引用图表：** 在正文中分析和讨论图表内容时，请使用“如图1所示...”等方式对图表进行编号和文字引用。
    *   **显示表格：** 当呈现数据时（例如，敏感性指数表），您**必须**使用Markdown的管道表格（pipe-table）格式来清晰地展示它们。您可以直接复用或重新格式化初步报告中的数据表格。
4.  **结构化报告：** 您的报告是关于一项**敏感性分析**。报告应包含以下部分：
    *   **摘要 (Abstract):** 简要概括本次敏感性研究的目的，明确指明分析的输入参数是 {param_names_str}，总结哪些参数对关键性能指标 ({metric_names_str}) 影响最显著，并陈述核心结论。
    *   **引言 (Introduction):** 描述进行这项敏感性分析的背景和重要性。阐述研究目标，即量化评估输入参数的变化对氚燃料循环系统性能的影响。
    *   **方法 (Methodology):** {selected_method['methodology']} 说明被评估的关键性能指标是 {metric_names_str}，以及输入参数 {param_names_str} 的变化范围。
    *   **结果与讨论 (Results and Discussion):** 这是报告的核心。请结合初步报告中的数据和您嵌入的图表，分点详细论述：
{selected_method['results_discussion']}
    *   **结论 (Conclusion):** 总结本次敏感性分析得出的主要学术结论，并对反应堆设计或未来研究方向提出具体建议。
5.  **输出格式：** 请直接输出完整的学术分析报告正文，确保所有内容都遵循正确的Markdown语法。

**输入文件：**
"""

        full_prompt = f"{ACADEMIC_REPORT_PROMPT_WRAPPER}\n\n---\n### 1. 初步分析报告\n---\n{analysis_report}\n\n---\n### 2. 专业术语表\n---\n{glossary_content}"

        max_retries = 3
        for attempt in range(max_retries):
            try:
                client = openai.OpenAI(api_key=api_key, base_url=base_url)
                logger.info(
                    f"Sending data for academic report to LLM (Attempt {attempt + 1}/{max_retries})..."
                )

                response = client.chat.completions.create(
                    model=ai_model,
                    messages=[{"role": "user", "content": full_prompt}],
                    max_tokens=4000,
                )
                academic_report = response.choices[0].message.content

                logger.info("LLM academic report generation successful.")
                return ACADEMIC_REPORT_PROMPT_WRAPPER, academic_report

            except Exception as e:
                logger.error(
                    f"Error calling LLM for academic report on attempt {attempt + 1}: {e}"
                )
                if attempt < max_retries - 1:
                    time.sleep(5)
                else:
                    logger.error(
                        f"Failed to get LLM academic report after {max_retries} attempts."
                    )
                    return None, None

    except Exception as e:
        logger.error(f"Error in call_llm_for_academic_report: {e}", exc_info=True)
        return None, None

call_llm_for_salib_analysis(report_content, api_key, base_url, ai_model, method)

Sends a SALib analysis report to an LLM for summarization and returns the prompt and summary.

Source code in tricys/analysis/salib.py

def call_llm_for_salib_analysis(
    report_content: str, api_key: str, base_url: str, ai_model: str, method: str
) -> Tuple[str, str]:
    """Sends a SALib analysis report to an LLM for summarization and returns the prompt and summary."""
    try:
        logger.info("Proceeding with LLM analysis for SALib report.")

        PROMPT_TEMPLATES = {
            "sobol": """**角色：** 你是一名在氚燃料循环领域具有深厚背景的敏感性分析专家。

**任务：** 请仔细审查并解读以下这份由SALib库生成的**Sobol敏感性分析**报告。你的目标是：
1.  **总结核心发现**：对于报告中提到的每一个输出指标（如“启动氚量”等），总结其敏感性分析结果。
2.  **识别关键参数**：明确指出哪些输入参数的**一阶敏感性指数（S1）**和**总敏感性指数（ST）**最高。
3.  **解读指数含义**：解释S1和ST指数的含义。例如，高S1值表示参数对输出有重要的直接影响，而ST与S1的显著差异表示参数存在强烈的交互作用或非线性效应。
4.  **提供综合结论**：基于所有分析结果，对模型的整体行为、参数间的相互作用，以及这些发现对工程实践的潜在启示，给出一个综合性的结论。

请确保你的分析清晰、专业，并直接切入要点。
""",
            "morris": """**角色：** 你是一名在氚燃料循环领域具有深厚背景的敏感性分析专家。

**任务：** 请仔细审查并解读以下这份由SALib库生成的**Morris敏感性分析**报告。你的目标是：
1.  **总结核心发现**：对于报告中提到的每一个输出指标（如“启动氚量”等），总结其敏感性分析结果。
2.  **识别关键参数**：根据**μ* (mu_star)**值对参数进行排序，识别出对模型输出影响最大的参数。
3.  **解读参数效应**：解释**μ***和**σ (sigma)**的含义。高μ*表示参数有重要影响，高σ表示参数存在非线性影响或与其他参数有交互作用。结合μ*-σ图进行分析。
4.  **提供综合结论**：基于所有分析结果，对模型的整体行为、参数间的相互作用，以及这些发现对工程实践的潜在启示，给出一个综合性的结论。

请确保你的分析清晰、专业，并直接切入要点。
""",
            "fast": """**角色：** 你是一名在氚燃料循环领域具有深厚背景的敏感性分析专家。

**任务：** 请仔细审查并解读以下这份由SALib库生成的**FAST敏感性分析**报告。你的目标是：
1.  **总结核心发现**：对于报告中提到的每一个输出指标（如“启动氚量”等），总结其敏感性分析结果。
2.  **识别关键参数**：明确指出哪些输入参数的**一阶敏感性指数（S1）**和**总敏感性指数（ST）**最高。
3.  **解读指数含义**：解释S1和ST指数的含义。高S1值表示参数对输出有重要的直接影响，而ST与S1的差异表示参数可能存在交互作用。
4.  **提供综合结论**：基于所有分析结果，对模型的整体行为、参数间的相互作用，以及这些发现对工程实践的潜在启示，给出一个综合性的结论。

请确保你的分析清晰、专业，并直接切入要点。
""",
            "latin": """**角色：** 你是一名在氚燃料循环领域具有深厚背景的统计学和不确定性分析专家。

**任务：** 请仔细审查并解读以下这份由拉丁超立方采样（LHS）生成的不确定性分析报告。你的目标是：
1.  **解读统计数据**：对于报告中的每一个输出指标（如“启动氚量”等），解读其均值、标准差、最大/最小值和百分位数。
2.  **评估不确定性**：基于标准差和5%/95%百分位数的范围，评估模型输出结果的不确定性或波动范围有多大。
3.  **提供综合结论**：总结在给定的参数不确定性下，模型的关键性能指标（KPIs）表现如何，是否存在较大的风险（例如，输出值波动范围过大），并对模型的稳定性给出评价。

请确保你的分析聚焦于不确定性的量化和解读，而不是参数的敏感性排序。
""",
        }

        wrapper_prompt = PROMPT_TEMPLATES.get(
            method,
            """**角色：** 你是一名在氚燃料循环领域具有深厚背景的敏感性分析专家。

**任务：** 请仔细审查并解读以下这份由SALib库生成的敏感性分析报告。你的目标是：
1.  **总结核心发现**：简明扼要地总结报告中的关键信息。
2.  **识别关键参数**：对于报告中提到的每一个输出指标（如“启动氚量”、“倍增时间”等），明确指出哪些输入参数对它的影响最大（即最敏感）。
3.  **提供综合结论**：基于所有分析结果，对模型的整体行为、参数间的相互作用（如果可能）以及这些发现对工程实践的潜在启示，给出一个综合性的结论。

请确保你的分析清晰、专业，并直接切入要点。
""",
        )

        full_prompt = f"{wrapper_prompt}\n\n---\n**分析报告原文：**\n\n{report_content}"

        max_retries = 3
        for attempt in range(max_retries):
            try:
                client = openai.OpenAI(api_key=api_key, base_url=base_url)
                logger.info(
                    f"Sending SALib report to LLM (Attempt {attempt + 1}/{max_retries})..."
                )

                response = client.chat.completions.create(
                    model=ai_model,
                    messages=[{"role": "user", "content": full_prompt}],
                    max_tokens=4000,
                )
                llm_summary = response.choices[0].message.content

                logger.info("LLM analysis successful for SALib report.")
                return wrapper_prompt, llm_summary  # Return wrapper prompt and summary

            except Exception as e:
                logger.error(
                    f"Error calling LLM for SALib report on attempt {attempt + 1}: {e}"
                )
                if attempt < max_retries - 1:
                    time.sleep(5)
                else:
                    logger.error(
                        f"Failed to get LLM summary for SALib report after {max_retries} attempts."
                    )
                    return None, None  # Return None on failure

    except Exception as e:
        logger.error(f"Error in call_llm_for_salib_analysis: {e}", exc_info=True)
        return None, None

run_salib_analysis(config)

Orchestrates the SALib sensitivity analysis workflow.

This function extracts the necessary configuration, defines the problem space for SALib, and then runs the analysis.

Parameters:

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

Source code in tricys/analysis/salib.py

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

    This function extracts the necessary configuration, defines the problem
    space for SALib, and then runs the analysis.

    Args:
        config: The main configuration dictionary.
    """
    # 1. Extract sensitivity analysis configuration
    sa_config = config.get("sensitivity_analysis")
    if not sa_config or not sa_config.get("enabled"):
        logger.info("Sensitivity analysis is not enabled in the configuration file.")
        return

    # 2. Create analyzer
    analyzer = TricysSALibAnalyzer(config)

    # 3. Define the problem space from configuration
    analysis_case = sa_config.get("analysis_case", {})
    param_names = analysis_case.get("independent_variable")
    sampling_details = analysis_case.get("independent_variable_sampling")

    if not isinstance(param_names, list):
        raise ValueError("'independent_variable' must be a list of parameter names.")
    if not isinstance(sampling_details, dict):
        raise ValueError(
            "'independent_variable_sampling' must be an object with parameter details."
        )

    param_bounds = {
        name: sampling_details[name]["bounds"]
        for name in param_names
        if name in sampling_details
    }
    param_dists = {
        name: sampling_details[name].get("distribution", "unif")
        for name in param_names
        if name in sampling_details
    }

    if len(param_bounds) != len(param_names):
        raise ValueError(
            "The keys of 'independent_variable' and 'independent_variable_sampling' do not match"
        )

    problem = analyzer.define_problem(param_bounds, param_dists)
    logger.info(
        f"\n🔍 The problem space with {problem['num_vars']} parameters was defined from the configuration file"
    )

    # 4. Generate samples from configuration
    analyzer_config = analysis_case.get("analyzer", {})
    enabled_method_name = analyzer_config.get("method")
    if not enabled_method_name:
        raise ValueError(
            "No method found in 'sensitivity_analysis.analysis_case.analyzer'"
        )

    N = analyzer_config.get("sample_N", 1024)

    sample_kwargs = {}

    samples = analyzer.generate_samples(
        method=enabled_method_name, N=N, **sample_kwargs
    )
    logger.info(f"✓ Generated {len(samples)} parameter samples")

    # 5. Run Tricys simulation
    output_metrics = analysis_case.get("dependent_variables", [])

    csv_file_path = analyzer.run_tricys_simulations(output_metrics=output_metrics)
    logger.info(f"✓ Parameter file has been generated: {csv_file_path}")

    summary_file = None
    try:
        logger.info("\nAttempting to run Tricys analysis directly...")
        summary_file = analyzer.run_tricys_analysis(
            csv_file_path=csv_file_path, output_metrics=output_metrics
        )
        if summary_file:
            logger.info(f"✓ Tricys analysis completed, result file: {summary_file}")
        else:
            logger.info("⚠️  Tricys analysis result file not found")
            return
    except Exception as e:
        logger.info(f"⚠️  Tricys analysis failed: {e}")
        logger.info("Please check if the model path and configuration are correct")
        return

    # 6. Run SALib analysis from Tricys results
    try:
        logger.info("\nRunning SALib analysis from Tricys results...")
        all_results = analyzer.run_salib_analysis_from_tricys_results(
            sensitivity_summary_csv=summary_file,
            param_bounds=param_bounds,
            output_metrics=output_metrics,
            methods=[enabled_method_name],
            save_dir=os.path.dirname(summary_file),
        )

        logger.info(f"\n✅ SALib {enabled_method_name.upper()} analysis completed!")
        logger.info(
            f"📁 The results have been saved to: {os.path.join(os.path.dirname(summary_file), f'salib_analysis_{enabled_method_name}')}"
        )

        logger.info("\n📈 Brief results:")
        for metric_name, metric_results in all_results.items():
            logger.info(f"\n--- {metric_name} ---")
            if enabled_method_name in metric_results:
                result_data = metric_results[enabled_method_name]
                if enabled_method_name == "sobol":
                    logger.info("🔥 Most sensitive parameters (Sobol ST):")
                    st_values = list(zip(analyzer.problem["names"], result_data["ST"]))
                    st_values.sort(key=lambda x: x[1], reverse=True)
                    for param, st in st_values[:3]:
                        logger.info(f"   {param}: {st:.4f}")
                elif enabled_method_name == "morris":
                    logger.info("📊 Most Sensitive Parameter (Morris μ*):")
                    mu_star_values = list(
                        zip(analyzer.problem["names"], result_data["mu_star"])
                    )
                    mu_star_values.sort(key=lambda x: x[1], reverse=True)
                    for param, mu_star in mu_star_values[:3]:
                        logger.info(f"   {param}: {mu_star:.4f}")
                elif enabled_method_name == "fast":
                    logger.info("⚡ Most Sensitive Parameter (Morris μ*):")
                    st_values = list(zip(analyzer.problem["names"], result_data["ST"]))
                    st_values.sort(key=lambda x: x[1], reverse=True)
                    for param, st in st_values[:3]:
                        logger.info(f"   {param}: {st:.4f}")

        return analyzer, all_results

    except Exception as e:
        logger.error(f"SALib analysis failed: {e}", exc_info=True)
        raise