MyEMS / myems

myems-modbus-tcp/acquisition.py

import os
import json
import math
import telnetlib3
import asyncio
import time
from datetime import datetime
from decimal import Decimal
import mysql.connector
from modbus_tk import modbus_tcp
import config
from byte_swap import byte_swap_32_bit, byte_swap_64_bit


########################################################################################################################
# Check connectivity to the host and port
########################################################################################################################
async def check_connectivity(host, port):
    reader, writer = await telnetlib3.open_connection(host, port)
    # Close the connection
    writer.close()

########################################################################################################################
# Acquisition Procedures
# Step 1: Update process id in database
# Step 2: Check connectivity to the host and port
# Step 3: Get point list
# Step 4: Read point values from Modbus slaves
# Step 5: Bulk insert point values and update latest values in historical database
########################################################################################################################


def process(logger, data_source_id, host, port, interval_in_seconds):
    ####################################################################################################################
    # Step 1: Update process id in database
    ####################################################################################################################
    cnx_system_db = None
    cursor_system_db = None
    try:
        cnx_system_db = mysql.connector.connect(**config.myems_system_db)
        cursor_system_db = cnx_system_db.cursor()
    except Exception as e:
        logger.error("Error in step 1.1 of acquisition process " + str(e))
        if cursor_system_db:
            cursor_system_db.close()
        if cnx_system_db:
            cnx_system_db.close()
        return

    update_row = (" UPDATE tbl_data_sources "
                  " SET process_id = %s "
                  " WHERE id = %s ")
    try:
        cursor_system_db.execute(update_row, (os.getpid(), data_source_id,))
        cnx_system_db.commit()
    except Exception as e:
        logger.error("Error in step 1.2 of acquisition process " + str(e))
        return
    finally:
        if cursor_system_db:
            cursor_system_db.close()
        if cnx_system_db:
            cnx_system_db.close()

    while True:
        # begin of the outermost while loop
        ################################################################################################################
        # Step 2: Check connectivity to the host and port
        ################################################################################################################
        try:
            asyncio.run(check_connectivity(host, port))
            print("Succeeded to connect %s:%s in acquisition process ", host, port)
        except Exception as e:
            logger.error("Failed to connect %s:%s in acquisition process: %s  ", host, port, str(e))
            # go to begin of the outermost while loop
            time.sleep(300)
            continue

        ################################################################################################################
        # Step 3: Get point list
        ################################################################################################################
        cnx_system_db = None
        cursor_system_db = None
        try:
            cnx_system_db = mysql.connector.connect(**config.myems_system_db)
            cursor_system_db = cnx_system_db.cursor()
        except Exception as e:
            logger.error("Error in step 3.1 of acquisition process " + str(e))
            if cursor_system_db:
                cursor_system_db.close()
            if cnx_system_db:
                cnx_system_db.close()
            # go to begin of the outermost while loop
            time.sleep(60)
            continue

        try:
            query = (" SELECT id, name, object_type, is_trend, ratio, offset_constant, address "
                     " FROM tbl_points "
                     " WHERE data_source_id = %s AND is_virtual = 0 "
                     " ORDER BY id ")
            cursor_system_db.execute(query, (data_source_id,))
            rows_point = cursor_system_db.fetchall()
        except Exception as e:
            logger.error("Error in step 3.2 of acquisition process: " + str(e))
            if cursor_system_db:
                cursor_system_db.close()
            if cnx_system_db:
                cnx_system_db.close()
            # go to begin of the outermost while loop
            time.sleep(60)
            continue

        if rows_point is None or len(rows_point) == 0:
            # there is no points for this data source
            logger.error("Point Not Found in Data Source (ID = %s), acquisition process terminated ", data_source_id)
            if cursor_system_db:
                cursor_system_db.close()
            if cnx_system_db:
                cnx_system_db.close()
            # go to begin of the outermost while loop
            time.sleep(60)
            continue

        # There are points for this data source
        point_list = list()
        for row_point in rows_point:
            point_list.append({"id": row_point[0],
                               "name": row_point[1],
                               "object_type": row_point[2],
                               "is_trend": row_point[3],
                               "ratio": row_point[4],
                               "offset_constant": row_point[5],
                               "address": row_point[6]})

        ################################################################################################################
        # Step 4: Read point values from Modbus slaves
        ################################################################################################################
        # connect to historical database
        cnx_historical_db = None
        cursor_historical_db = None
        try:
            cnx_historical_db = mysql.connector.connect(**config.myems_historical_db)
            cursor_historical_db = cnx_historical_db.cursor()
        except Exception as e:
            logger.error("Error in step 4.1 of acquisition process " + str(e))
            if cursor_historical_db:
                cursor_historical_db.close()
            if cnx_historical_db:
                cnx_historical_db.close()

            if cursor_system_db:
                cursor_system_db.close()
            if cnx_system_db:
                cnx_system_db.close()
            # go to begin of the outermost while loop
            time.sleep(60)
            continue

        # connect to the Modbus data source
        master = modbus_tcp.TcpMaster(host=host, port=port, timeout_in_sec=5.0)
        master.set_timeout(5.0)
        print("Ready to connect to %s:%s ", host, port)

        # inner while loop to read all point values periodically
        while True:
            # begin of the inner while loop
            is_modbus_tcp_timed_out = False
            energy_value_list = list()
            analog_value_list = list()
            digital_value_list = list()

            # TODO: update point list in another thread
            # foreach point loop
            for point in point_list:
                # begin of foreach point loop
                try:
                    address = json.loads(point['address'])
                except Exception as e:
                    logger.error("Error in step 4.2 of acquisition process: Invalid point address in JSON " + str(e))
                    continue

                if 'slave_id' not in address.keys() \
                        or 'function_code' not in address.keys() \
                        or 'offset' not in address.keys() \
                        or 'number_of_registers' not in address.keys() \
                        or 'format' not in address.keys() \
                        or 'byte_swap' not in address.keys() \
                        or address['slave_id'] < 1 \
                        or address['function_code'] not in (1, 2, 3, 4) \
                        or address['offset'] < 0 \
                        or address['number_of_registers'] < 0 \
                        or len(address['format']) < 1 \
                        or not isinstance(address['byte_swap'], bool):
                    logger.error('Data Source(ID=%s), Point(ID=%s) Invalid address data.',
                                 data_source_id, point['id'])
                    # invalid point is found
                    # go to begin of foreach point loop to process next point
                    continue

                # read point value
                try:
                    result = master.execute(slave=address['slave_id'],
                                            function_code=address['function_code'],
                                            starting_address=address['offset'],
                                            quantity_of_x=address['number_of_registers'],
                                            data_format=address['format'])
                except Exception as e:
                    logger.error(str(e) +
                                 " host:" + host + " port:" + str(port) +
                                 " slave_id:" + str(address['slave_id']) +
                                 " function_code:" + str(address['function_code']) +
                                 " starting_address:" + str(address['offset']) +
                                 " quantity_of_x:" + str(address['number_of_registers']) +
                                 " data_format:" + str(address['format']) +
                                 " byte_swap:" + str(address['byte_swap']))

                    if 'timed out' in str(e):
                        is_modbus_tcp_timed_out = True
                        # timeout error
                        # break the foreach point loop
                        break
                    else:
                        # exception occurred when read register value,
                        # go to begin of foreach point loop to process next point
                        continue

                if result is None or not isinstance(result, tuple) or len(result) == 0:
                    logger.error("Error in step 4.3 of acquisition process: \n"
                                 " invalid result: None "
                                 " for point_id: " + str(point['id']))
                    # invalid result
                    # go to begin of foreach point loop to process next point
                    continue

                if not isinstance(result[0], float) and not isinstance(result[0], int) or math.isnan(result[0]):
                    logger.error(" Error in step 4.4 of acquisition process:\n"
                                 " invalid result: not float and not int or not a number "
                                 " for point_id: " + str(point['id']))
                    # invalid result
                    # go to begin of foreach point loop to process next point
                    continue

                if address['byte_swap']:
                    if address['number_of_registers'] == 2:
                        value = byte_swap_32_bit(result[0])
                    elif address['number_of_registers'] == 4:
                        value = byte_swap_64_bit(result[0])
                    else:
                        value = result[0]
                else:
                    value = result[0]

                if point['object_type'] == 'ANALOG_VALUE':
                    # Standard SQL requires that DECIMAL(18, 3) be able to store any value with 18 digits and
                    # 3 decimals, so values that can be stored in the column range
                    # from -999999999999999.999 to 999999999999999.999.
                    if Decimal(-999999999999999.999) <= Decimal(value) <= Decimal(999999999999999.999):
                        analog_value_list.append({'point_id': point['id'],
                                                  'is_trend': point['is_trend'],
                                                  'value': Decimal(value) * point['ratio'] + point['offset_constant']})
                elif point['object_type'] == 'ENERGY_VALUE':
                    # Standard SQL requires that DECIMAL(18, 3) be able to store any value with 18 digits and
                    # 3 decimals, so values that can be stored in the column range
                    # from -999999999999999.999 to 999999999999999.999.
                    if Decimal(-999999999999999.999) <= Decimal(value) <= Decimal(999999999999999.999):
                        energy_value_list.append({'point_id': point['id'],
                                                  'is_trend': point['is_trend'],
                                                  'value': Decimal(value) * point['ratio'] + point['offset_constant']})
                elif point['object_type'] == 'DIGITAL_VALUE':
                    digital_value_list.append({'point_id': point['id'],
                                               'is_trend': point['is_trend'],
                                               'value': int(value) * int(point['ratio']) + int(point['offset_constant'])
                                               })

            # end of foreach point loop

            if is_modbus_tcp_timed_out:
                # Modbus TCP connection timeout

                # destroy the Modbus master
                del master

                # close the connection to database
                if cursor_historical_db:
                    cursor_historical_db.close()
                if cnx_historical_db:
                    cnx_historical_db.close()
                if cursor_system_db:
                    cursor_system_db.close()
                if cnx_system_db:
                    cnx_system_db.close()

                # break the inner while loop
                # go to begin of the outermost while loop
                time.sleep(60)
                break

            ############################################################################################################
            # Step 5: Bulk insert point values and update latest values in historical database
            ############################################################################################################
            # check the connection to the Historical Database
            if not cnx_historical_db.is_connected():
                try:
                    cnx_historical_db = mysql.connector.connect(**config.myems_historical_db)
                    cursor_historical_db = cnx_historical_db.cursor()
                except Exception as e:
                    logger.error("Error in step 5.1 of acquisition process: " + str(e))
                    if cursor_historical_db:
                        cursor_historical_db.close()
                    if cnx_historical_db:
                        cnx_historical_db.close()
                    # go to begin of the inner while loop
                    time.sleep(60)
                    continue

            # check the connection to the System Database
            if not cnx_system_db.is_connected():
                try:
                    cnx_system_db = mysql.connector.connect(**config.myems_system_db)
                    cursor_system_db = cnx_system_db.cursor()
                except Exception as e:
                    logger.error("Error in step 5.2 of acquisition process: " + str(e))
                    if cursor_system_db:
                        cursor_system_db.close()
                    if cnx_system_db:
                        cnx_system_db.close()
                    # go to begin of the inner while loop
                    time.sleep(60)
                    continue

            current_datetime_utc = datetime.utcnow()
            # bulk insert values into historical database within a period
            # and then update latest values
            while len(analog_value_list) > 0:

