API 参考 - 核心模块 (Core)

核心模块 (Core)

核心模块 (Core) 包含了 TRICYS 的核心功能，如作业管理、Modelica 交互和事件拦截器。 请在下方的标签页中选择您感兴趣的特定模块。

Jobs (作业管理)Modelica (模型交互)Interceptor (拦截器)

generate_simulation_jobs(simulation_params)

Generates a list of simulation jobs from parameters, handling sweeps and array expansion.

Parameters:

Name	Type	Description	Default
simulation_params	Dict[str, Any]	Dictionary of simulation parameters. Can include: - "file": Path to CSV file for batch job loading - Parameter names with values (single values, lists, or special format strings) - Array-like parameters in format "{value1, value2, ...}"	required

Returns:

Type	Description
List[Dict[str, Any]]	A list of job dictionaries. Each job contains one combination of parameter
List[Dict[str, Any]]	values for simulation. Single-value parameters are included in all jobs.

Note

If "file" parameter is present, jobs are loaded from CSV and other parameters are merged into each CSV job. For parameter sweeps, generates all combinations using Cartesian product. Array-like parameters are expanded using 1-based indexing before sweep generation. Returns empty dict list [{}] if no parameters provided.

Source code in tricys/core/jobs.py

def generate_simulation_jobs(
    simulation_params: Dict[str, Any],
) -> List[Dict[str, Any]]:
    """Generates a list of simulation jobs from parameters, handling sweeps and array expansion.

    Args:
        simulation_params: Dictionary of simulation parameters. Can include:
            - "file": Path to CSV file for batch job loading
            - Parameter names with values (single values, lists, or special format strings)
            - Array-like parameters in format "{value1, value2, ...}"

    Returns:
        A list of job dictionaries. Each job contains one combination of parameter
        values for simulation. Single-value parameters are included in all jobs.

    Note:
        If "file" parameter is present, jobs are loaded from CSV and other parameters
        are merged into each CSV job. For parameter sweeps, generates all combinations
        using Cartesian product. Array-like parameters are expanded using 1-based indexing
        before sweep generation. Returns empty dict list [{}] if no parameters provided.
    """

    logger.info(
        "Generating simulation jobs",
        extra={
            "simulation_parameters": simulation_params,
        },
    )
    if "file" in simulation_params:
        file_value = simulation_params["file"]
        if isinstance(file_value, str):
            csv_jobs = _load_jobs_from_csv(file_value)

            other_params = {k: v for k, v in simulation_params.items() if k != "file"}
            for job in csv_jobs:
                job.update(other_params)

            return csv_jobs

    # First, expand any array-like parameters before processing.
    processed_params = _expand_array_parameters(simulation_params)

    sweep_params = {}
    single_value_params = {}

    for name, value in processed_params.items():
        parsed_values = parse_parameter_value(value)
        if len(parsed_values) > 1:
            sweep_params[name] = parsed_values
        else:
            single_value_params[name] = parsed_values[0] if parsed_values else None

    if not sweep_params:
        return [single_value_params] if single_value_params else [{}]

    sweep_names = list(sweep_params.keys())
    sweep_values = list(sweep_params.values())
    jobs = []
    for combo in itertools.product(*sweep_values):
        job = single_value_params.copy()
        job.update(dict(zip(sweep_names, combo)))
        jobs.append(job)
    return jobs

parse_parameter_value(value)

Parses a parameter value which can be a single value, a list, or a string with special formats.

Supported special string formats

• "start:stop:step" -> e.g., "1:10:2" for a linear range with step
• "linspace:start:stop:num" -> e.g., "linspace:0:10:5" for 5 evenly spaced points
• "log:start:stop:num" -> e.g., "log:1:1000:4" for 4 points on log scale
• "rand:min:max:count" -> e.g., "rand:0:1:10" for 10 random uniform values
• "file:path:column" -> e.g., "file:data.csv:voltage" to read a CSV column
• "file:path" -> e.g., "file:sampling.csv" to read all parameters from CSV

Parameters:

Name	Type	Description	Default
value	Any	The parameter value to parse. Can be a single value, list, or special format string.	required

Returns:

Type	Description
List[Any]	A list of parsed values. Single values and strings without colons return
List[Any]	a single-element list. Special format strings return expanded value lists.

Note

Numbers are rounded to 8 decimal places to avoid floating point precision issues. If parsing fails, returns the original value as a single-element list and logs an error. For "file:" format, Windows paths with drive letters (C:) are handled.

Source code in tricys/core/jobs.py

def parse_parameter_value(value: Any) -> List[Any]:
    """Parses a parameter value which can be a single value, a list, or a string with special formats.

    Supported special string formats:
        - "start:stop:step" -> e.g., "1:10:2" for a linear range with step
        - "linspace:start:stop:num" -> e.g., "linspace:0:10:5" for 5 evenly spaced points
        - "log:start:stop:num" -> e.g., "log:1:1000:4" for 4 points on log scale
        - "rand:min:max:count" -> e.g., "rand:0:1:10" for 10 random uniform values
        - "file:path:column" -> e.g., "file:data.csv:voltage" to read a CSV column
        - "file:path" -> e.g., "file:sampling.csv" to read all parameters from CSV

    Args:
        value: The parameter value to parse. Can be a single value, list, or special
            format string.

    Returns:
        A list of parsed values. Single values and strings without colons return
        a single-element list. Special format strings return expanded value lists.

    Note:
        Numbers are rounded to 8 decimal places to avoid floating point precision issues.
        If parsing fails, returns the original value as a single-element list and logs
        an error. For "file:" format, Windows paths with drive letters (C:\\) are handled.
    """
    if not isinstance(value, str):
        return value if isinstance(value, list) else [value]

    if ":" not in value:
        return [value]  # Just a plain string

    try:
        prefix, args_str = value.split(":", 1)
        prefix = prefix.lower()

        if prefix == "linspace":
            start, stop, num = map(float, args_str.split(":"))
            return np.linspace(start, stop, int(num)).round(8).tolist()

        if prefix == "log":
            start, stop, num = map(float, args_str.split(":"))
            if start <= 0 or stop <= 0:
                raise ValueError("Log scale start and stop values must be positive.")
            return (
                np.logspace(np.log10(start), np.log10(stop), int(num)).round(8).tolist()
            )

        if prefix == "rand":
            low, high, count = map(float, args_str.split(":"))
            return np.random.uniform(low, high, int(count)).round(8).tolist()

        if prefix == "file":
            # Handle file paths that may contain colons (e.g., Windows C:\...)
            try:
                parts = args_str.rsplit(":", 1)
                # A column is present if there are two parts and the second part does not contain path separators.
                if len(parts) == 2 and not any(c in parts[1] for c in "/\\"):
                    file_path, column_name = parts
                    if not os.path.isabs(file_path.strip()):
                        abs_file_path = os.path.abspath(
                            os.path.join(os.getcwd(), file_path.strip())
                        )
                    else:
                        abs_file_path = file_path.strip()
                    df = pd.read_csv(abs_file_path)
                    return df[column_name.strip()].tolist()
                else:
                    # No column name or it's part of the path
                    file_path = args_str
                    # Return the file path for later processing
                    return [file_path.strip()]
            except (ValueError, FileNotFoundError, KeyError):
                # Re-raise to be caught by the outer try-except block
                raise

        # Fallback to original start:stop:step logic if no prefix matches
        start, stop, step = map(float, value.split(":"))
        return np.arange(start, stop + step / 2, step).round(8).tolist()

    except (ValueError, FileNotFoundError, KeyError, IndexError) as e:
        logger.error(
            "Invalid format or error processing parameter value",
            extra={
                "value": value,
                "error": str(e),
            },
        )
        return [value]  # On any error, treat as a single literal value

Utilities for interacting with OpenModelica via OMPython.

This module provides a set of functions to manage an OpenModelica session, load models, retrieve parameter details, and format parameter values for simulation.

format_parameter_value(name, value)

Formats a parameter value into a string recognized by OpenModelica.

Parameters:

Name	Type	Description	Default
name	str	The name of the parameter.	required
value	Any	The value of the parameter (can be number, string, list, or bool).	required

Returns:

Type	Description
str	A formatted string for use in simulation overrides (e.g., "p=1.0",
str	"name={1,2,3}", or 'path="value"').

Note

Lists are formatted as {v1,v2,...}. Strings are quoted with double quotes. Numbers and booleans use direct string conversion.

Source code in tricys/core/modelica.py

def format_parameter_value(name: str, value: Any) -> str:
    """Formats a parameter value into a string recognized by OpenModelica.

    Args:
        name: The name of the parameter.
        value: The value of the parameter (can be number, string, list, or bool).

    Returns:
        A formatted string for use in simulation overrides (e.g., "p=1.0",
        "name={1,2,3}", or 'path="value"').

    Note:
        Lists are formatted as {v1,v2,...}. Strings are quoted with double quotes.
        Numbers and booleans use direct string conversion.
    """
    if isinstance(value, list):
        # In Modelica, strings in records should be quoted.
        # This regex checks if the string is already quoted.
        def format_element(elem):
            if isinstance(elem, str):
                if re.match(r'^".*"$', elem):
                    return elem
                else:
                    return f'"{elem}"'
            return str(elem)

        return f"{name}={{{','.join(map(format_element, value))}}}"
    elif isinstance(value, bool):
        return f"{name}={str(value).lower()}"
    elif isinstance(value, str):
        # Format strings as "value"
        return f'{name}="{value}"'
    # For numbers, direct string conversion is fine
    return f"{name}={value}"

get_all_parameters_details(omc, model_name)

Recursively retrieves detailed information for all parameters in a given model.

Parameters:

Name	Type	Description	Default
omc	OMCSessionZMQ	The active OpenModelica session object.	required
model_name	str	The full name of the model.	required

Returns:

Type	Description
List[Dict[str, Any]]	A list of dictionaries, where each dictionary contains the detailed
List[Dict[str, Any]]	information of a single parameter including name, type, defaultValue,
List[Dict[str, Any]]	comment, and dimensions.

Note

Uses _recursive_get_parameters() to traverse the model hierarchy. Returns empty list if model is not found or on error. Each parameter dict includes 'name' (hierarchical), 'type', 'defaultValue', 'comment', and 'dimensions' fields.

Source code in tricys/core/modelica.py

def get_all_parameters_details(
    omc: OMCSessionZMQ, model_name: str
) -> List[Dict[str, Any]]:
    """Recursively retrieves detailed information for all parameters in a given model.

    Args:
        omc: The active OpenModelica session object.
        model_name: The full name of the model.

    Returns:
        A list of dictionaries, where each dictionary contains the detailed
        information of a single parameter including name, type, defaultValue,
        comment, and dimensions.

    Note:
        Uses _recursive_get_parameters() to traverse the model hierarchy.
        Returns empty list if model is not found or on error. Each parameter
        dict includes 'name' (hierarchical), 'type', 'defaultValue', 'comment',
        and 'dimensions' fields.
    """
    logger.info(
        "Getting detailed parameters via recursion", extra={"model_name": model_name}
    )
    all_params_details = []
    try:
        if not omc.sendExpression(f"isModel({model_name})"):
            logger.error("Model not found in package", extra={"model_name": model_name})
            return []
        _recursive_get_parameters(omc, model_name, "", all_params_details)
        logger.info(
            "Successfully found parameter details",
            extra={"count": len(all_params_details)},
        )
        return all_params_details
    except Exception as e:
        logger.error(
            "Failed to get detailed parameters via recursion",
            exc_info=True,
            extra={"error": str(e)},
        )
        return []

get_model_default_parameters(omc, model_name)

Retrieves the default values for all parameters in a given model.

This function leverages get_all_parameters_details to fetch detailed parameter information and then extracts and parses the name and default value into a dictionary.

Parameters:

Name	Type	Description	Default
omc	OMCSessionZMQ	The active OpenModelica session object.	required
model_name	str	The full name of the model.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary mapping parameter names to their default values
Dict[str, Any]	(e.g., float, list, bool, str). Returns an empty dictionary if
Dict[str, Any]	the model is not found or has no parameters.

Note

Values are parsed from OpenModelica string format to Python types using _parse_om_value(). Handles arrays, booleans, strings, and numeric values.

Source code in tricys/core/modelica.py

