Skip to content
Snippets Groups Projects
Commit c371105a authored by Zizhe Wang's avatar Zizhe Wang
Browse files

feat case study with orchestration

parent fa98308f
Branches
No related tags found
No related merge requests found
import json
import os
import numpy as np
import re
import shutil
import tempfile
from OMPython import OMCSessionZMQ
from edge_cloud_wrapper import MOO4ModelicaWrapper
# Configuration parameters
model_path = 'ITSystem.mo'
model_name = 'ITSystem'
simulation_time = 100
data_file = 'energy_available_and_user_demand_data.txt'
start_hour = 8
end_hour = 13
def init_omc_session(temp_dir):
omc = OMCSessionZMQ()
omc.sendExpression(f'cd("{temp_dir}")')
omc.sendExpression(f'loadFile("{model_path}")')
return omc
def build_model(omc):
build_result = omc.sendExpression(f'buildModel({model_name})')
if 'error' in build_result:
raise Exception(f"Model build failed: {build_result}")
def simulate_and_evaluate(parameters, simulation_time, available_energy, user_demand):
# Create a temporary directory to avoid GUID mismatch issues
temp_dir = tempfile.mkdtemp()
try:
omc = init_omc_session(temp_dir)
build_model(omc) # Ensure the model is built
if parameters is None:
raise ValueError("Parameters must not be None")
for param, value in parameters.items():
response = omc.sendExpression(f'setParameterValue({model_name}, {param}, {value})')
print(f"Set parameter {param} to {value}: {response}")
# Set the additional parameters availableEnergy and userDemand
omc.sendExpression(f'setParameterValue({model_name}, availableEnergy, {available_energy})')
omc.sendExpression(f'setParameterValue({model_name}, userDemand, {user_demand})')
result = omc.sendExpression(f'simulate({model_name}, stopTime={simulation_time})')
termination_message = result.get('messages', "")
print("Termination message:", termination_message)
match = re.search(r'Simulation call terminate\(\) at time ([\d\.]+)', termination_message)
if match:
depletion_time = float(match.group(1))
print(f"The IT system can run for {depletion_time} seconds until the remaining energy is depleted.")
else:
depletion_time = simulation_time
print(f"The IT system ran until the end of the simulation time ({simulation_time} seconds).")
performance = omc.sendExpression(f"val(performance, {depletion_time})")
print(f"Performance: {performance}")
remaining_energy = omc.sendExpression(f"val(remainingEnergy, {depletion_time})")
print(f"Remaining energy at time {depletion_time}: {remaining_energy}")
energy_consumption = omc.sendExpression(f"val(energyConsumption, {depletion_time})")
print(f"Energy Consumption: {energy_consumption}")
user_demand_satisfied = False
if performance and performance >= user_demand:
print("User demand satisfied.")
user_demand_satisfied = True
else:
print("User demand not satisfied.")
return remaining_energy, user_demand_satisfied, parameters, depletion_time
finally:
shutil.rmtree(temp_dir) # Clean up the temporary directory
def read_data_from_file(filename):
data = []
with open(filename, 'r') as file:
for line in file:
time_str, energy, demand = line.split()
hour = int(time_str.split(':')[0])
data.append((hour, int(energy), int(demand)))
return data
def adaptive_control_loop(data, moo_wrapper):
report = []
for hour, energy_available, user_demand in data:
if start_hour <= hour < end_hour:
print(f"\nProcessing hour: {hour}")
# Update the config for the optimization framework
moo_wrapper.update_config(energy_available, user_demand, simulation_time)
# Run the optimization using the MOO4Modelica wrapper
moo_wrapper.run_optimization()
# Get the list of best parameters from the optimization results
parameter_list = moo_wrapper.get_parameters()
# Try each parameter set in order until user demand is satisfied or options are exhausted
user_demand_satisfied = False
for best_parameters in parameter_list:
remaining_energy, user_demand_satisfied, parameters, depletion_time = simulate_and_evaluate(best_parameters, simulation_time, energy_available, user_demand)
if user_demand_satisfied:
break
# Generate the report for each hour
if user_demand_satisfied:
report.append(f"Hour {hour}: User demand satisfied with configuration {parameters}. Simulation ran for {depletion_time} seconds out of {simulation_time} seconds.")
else:
report.append(f"Hour {hour}: No sufficient configuration found. Simulation ran for {depletion_time} seconds out of {simulation_time} seconds.")
# Print the final report
print("\nFinal Report:")
for line in report:
print(line)
if __name__ == "__main__":
data = read_data_from_file(data_file)
moo_wrapper = MOO4ModelicaWrapper(config_path='config.json')
adaptive_control_loop(data, moo_wrapper)
\ No newline at end of file
import json
import os
from optimize_main import run_optimization
class MOO4ModelicaWrapper:
def __init__(self, config_path):
self.config_path = config_path
self.load_config()
self.model_path = self.config['MODEL_FILE']
self.model_name = self.config['MODEL_NAME']
def load_config(self):
with open(self.config_path, 'r') as f:
self.config = json.load(f)
def update_config(self, available_energy, user_demand, simulation_time):
# Modify the configuration with new parameters
self.config['PARAMETERS'] = ["activeCores", "cpuFrequency"]
self.config['PARAM_BOUNDS'] = {
"activeCores": {
"bounds": [1, 4],
"type": "int"
},
"cpuFrequency": {
"bounds": [1.0, 3.0],
"type": "float"
}
}
self.config['SIMULATION_STOP_TIME'] = simulation_time
self.config['available_energy'] = available_energy
self.config['user_demand'] = user_demand
# Save the updated configuration to a file
with open(self.config_path, 'w') as f:
json.dump(self.config, f, indent=4)
def run_optimization(self):
run_optimization()
def get_parameters(self):
# Assuming the results are saved to a file "optimization_results.json"
results_path = os.path.join(os.path.dirname(self.config_path), 'results', 'optimization_results.json')
with open(results_path, 'r') as f:
optimization_results = json.load(f)
# The parameters are returned as they are in the results
parameters = optimization_results["parameters"]
return [
{
"activeCores": int(params[0]),
"cpuFrequency": float(params[1])
}
for params in parameters
]
\ No newline at end of file
import json
import os
from optimize_main import run_optimization
from edge_cloud_wrapper import MOO4ModelicaWrapper
from edge_cloud_configurator import adaptive_control_loop, read_data_from_file
base_path = os.path.dirname(__file__)
data_file_path = os.path.join(base_path, 'energy_available_and_user_demand_data.txt')
config_path = os.path.join(base_path, 'config.json')
def main():
data = read_data_from_file(data_file_path)
moo_wrapper = MOO4ModelicaWrapper(config_path=config_path)
adaptive_control_loop(data, moo_wrapper)
if __name__ == "__main__":
main()
\ No newline at end of file
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment