OiO.lk Community platform!

Oio.lk is an excellent forum for developers, providing a wide range of resources, discussions, and support for those in the developer community. Join oio.lk today to connect with like-minded professionals, share insights, and stay updated on the latest trends and technologies in the development field.
  You need to log in or register to access the solved answers to this problem.
  • You have reached the maximum number of guest views allowed
  • Please register below to remove this limitation

Building a Aspen Plus and Python interface via a xls file?

  • Thread starter Thread starter user25664963
  • Start date Start date
U

user25664963

Guest
I hope you can help me with this problem: I am trying to implement a custom Python ML-Model to predict the performance of a chemical separation process. A standard software to use in these applications is Aspen Plus wich is already equipped with a linkage to an old excel .xls file to transfer input data to an excel file, where it gets calculated by the user and send back to the simulation program. I would Python to get into this pipeline and import the incoming input values, process them in a ML script and export the output in the corresponding output cells in the xls file. I already tried some things with the xlrd library and the

conncetion between Aspen-Excel and Excel-Python is already working well. The problem is that when I run all three together they wont exchange data or chrash or prevent the script from writing something into the xls file. Does someone has experience in accessing one excel file by two scripts in parallel or has some ideas?

Best regards Goupus

Python Observer code to detect changes in the excel, send values to predict and implement in excel

Code:
import os
import time
import logging
import pandas as pd
import psutil
from threading import Thread, Event
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import shutil
import xlrd
import xlwt
from xlutils.copy import copy

class ExcelChangeHandler(FileSystemEventHandler):
    def __init__(self, file_path, interval=1):  # Set interval to 1 second
        self.file_path = file_path
        self.interval = interval
        self.stop_event = Event()
        try:
            self.previous_values = self.read_excel_cells(file_path)
            print(f"Initial values in B2 to B7: {self.previous_values}")
        except Exception as e:
            logging.error(f"Error initializing ExcelChangeHandler: {e}")
            self.previous_values = [None] * 6
        self.last_modified = os.path.getmtime(file_path)

    def read_excel_cells(self, file_path):
        try:
            workbook = xlrd.open_workbook(file_path)
            sheet = workbook.sheet_by_name('Aspen_Input')
            values = [sheet.cell_value(i, 1) for i in range(1, 7)]  # B2 to B7 are at column index 1
            logging.info(f"Read values from B2 to B7: {values}")
            print(f"Read values from B2 to B7: {values}")
            return values
        except Exception as e:
            logging.error(f"Error reading Excel file: {e}")
            raise

    def start(self):
        self.thread = Thread(target=self.run)
        self.thread.start()

    def stop(self):
        self.stop_event.set()
        self.thread.join()

    def run(self):
        while not self.stop_event.is_set():
            try:
                current_modified = os.path.getmtime(self.file_path)
                if current_modified != self.last_modified:
                    self.last_modified = current_modified
                    self.check_for_changes()
            except Exception as e:
                logging.error(f"Error monitoring Excel file: {e}")
            time.sleep(self.interval)

    def check_for_changes(self):
        try:
            current_values = self.read_excel_cells(self.file_path)
            if current_values != self.previous_values:
                for i, (prev, curr) in enumerate(zip(self.previous_values, current_values)):
                    if prev != curr:
                        logging.info(f"Change detected in B{i+2}: {prev} -> {curr}")
                        print(f"Change detected in B{i+2}: {prev} -> {curr}")
                        self.export_value(curr, f"B{i+2}")
                        result = self.run_calculation_script()
                        if result is not None:
                            self.write_result_to_excel(result, f"C{i+2}")
                self.previous_values = current_values
        except Exception as e:
            logging.error(f"Error reading Excel file: {e}")

    def export_value(self, value, cell):
        with open('value_to_process.txt', 'w') as f:
            f.write(str(value))
        logging.info(f"Exported value from {cell}: {value}")
        print(f"Exported value from {cell}: {value}")

    def run_calculation_script(self):
        try:
            result = os.system('python PredictV4.py')
            logging.info(f"Ran PredictV4.py with result: {result}")
            print(f"Ran PredictV4.py with result: {result}")
            with open('processed_value.txt', 'r') as f:
                result = f.read().strip()
            logging.info(f"Read result: {result}")
            print(f"Read result: {result}")
            # Clean up temporary files
            os.remove('value_to_process.txt')
            os.remove('processed_value.txt')
            return result
        except FileNotFoundError:
            logging.error("Error: processed_value.txt not found")
            print("Error: processed_value.txt not found")
            return None

    def write_result_to_excel(self, result, cell):
        try:
            workbook = xlrd.open_workbook(self.file_path, formatting_info=True)
            writable_workbook = copy(workbook)
            writable_sheet = writable_workbook.get_sheet('Aspen_Input')

            row_idx = int(cell[1:]) - 1  # Convert cell to row index (e.g., B2 -> 1)
            col_idx = ord(cell[0].upper()) - ord('A')  # Convert cell to column index (e.g., B -> 1, C -> 2)

            writable_sheet.write(row_idx, col_idx, result)  # Write the result to the specified cell

            writable_workbook.save(self.file_path)
            logging.info(f"Wrote result to Excel in {cell}: {result}")
            print(f"Wrote result to Excel in {cell}: {result}")
        except PermissionError as e:
            logging.error(f"Permission denied: {e}")
            print(f"Permission denied: {e}")
        except Exception as e:
            logging.error(f"Error writing to Excel file: {e}")
            print(f"Error writing to Excel file: {e}")