def get_model_default_parameters(omc: OMCSessionZMQ, model_name: str) -> Dict[str, Any]:
    """Retrieves the default values for all parameters in a given model.

    This function leverages get_all_parameters_details to fetch detailed
    parameter information and then extracts and parses the name and default value
    into a dictionary.

    Args:
        omc: The active OpenModelica session object.
        model_name: The full name of the model.

    Returns:
        A dictionary mapping parameter names to their default values
        (e.g., float, list, bool, str). Returns an empty dictionary if
        the model is not found or has no parameters.

    Note:
        Values are parsed from OpenModelica string format to Python types using
        _parse_om_value(). Handles arrays, booleans, strings, and numeric values.
    """
    logger.info(
        "Getting and parsing default parameter values", extra={"model_name": model_name}
    )

    # Use the existing detailed function to get all parameter info
    all_params_details = get_all_parameters_details(omc, model_name)

    if not all_params_details:
        logger.warning(
            "No parameters found for model", extra={"model_name": model_name}
        )
        return {}

    # Convert the list of dicts into a single dict of name: parsed_defaultValue
    default_params = {
        param["name"]: _parse_om_value(param["defaultValue"])
        for param in all_params_details
    }

    logger.info(
        "Found and parsed default parameters",
        extra={
            "count": len(default_params),
            "model_name": model_name,
        },
    )
    return default_params

get_model_parameter_names(omc, model_name)

Parses and returns all subcomponent parameter names for a given model.

Parameters:

Name	Type	Description	Default
omc	OMCSessionZMQ	The active OpenModelica session object.	required
model_name	str	The full name of the model (e.g., 'example.Cycle').	required

Returns:

Type	Description
List[str]	A list of all available parameter names in hierarchical format
List[str]	(e.g., ['blanket.TBR', 'divertor.heatLoad']).

Note

Only traverses components whose type starts with the package name. Returns empty list if model is not found or has no components. Uses getComponents() and getParameterNames() OMC API calls.

Source code in tricys/core/modelica.py

def get_model_parameter_names(omc: OMCSessionZMQ, model_name: str) -> List[str]:
    """Parses and returns all subcomponent parameter names for a given model.

    Args:
        omc: The active OpenModelica session object.
        model_name: The full name of the model (e.g., 'example.Cycle').

    Returns:
        A list of all available parameter names in hierarchical format
        (e.g., ['blanket.TBR', 'divertor.heatLoad']).

    Note:
        Only traverses components whose type starts with the package name.
        Returns empty list if model is not found or has no components. Uses
        getComponents() and getParameterNames() OMC API calls.
    """
    logger.info("Getting parameter names for model", extra={"model_name": model_name})
    all_params = []
    try:
        if not omc.sendExpression(f"isModel({model_name})"):
            logger.warning(
                "Model not found in package", extra={"model_name": model_name}
            )
            return []

        components = omc.sendExpression(f"getComponents({model_name})")
        if not components:
            logger.warning(
                "No components found for model", extra={"model_name": model_name}
            )
            return []

        for comp in components:
            comp_type, comp_name = comp[0], comp[1]
            if comp_type.startswith(model_name.split(".")[0]):
                params = omc.sendExpression(f"getParameterNames({comp_type})")
                for param in params:
                    full_param = f"{comp_name}.{param}"
                    if full_param not in all_params:
                        all_params.append(full_param)

        logger.info("Found parameter names", extra={"count": len(all_params)})
        return all_params

    except Exception as e:
        logger.error(
            "Failed to get parameter names", exc_info=True, extra={"error": str(e)}
        )
        return []

get_om_session()

Initializes and returns a new OMCSessionZMQ session.

Returns:

Type	Description
OMCSessionZMQ	An active OpenModelica session object.

Note

Creates a new ZMQ-based connection to OpenModelica Compiler. Each call creates an independent session that should be properly closed after use.

Source code in tricys/core/modelica.py

def get_om_session() -> OMCSessionZMQ:
    """Initializes and returns a new OMCSessionZMQ session.

    Returns:
        An active OpenModelica session object.

    Note:
        Creates a new ZMQ-based connection to OpenModelica Compiler. Each call
        creates an independent session that should be properly closed after use.
    """
    logger.debug("Initializing new OMCSessionZMQ session")
    return OMCSessionZMQ()

load_modelica_package(omc, package_path)

Loads a Modelica package into the OpenModelica session.

Parameters:

Name	Type	Description	Default
omc	OMCSessionZMQ	The active OpenModelica session object.	required
package_path	str	The file path to the Modelica package (package.mo).	required

Returns:

Type	Description
bool	True if the package was loaded successfully, False otherwise.

Note

Uses sendExpression('loadFile(...)') command. Logs error if loading fails. The package must be a valid Modelica package file.

