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.

[deanhystad]

I'm displaying my ignorance here, but if you get notified when the recipe changes, isn't it too late to log the last values from the previous recipe?

[Pedroski55]

Whatever you want to do with Python in a MySQL database, I recommend you try out the MySQL queries first. I use phpMyAdmin.

A column id is always useful, but I also made recipe_name unique, I assume you can only have 1 example of each recipe name.

Given a table created like this with the columns id, recipe_name, A, B, C, D, E:

Output:
CREATE TABLE recipes (id INT PRIMARY KEY NOT NULL AUTO_INCREMENT, recipe_name VARCHAR(20) UNIQUE, A INT NOT NULL DEFAULT 0, B INT NOT NULL DEFAULT 0,  C INT NOT NULL DEFAULT 0, D FLOAT NOT NULL DEFAULT 0.0, E FLOAT NOT NULL DEFAULT 0.0) DEFAULT CHARACTER SET UTF8MB4 ENGINE=InnoDB;
ALTER TABLE recipes AUTO_INCREMENT=1;

Now set default values:

Output:
INSERT INTO `recipes`( `recipe_name`) VALUES ('recipe_A');
INSERT INTO `recipes`( `recipe_name`) VALUES ('recipe_B');
INSERT INTO `recipes`( `recipe_name`) VALUES ('recipe_C');
INSERT INTO `recipes`( `recipe_name`) VALUES ('recipe_D');
INSERT INTO `recipes`( `recipe_name`) VALUES ('recipe_E');
INSERT INTO `recipes`( `recipe_name`) VALUES ('recipe_F');
INSERT INTO `recipes`( `recipe_name`) VALUES ('recipe_G');

Now you can UPDATE the values for any given recipe_X in column recipe_name:

Output:
UPDATE `recipes` SET `A`= 6,`B` = 5,`C` = 4,`D` = 0.1,`E` = 0.3 WHERE recipe_name = 'recipe_A';
UPDATE `recipes` SET `A`= 6,`B` = 5,`C` = 4,`D` = 0.1,`E` = 0.3 WHERE recipe_name = 'recipe_B';
UPDATE `recipes` SET `A`= 6,`B` = 5,`C` = 4,`D` = 0.1,`E` = 0.3 WHERE recipe_name = 'recipe_C';
UPDATE `recipes` SET `A`= 6,`B` = 5,`C` = 4,`D` = 0.1,`E` = 0.3 WHERE recipe_name = 'recipe_D';
UPDATE `recipes` SET `A`= 6,`B` = 5,`C` = 4,`D` = 0.1,`E` = 0.3 WHERE recipe_name = 'recipe_E';
UPDATE `recipes` SET `A`= 6,`B` = 5,`C` = 4,`D` = 0.1,`E` = 0.3 WHERE recipe_name = 'recipe_F';
UPDATE `recipes` SET `A`= 6,`B` = 5,`C` = 4,`D` = 0.1,`E` = 0.3 WHERE recipe_name = 'recipe_G';

The above is what you need to implement using Python.

Basically, if something changes, you need to know which recipe_X, and the changed values. Not sure where you are getting them from. Say you get a list:

data = [2555, 3225, 5424, 0.11, 0.23, 'recipe_B'] 

Then your UPDATE command sent from Python looks like this:

Output:
UPDATE `recipes` SET `A`= data[0],`B` = data[1],`C` = data[2],`D` = data[3],`E` = data[4] WHERE recipe_name = data[5];

If you supply some data sets as a csv, we can try this out.

[deanhystad]

Sounds like the problem is that you don't know when the recipe change notification is sent relative to when the data is reset.

One way around this problem is not use the value change notification for the data values. Since your program does nothing while waiting for event change notification, you could poll to get the current values. That would make your program simpler and remove the uncertainty of knowing when data changed during the process versus a reset when the recipe changes.

If you don't like polling, you could take advantage of your data being cumulative. Ignore data that is less than the archived value, assuming any decrease is an artifact of a recipe reset. Something like this:

# Tags in same order as db columns
DATA_TAGS = ["tag_recipe", "data_tag_3", "data_tag_2", "data_tag_1", "data_tag_4", "tag_recipe"]

def reset_data(recipe=None):
    global data
    data = dict(zip(DATA_TAGS, (recipe, 0, 0, 0, 0)))


def update_mysql_database():
    """Add data to database."""
    try:
        # Write a new row to the database with cumulative values for the batch
        data["data_tag_1"] /= 1000  # convert grams to kilograms.
        data["data_tag_4"] /= 1000
        cursor.execute(
            "INSERT INTO your_database_table (sql_col1, sql_col2, sql_col3, sql_col4, sql_col5) VALUES (%s, %s, %s, %s, %s)",
            list(data.values()
        )
        mydb.commit()
    except mysql.connector.Error as error:
        print("Error inserting data into MySQL:", error)


class subhandler():  # Should have better name, like DataChangeHandler
    """Data change handler.  Same class used for all data change events."""
    def datachange_notification(self, node, val, data):
        if "tag_recipe" in node.nodeid.Identifier:
            if data["tag_recipe"] not in (None, val):
                # Recipe changed.  Save old recipe data
                update_mysql_database()
                reset_data(val)
            return

        for tag in data:
            if tag in node.nodeid.Identifier:
                # Assume values always increase unless reset.
                data[tag] = max(val, data[tag])
                break