def monitor_cpu(interval=1):
    try:
        while True:
            cpu_usage = psutil.cpu_percent(interval=1)
            print(f"CPU Usage: {cpu_usage}%")
            time.sleep(interval)
    except KeyboardInterrupt:
        print("CPU monitoring stopped.")

if __name__ == "__main__":
    logging.basicConfig(filename="excel_changes.log", level=logging.INFO,
                        format="%(asctime)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S")

    path_to_monitor = r"C:\Users\germa\Desktop\Simulation_Test\ObserverV43.xls"  # Use raw string
    interval = 1  # Polling interval in seconds

    handler = ExcelChangeHandler(path_to_monitor, interval)
    handler.start()

    # Start CPU monitoring in a separate thread
    cpu_thread = Thread(target=monitor_cpu, daemon=True)
    cpu_thread.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        handler.stop()

Running in parallel will block Aspen Plus from accessing the xls file
<p>I hope you can help me with this problem: I am trying to implement a custom Python ML-Model to predict the performance of a chemical separation process. A standard software to use in these applications is Aspen Plus wich is already equipped with a linkage to an old excel .xls file to transfer input data to an excel file, where it gets calculated by the user and send back to the simulation program. I would Python to get into this pipeline and import the incoming input values, process them in a ML script and export the output in the corresponding output cells in the xls file. I already tried some things with the xlrd library and the</p>
<p>conncetion between Aspen-Excel and Excel-Python is already working well. The problem is that when I run all three together they wont exchange data or chrash or prevent the script from writing something into the xls file. Does someone has experience in accessing one excel file by two scripts in parallel or has some ideas?</p>
<p>Best regards
Goupus</p>
<p>Python Observer code to detect changes in the excel, send values to predict and implement in excel</p>
<pre><code>import os
import time
import logging
import pandas as pd
import psutil
from threading import Thread, Event
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import shutil
import xlrd
import xlwt
from xlutils.copy import copy

class ExcelChangeHandler(FileSystemEventHandler):
def __init__(self, file_path, interval=1): # Set interval to 1 second
self.file_path = file_path
self.interval = interval
self.stop_event = Event()
try:
self.previous_values = self.read_excel_cells(file_path)
print(f"Initial values in B2 to B7: {self.previous_values}")
except Exception as e:
logging.error(f"Error initializing ExcelChangeHandler: {e}")
self.previous_values = [None] * 6
self.last_modified = os.path.getmtime(file_path)

def read_excel_cells(self, file_path):
try:
workbook = xlrd.open_workbook(file_path)
sheet = workbook.sheet_by_name('Aspen_Input')
values = [sheet.cell_value(i, 1) for i in range(1, 7)] # B2 to B7 are at column index 1
logging.info(f"Read values from B2 to B7: {values}")
print(f"Read values from B2 to B7: {values}")
return values
except Exception as e:
logging.error(f"Error reading Excel file: {e}")
raise