This code seems to be duplicated in your project.

                analog_value_list_100 = analog_value_list[:100]
                analog_value_list = analog_value_list[100:]

                add_values = (" INSERT INTO tbl_analog_value (point_id, utc_date_time, actual_value) "
                              " VALUES  ")
                trend_value_count = 0

                for point_value in analog_value_list_100:
                    if point_value['is_trend']:
                        add_values += " (" + str(point_value['point_id']) + ","
                        add_values += "'" + current_datetime_utc.isoformat() + "',"
                        add_values += str(point_value['value']) + "), "
                        trend_value_count += 1

                if trend_value_count > 0:
                    try:
                        # trim ", " at the end of string and then execute
                        cursor_historical_db.execute(add_values[:-2])
                        cnx_historical_db.commit()
                    except Exception as e:
                        logger.error("Error in step 5.3.1 of acquisition process " + str(e))
                        # ignore this exception

                # update tbl_analog_value_latest
                delete_values = " DELETE FROM tbl_analog_value_latest WHERE point_id IN ( "
                latest_values = (" INSERT INTO tbl_analog_value_latest (point_id, utc_date_time, actual_value) "
                                 " VALUES  ")
                latest_value_count = 0

                for point_value in analog_value_list_100:
                    delete_values += str(point_value['point_id']) + ","
                    latest_values += " (" + str(point_value['point_id']) + ","
                    latest_values += "'" + current_datetime_utc.isoformat() + "',"
                    latest_values += str(point_value['value']) + "), "
                    latest_value_count += 1

                if latest_value_count > 0:
                    try:
                        # replace "," at the end of string with ")"
                        cursor_historical_db.execute(delete_values[:-1] + ")")
                        cnx_historical_db.commit()
                    except Exception as e:
                        logger.error("Error in step 5.3.2 of acquisition process " + str(e))
                        # ignore this exception

                    try:
                        # trim ", " at the end of string and then execute
                        cursor_historical_db.execute(latest_values[:-2])
                        cnx_historical_db.commit()
                    except Exception as e:
                        logger.error("Error in step 5.3.3 of acquisition process " + str(e))
                        # ignore this exception

            while len(energy_value_list) > 0:

This code seems to be duplicated in your project.

                energy_value_list_100 = energy_value_list[:100]
                energy_value_list = energy_value_list[100:]

                add_values = (" INSERT INTO tbl_energy_value (point_id, utc_date_time, actual_value) "
                              " VALUES  ")
                trend_value_count = 0

                for point_value in energy_value_list_100:
                    if point_value['is_trend']:
                        add_values += " (" + str(point_value['point_id']) + ","
                        add_values += "'" + current_datetime_utc.isoformat() + "',"
                        add_values += str(point_value['value']) + "), "
                        trend_value_count += 1

                if trend_value_count > 0:
                    try:
                        # trim ", " at the end of string and then execute
                        cursor_historical_db.execute(add_values[:-2])
                        cnx_historical_db.commit()
                    except Exception as e:
                        logger.error("Error in step 5.4.1 of acquisition process: " + str(e))
                        # ignore this exception

                # update tbl_energy_value_latest
                delete_values = " DELETE FROM tbl_energy_value_latest WHERE point_id IN ( "
                latest_values = (" INSERT INTO tbl_energy_value_latest (point_id, utc_date_time, actual_value) "
                                 " VALUES  ")
                latest_value_count = 0
                for point_value in energy_value_list_100:
                    delete_values += str(point_value['point_id']) + ","
                    latest_values += " (" + str(point_value['point_id']) + ","
                    latest_values += "'" + current_datetime_utc.isoformat() + "',"
                    latest_values += str(point_value['value']) + "), "
                    latest_value_count += 1

                if latest_value_count > 0:
                    try:
                        # replace "," at the end of string with ")"
                        cursor_historical_db.execute(delete_values[:-1] + ")")
                        cnx_historical_db.commit()

                    except Exception as e:
                        logger.error("Error in step 5.4.2 of acquisition process " + str(e))
                        # ignore this exception

                    try:
                        # trim ", " at the end of string and then execute
                        cursor_historical_db.execute(latest_values[:-2])
                        cnx_historical_db.commit()

                    except Exception as e:
                        logger.error("Error in step 5.4.3 of acquisition process " + str(e))
                        # ignore this exception

            while len(digital_value_list) > 0:

This code seems to be duplicated in your project.

                digital_value_list_100 = digital_value_list[:100]
                digital_value_list = digital_value_list[100:]

                add_values = (" INSERT INTO tbl_digital_value (point_id, utc_date_time, actual_value) "
                              " VALUES  ")
                trend_value_count = 0

                for point_value in digital_value_list_100:
                    if point_value['is_trend']:
                        add_values += " (" + str(point_value['point_id']) + ","
                        add_values += "'" + current_datetime_utc.isoformat() + "',"
                        add_values += str(point_value['value']) + "), "
                        trend_value_count += 1

                if trend_value_count > 0:
                    try:
                        # trim ", " at the end of string and then execute
                        cursor_historical_db.execute(add_values[:-2])
                        cnx_historical_db.commit()
                    except Exception as e:
                        logger.error("Error in step 5.5.1 of acquisition process: " + str(e))
                        # ignore this exception

                # update tbl_digital_value_latest
                delete_values = " DELETE FROM tbl_digital_value_latest WHERE point_id IN ( "
                latest_values = (" INSERT INTO tbl_digital_value_latest (point_id, utc_date_time, actual_value) "
                                 " VALUES  ")
                latest_value_count = 0
                for point_value in digital_value_list_100:
                    delete_values += str(point_value['point_id']) + ","
                    latest_values += " (" + str(point_value['point_id']) + ","
                    latest_values += "'" + current_datetime_utc.isoformat() + "',"
                    latest_values += str(point_value['value']) + "), "
                    latest_value_count += 1

                if latest_value_count > 0:
                    try:
                        # replace "," at the end of string with ")"
                        cursor_historical_db.execute(delete_values[:-1] + ")")
                        cnx_historical_db.commit()
                    except Exception as e:
                        logger.error("Error in step 5.5.2 of acquisition process " + str(e))
                        # ignore this exception

                    try:
                        # trim ", " at the end of string and then execute
                        cursor_historical_db.execute(latest_values[:-2])
                        cnx_historical_db.commit()
                    except Exception as e:
                        logger.error("Error in step 5.5.3 of acquisition process " + str(e))
                        # ignore this exception

            # update data source last seen datetime
            update_row = (" UPDATE tbl_data_sources "
                          " SET last_seen_datetime_utc = '" + current_datetime_utc.isoformat() + "' "
                          " WHERE id = %s ")
            try:
                cursor_system_db.execute(update_row, (data_source_id,))
                cnx_system_db.commit()
            except Exception as e:
                logger.error("Error in step 5.6 of acquisition process " + str(e))
                if cursor_system_db:
                    cursor_system_db.close()
                if cnx_system_db:
                    cnx_system_db.close()
                # go to begin of the inner while loop
                time.sleep(60)
                continue

            # Sleep interval in seconds and continue the inner while loop
            # this argument may be a floating point number for subsecond precision
            time.sleep(interval_in_seconds)

        # end of the inner while loop

    # end of the outermost while loop