Source code in tricys/core/modelica.py

def load_modelica_package(omc: OMCSessionZMQ, package_path: str) -> bool:
    """Loads a Modelica package into the OpenModelica session.

    Args:
        omc: The active OpenModelica session object.
        package_path: The file path to the Modelica package (`package.mo`).

    Returns:
        True if the package was loaded successfully, False otherwise.

    Note:
        Uses sendExpression('loadFile(...)') command. Logs error if loading fails.
        The package must be a valid Modelica package file.
    """
    logger.info("Loading package", extra={"package_path": package_path})
    load_result = omc.sendExpression(f'loadFile("{package_path}")')
    if not load_result:
        logger.error("Failed to load package", extra={"package_path": package_path})
        return False
    return True

integrate_interceptor_model(package_path, model_name, interception_configs)

Integrates CSV data replacement into a system model.

This function supports two modes (all handlers must use the same mode): 1. "interceptor" (default): Creates interceptor models between submodels and system. 2. "replacement": Directly modifies submodels to use CSV data.

Parameters:

Name	Type	Description	Default
package_path	str	The file path to the Modelica package. For multi-file packages, this should be the path to package.mo. For single-file packages, this should be the path to the .mo file containing the package.	required
model_name	str	The full name of the system model to be modified.	required
interception_configs	list[Dict[str, Any]]	A list of dictionaries, each defining an interception task. All configs must have the same 'mode' field. Each dict should contain 'submodel_name', 'csv_uri', 'instance_name', 'output_placeholder', and optionally 'mode' (defaults to 'interceptor').	required

Returns:

Type	Description
Dict[str, Any]	A dictionary containing the paths to modified models. Structure varies by mode: - interceptor mode: interceptor_model_paths, system_model_path - replacement mode: replaced_models, system_model_path

Raises:

Type	Description
ValueError	If interception_configs is empty or if mixed modes are detected.
FileNotFoundError	If package_path is invalid or package.mo not found.

Note

Mode is determined from the first config's 'mode' field. All configs must use the same mode or ValueError is raised. Automatically detects single-file vs multi-file package structure and routes to appropriate handler.

Source code in tricys/core/interceptor.py

def integrate_interceptor_model(
    package_path: str, model_name: str, interception_configs: list[Dict[str, Any]]
) -> Dict[str, Any]:
    """Integrates CSV data replacement into a system model.

    This function supports two modes (all handlers must use the same mode):
    1. "interceptor" (default): Creates interceptor models between submodels and system.
    2. "replacement": Directly modifies submodels to use CSV data.

    Args:
        package_path: The file path to the Modelica package. For multi-file packages,
            this should be the path to `package.mo`. For single-file packages,
            this should be the path to the `.mo` file containing the package.
        model_name: The full name of the system model to be modified.
        interception_configs: A list of dictionaries, each defining an interception task.
            All configs must have the same 'mode' field. Each dict should contain
            'submodel_name', 'csv_uri', 'instance_name', 'output_placeholder', and
            optionally 'mode' (defaults to 'interceptor').

    Returns:
        A dictionary containing the paths to modified models. Structure varies by mode:
            - interceptor mode: interceptor_model_paths, system_model_path
            - replacement mode: replaced_models, system_model_path

    Raises:
        ValueError: If interception_configs is empty or if mixed modes are detected.
        FileNotFoundError: If package_path is invalid or package.mo not found.

    Note:
        Mode is determined from the first config's 'mode' field. All configs must use
        the same mode or ValueError is raised. Automatically detects single-file vs
        multi-file package structure and routes to appropriate handler.
    """
    if not interception_configs:
        raise ValueError("interception_configs cannot be empty")

    # Get mode from first config (all should be the same)
    mode = interception_configs[0].get("mode", "interceptor")

    # Validate that all configs use the same mode
    for config in interception_configs:
        config_mode = config.get("mode", "interceptor")
        if config_mode != mode:
            raise ValueError(
                f"Mixed modes are not supported. All handlers must use the same mode. "
                f"Expected '{mode}', but found '{config_mode}' in config for '{config.get('submodel_name')}'"
            )

    logger.info(
        "Integrating CSV data replacement",
        extra={
            "mode": mode,
            "num_submodels": len(interception_configs),
        },
    )

    # Route to appropriate handler based on mode
    if mode == "replacement":
        return _integrate_replacement(package_path, model_name, interception_configs)
    else:  # mode == "interceptor"
        # Determine package type and route to appropriate handler
        if os.path.isdir(package_path):
            package_file = os.path.join(package_path, "package.mo")
            if os.path.exists(package_file):
                return _integrate_interceptor_multi_file(
                    package_file, model_name, interception_configs
                )
            else:
                raise FileNotFoundError(
                    f"No package.mo found in directory: {package_path}"
                )
        elif os.path.isfile(package_path) and package_path.endswith("package.mo"):
            return _integrate_interceptor_multi_file(
                package_path, model_name, interception_configs
            )
        elif os.path.isfile(package_path):
            return _integrate_interceptor_single_file(
                package_path, model_name, interception_configs
            )
        else:
            raise FileNotFoundError(f"Invalid package path: {package_path}")