def start(self):
self.thread = Thread(target=self.run)
self.thread.start()

def stop(self):
self.stop_event.set()
self.thread.join()

def run(self):
while not self.stop_event.is_set():
try:
current_modified = os.path.getmtime(self.file_path)
if current_modified != self.last_modified:
self.last_modified = current_modified
self.check_for_changes()
except Exception as e:
logging.error(f"Error monitoring Excel file: {e}")
time.sleep(self.interval)

def check_for_changes(self):
try:
current_values = self.read_excel_cells(self.file_path)
if current_values != self.previous_values:
for i, (prev, curr) in enumerate(zip(self.previous_values, current_values)):
if prev != curr:
logging.info(f"Change detected in B{i+2}: {prev} -> {curr}")
print(f"Change detected in B{i+2}: {prev} -> {curr}")
self.export_value(curr, f"B{i+2}")
result = self.run_calculation_script()
if result is not None:
self.write_result_to_excel(result, f"C{i+2}")
self.previous_values = current_values
except Exception as e:
logging.error(f"Error reading Excel file: {e}")

def export_value(self, value, cell):
with open('value_to_process.txt', 'w') as f:
f.write(str(value))
logging.info(f"Exported value from {cell}: {value}")
print(f"Exported value from {cell}: {value}")

def run_calculation_script(self):
try:
result = os.system('python PredictV4.py')
logging.info(f"Ran PredictV4.py with result: {result}")
print(f"Ran PredictV4.py with result: {result}")
with open('processed_value.txt', 'r') as f:
result = f.read().strip()
logging.info(f"Read result: {result}")
print(f"Read result: {result}")
# Clean up temporary files
os.remove('value_to_process.txt')
os.remove('processed_value.txt')
return result
except FileNotFoundError:
logging.error("Error: processed_value.txt not found")
print("Error: processed_value.txt not found")
return None

def write_result_to_excel(self, result, cell):
try:
workbook = xlrd.open_workbook(self.file_path, formatting_info=True)
writable_workbook = copy(workbook)
writable_sheet = writable_workbook.get_sheet('Aspen_Input')

row_idx = int(cell[1:]) - 1 # Convert cell to row index (e.g., B2 -> 1)
col_idx = ord(cell[0].upper()) - ord('A') # Convert cell to column index (e.g., B -> 1, C -> 2)

writable_sheet.write(row_idx, col_idx, result) # Write the result to the specified cell

writable_workbook.save(self.file_path)
logging.info(f"Wrote result to Excel in {cell}: {result}")
print(f"Wrote result to Excel in {cell}: {result}")
except PermissionError as e:
logging.error(f"Permission denied: {e}")
print(f"Permission denied: {e}")
except Exception as e:
logging.error(f"Error writing to Excel file: {e}")
print(f"Error writing to Excel file: {e}")

def monitor_cpu(interval=1):
try:
while True:
cpu_usage = psutil.cpu_percent(interval=1)
print(f"CPU Usage: {cpu_usage}%")
time.sleep(interval)
except KeyboardInterrupt:
print("CPU monitoring stopped.")

if __name__ == "__main__":
logging.basicConfig(filename="excel_changes.log", level=logging.INFO,
format="%(asctime)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S")

path_to_monitor = r"C:\Users\germa\Desktop\Simulation_Test\ObserverV43.xls" # Use raw string
interval = 1 # Polling interval in seconds

handler = ExcelChangeHandler(path_to_monitor, interval)
handler.start()

# Start CPU monitoring in a separate thread
cpu_thread = Thread(target=monitor_cpu, daemon=True)
cpu_thread.start()

try:
while True:
time.sleep(1)
except KeyboardInterrupt:
handler.stop()
</code></pre>
<p>Running in parallel will block Aspen Plus from accessing the xls file</p>
 

Latest posts

Online statistics

Members online
0
Guests online
4
Total visitors
4
Top