parallel_computing.py

# (c) Zizhe Wang
# https://zizhe.wang

######################
#                    #
# PARALLEL COMPUTING #
#                    #
######################

import os
import tempfile
import shutil
from time import sleep
from OMPython import OMCSessionZMQ
import numpy as np
from config import MODEL_FILE, MODEL_NAME, SIMULATION_STOP_TIME, RESULTS, MODEL_PATH, PRECISION

temp_dirs = []  # List to store paths of temporary directories

def optimization_function(param_values):
    """
    Function to optimize the Modelica model given specific parameter values.
    This function runs the simulation and retrieves the results.
    """
    temp_dir = tempfile.mkdtemp()  # Create a unique temporary directory for each worker
    temp_dirs.append(temp_dir)  # Store the path for later cleanup
    try:
        omc = OMCSessionZMQ()  # Create a new OpenModelica session
        temp_dir = temp_dir.replace('\\', '/')
        omc.sendExpression(f'cd("{temp_dir}")')

        # Copy model file to temporary directory
        temp_model_path = os.path.join(temp_dir, MODEL_FILE).replace('\\', '/')
        shutil.copy(MODEL_PATH, temp_model_path)

        # Load the model in the worker session
        load_temp_model_result = omc.sendExpression(f"loadFile(\"{temp_model_path}\")")
        print(f"Load model result in worker: {load_temp_model_result}")
        if load_temp_model_result is not True:
            messages = omc.sendExpression("getErrorString()")
            print(f"OpenModelica error messages: {messages}")
            raise RuntimeError(f"Failed to load model: {MODEL_NAME}")

        # Build the model in the worker session
        build_model_result = omc.sendExpression(f"buildModel({MODEL_NAME})")
        # uncomment this to print
        # print(f"Build model result in worker: {build_model_result}")
        if not (isinstance(build_model_result, tuple) and build_model_result[0]):
            messages = omc.sendExpression("getErrorString()")
            print(f"OpenModelica error messages: {messages}")
            raise RuntimeError(f"Failed to build model: {MODEL_NAME}")

        # Check if model files are created
        expected_files = [
            os.path.join(temp_dir, f"{MODEL_NAME}.exe").replace('\\', '/'),
            os.path.join(temp_dir, f"{MODEL_NAME}_init.xml").replace('\\', '/')
        ]
        for file in expected_files:
            if not os.path.isfile(file):
                raise RuntimeError(f"Expected file not found: {file}")
        
        # Set model parameters
        rounded_param_values = {param: round(value, PRECISION) for param, value in param_values.items()}
        for param, value in rounded_param_values.items():
            set_param_result = omc.sendExpression(f"setParameterValue({MODEL_NAME}, {param}, {value})")
            if not set_param_result:
                raise RuntimeError(f"Failed to set model parameter: {param} = {value}")
        print(f"Parameters set: {rounded_param_values}")
        
        # Simulate the model
        simulate_result = omc.sendExpression(f"simulate({MODEL_NAME}, stopTime={SIMULATION_STOP_TIME})")
        # uncomment this to print
        # print(f"Simulate result: {simulate_result}")
        # Retrieve simulation results
        result_values = {}
        for result in RESULTS:
            value_command = f"val({result}, {SIMULATION_STOP_TIME})"
            # uncomment this to print
            # print(f"Executing command: {value_command}")
            value = omc.sendExpression(value_command)
            print(f"Value for {result} at {SIMULATION_STOP_TIME}: {value}")
            # Round the result
            result_values[result] = round(value, PRECISION)
            if value is None:
                raise ValueError(f"Simulation result returned None for {result}")
            if not isinstance(value, (float, int)):
                raise ValueError(f"Simulation result for {result} is not in expected format (float or int)")
        
        print(f"Simulation results: {result_values}")
        return list(result_values.values())

    except Exception as e:
        print(f"Error during simulation: {e}")
        return [np.nan] * len(RESULTS)

# Cleanup temporary directories at the end of the script execution
def cleanup_temp_dirs():
    for temp_dir in temp_dirs:
        try:
            shutil.rmtree(temp_dir)
        except PermissionError:
            print(f"PermissionError: Retrying to remove {temp_dir}")
            sleep(2)
            try:
                shutil.rmtree(temp_dir)
            except PermissionError:
                print(f"Failed to remove {temp_dir} after multiple attempts")