Help with writing monitored data to mysql upon change of one particular variable

[donottrackmymetadata]

I am trying to monitor a set of variables which record a product recipe batch and record the final values (or very close to the final values) and write this into mysql when recipe name changes.

(If there is a better alternative to achieve this in another language or low cost software, I would be interested in hearing also)

This data is extracted via OPCUA via data change notification. The operator chooses the recipe name (tag_recipe) and upon this change, accumulated values (data_tag1-5) are reset to 0 instantaneously while the process is still operating.

What I am trying to do is store these values so when the recipe is changed I can take the final batch to total values and write them to my db, however what is happening is that I end up writing the values at the start of the newest batch so I will get results looking like:

sql_col1 | sql_col2 | sql_col3 | sql_col4 | sql_col5
Recipe_C | 5 | 2 | 0.1 | 0.3 (where as it should be Recipe_C | 2034 | 3456 | 403.038 | 234.034
Recipe_H | 2 | 0 | 0.1 | 0.1 (where as it should Recipe_C| 5506 | 9765 | 964.650 | 533.964

A batch run can run for around 15 minutes to a few hours with products flowing through at up to 100 items per minute at full speed so I don’t know if the rapid data change rate is causing issues or it is storing too much accumulative data in the dictionary?

I have tried various edits but seem to be getting the same results and are banging my head up against the wall here. Is there something obvious I am missing or am I going about this the entirely wrong way?

(Note values have been changed to generic names)

import mysql.connector
from opcua import Client

# Generic MySQL connection details
MYSQL_HOST = "your_mysql_host"
MYSQL_USER = "your_mysql_username"
MYSQL_PASSWORD = "your_mysql_password"
MYSQL_DATABASE = "your_mysql_database"

# Generic OPC UA server URL
OPC_SERVER_URL = "opc.tcp://your_opc_server_ip:your_opc_server_port"

# Dictionary to store cumulative data for each batch identified by recipe Name
batch_data = {}

# Dictionary to store cumulative values for each tag
cumulative_data = {}

# Function to update MySQL database with cumulative values for a specific batch
def update_mysql_database(batch_name):
    print("Updating MySQL database for batch:", batch_name)
    batch_cumulative_data = batch_data[batch_name]
    print("Batch Cumulative Data:", batch_cumulative_data)
    # Convert weight from grams to kilograms
    acc1_kg = batch_cumulative_data.get("data_tag_1", 0) / 1000.0
    acc2_weight_kg = batch_cumulative_data.get("data_tag_4", 0) / 1000.0
    print("Acc1 Weight (kg):", acc1_weight_kg)
    print("Acc2 Weight (kg):", acc2_weight_kg)

    try:
        # Write a new row to the database with cumulative values for the batch
        sql = "INSERT INTO your_database_table (sql_col1, sql_col2, sql_col3, sql_col4, sql_col5) VALUES (%s, %s, %s, %s, %s)"
        val = (
            batch_cumulative_data.get("tag_recipe", "N/A"),
            batch_cumulative_data.get("data_tag_3", 0),
            batch_cumulative_data.get("data_tag_2", 0),
            acc1_weight_kg,
            acc2_weight_kg   
        )

        print("Executing SQL:", sql)
        print("Values:", val)

        cursor.execute(sql, val)
        mydb.commit()
        print("Data inserted successfully for batch:", batch_name)
    except mysql.connector.Error as error:
        print("Error inserting data into MySQL:", error)

# Define the subhandler class to handle data change notifications
class subhandler(object):

    def datachange_notification(self, node, val, data):
        print("Data Change Notification Received:")
        print("Node: ", node.nodeid)
        print("Value: ", val)

        # Update cumulative data for relevant nodes
        cumulative_tags = [
            "data_tag_1",
            "data_tag_2",
            "data_tag_3",
            "data_tag_4",
            "tag_recipe"
        ]

        for tag in cumulative_tags: 
            if f"OPC-LINK.{tag}" in node.nodeid.Identifier:
                print(f"Updating cumulative data for tag: {tag}")
                cumulative_data[tag] = val

        # Check if recipe Name has changed, indicating the end of a production run
        if "tag_recipe" in node.nodeid.Identifier:
            print("recipe changed:", val)
            print("Cumulative Data:", cumulative_data)
            batch_name = cumulative_data["tag_recipe"]
            if batch_name not in batch_data:
                batch_data[batch_name] = {}
            batch_data[batch_name].update(cumulative_data)
            print("Updating MySQL database...")
            update_mysql_database(batch_name)

# Connect to MySQL database
try:
    mydb = mysql.connector.connect(
        host=MYSQL_HOST,
        user=MYSQL_USER,
        password=MYSQL_PASSWORD,
        database=MYSQL_DATABASE
    )
    print("Connected to MySQL database successfully.")
except mysql.connector.Error as e:
    print("Error connecting to MySQL database:", e)
    exit()

cursor = mydb.cursor()

# Connect to OPC UA server
client = Client(OPC_SERVER_URL)
client.session_timeout = 10000

client.connect()
print(f"Connected to: {OPC_SERVER_URL}")

# Subscribe to data change events for the specified tags
tags_to_monitor = [
    "data_tag_1",
    "data_tag_2",
    "data_tag_3",
    "data_tag_4",
    "tag_recipe"
]

for tag in tags_to_monitor:
    try:
        node = client.get_node(f"ns=2;s=OPC-LINK.{tag}")
        
        # Fetch and print the current value of the node
        current_value = node.get_value()
        print(f"Current value of {tag}: {current_value}")
        
        # Create the handler and subscription
        handler = subhandler()
        print(f"Handler created for tag: {tag}")
        subscription = client.create_subscription(2000, handler)
        handle = subscription.subscribe_data_change(node)
        print(f"Subscription created for tag: {tag}")
    except Exception as e:
        print(f"Error subscribing to tag {tag}: {e}")

try:
    while True:
        pass
except KeyboardInterrupt:
    client.disconnect()
    cursor.close()
    mydb.close()

Sorry for my formatting in the post if it ends up looking incorrect

deanhystad write Apr-17-2024, 04:59 AM:
Please post all code, output and errors (it it's entirety) between their respective tags. Refer to BBCode help topic on how to post. Use the "Preview Post" button to make sure the code is presented as you expect before hitting the "Post Reply/Thread" button.