replace_submodels_with_csv(package_path, replacement_configs)

Replaces multiple submodels with CSV data sources.

Parameters:

Name	Type	Description	Default
package_path	str	Path to the Modelica package directory or package.mo file.	required
replacement_configs	list[Dict[str, Any]]	A list of dictionaries, each defining a replacement task: - submodel_name: Full name of the submodel (e.g., 'MyPackage.MyModel') - output_ports: List of output port definitions - csv_file: Path to the CSV file	required

Returns:

Type	Description
Dict[str, Any]	A dictionary containing: - replaced_models: List of results from each replacement - package_path: Original package path

Note

Automatically determines if package is directory-based or single-file. Continues processing remaining models if one fails, but re-raises the error after logging. Each submodel file is identified by splitting the full name and locating {ModelName}.mo in the package directory.

Source code in tricys/core/interceptor.py

def replace_submodels_with_csv(
    package_path: str,
    replacement_configs: list[Dict[str, Any]],
) -> Dict[str, Any]:
    """Replaces multiple submodels with CSV data sources.

    Args:
        package_path: Path to the Modelica package directory or package.mo file.
        replacement_configs: A list of dictionaries, each defining a replacement task:
            - submodel_name: Full name of the submodel (e.g., 'MyPackage.MyModel')
            - output_ports: List of output port definitions
            - csv_file: Path to the CSV file

    Returns:
        A dictionary containing:
            - replaced_models: List of results from each replacement
            - package_path: Original package path

    Note:
        Automatically determines if package is directory-based or single-file.
        Continues processing remaining models if one fails, but re-raises the error
        after logging. Each submodel file is identified by splitting the full name
        and locating {ModelName}.mo in the package directory.
    """
    logger.info(
        "Starting batch submodel replacement",
        extra={"num_configs": len(replacement_configs)},
    )

    # Determine package directory
    if os.path.isdir(package_path):
        package_dir = package_path
    elif os.path.isfile(package_path) and package_path.endswith("package.mo"):
        package_dir = os.path.dirname(package_path)
    elif os.path.isfile(package_path):
        package_dir = os.path.dirname(package_path)
    else:
        raise FileNotFoundError(f"Invalid package path: {package_path}")

    replaced_models = []

    for config in replacement_configs:
        submodel_name = config["submodel_name"]
        output_ports = config["output_ports"]
        csv_file = config["csv_file"]

        # Construct submodel file path
        model_simple_name = submodel_name.split(".")[-1]
        submodel_file = os.path.join(package_dir, f"{model_simple_name}.mo")

        if not os.path.exists(submodel_file):
            logger.warning(
                "Submodel file not found, skipping",
                extra={"submodel_name": submodel_name, "expected_path": submodel_file},
            )
            continue

        try:
            result = _replace_submodel_with_csv(
                submodel_path=submodel_file,
                output_ports=output_ports,
                csv_file=csv_file,
            )
            result["submodel_name"] = submodel_name
            replaced_models.append(result)

            logger.info(
                "Successfully replaced submodel",
                extra={"submodel_name": submodel_name},
            )

        except Exception as e:
            logger.error(
                "Failed to replace submodel",
                extra={"submodel_name": submodel_name, "error": str(e)},
            )
            raise

    logger.info(
        "Batch replacement completed",
        extra={"num_replaced": len(replaced_models)},
    )

    return {
        "replaced_models": replaced_models,
        "package_path": package_path,
    }