virtualpoint.calculate() - Code Metrics - MyEMS/myems - Measure and Improve Code Quality continuously with Scrutinizer

virtualpoint.calculate() F
last analyzed 2025-08-28 05:12 UTC

↳ Parent: virtualpoint

Complexity

Conditions

Size

Total Lines	72
Code Lines	52

Duplication

Lines	72
Ratio	100 %

Importance

Changes

Metric	Value
eloc	52
dl	72
loc	72
rs	3.6
c	0
b	0
f	0
cc	14
nop	1

How to fix Long Method Complexity

import json
import random
import re
import time
from datetime import datetime
from decimal import Decimal
from multiprocessing import Pool
import mysql.connector
from sympy import sympify, Piecewise, symbols
import config


########################################################################################################################
# PROCEDURES:
# Step 1: Query all virtual points
# Step 2: Create multiprocessing pool to call worker in parallel
########################################################################################################################

def calculate(logger):

    while True:
        # the outermost while loop to reconnect server if there is a connection error
        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 0 of virtual point calculate " + str(e))
            if cursor_system_db:
                cursor_system_db.close()
            if cnx_system_db:
                cnx_system_db.close()
            # sleep and continue the outer loop to reconnect the database
            time.sleep(60)
            continue

        print("Connected to MyEMS System Database")

        virtual_point_list = list()
        try:
            cursor_system_db.execute(" SELECT id, name, data_source_id, object_type, high_limit, low_limit, address "
                                     " FROM tbl_points "
                                     " WHERE is_virtual = 1 ")
            rows_virtual_points = cursor_system_db.fetchall()

            if rows_virtual_points is None or len(rows_virtual_points) == 0:
                # sleep several minutes and continue the outer loop to reconnect the database
                time.sleep(60)
                continue

            for row in rows_virtual_points:
                meta_result = {"id": row[0],
                               "name": row[1],
                               "data_source_id": row[2],
                               "object_type": row[3],
                               "high_limit": row[4],
                               "low_limit": row[5],
                               "address": row[6]}
                virtual_point_list.append(meta_result)

        except Exception as e:
            logger.error("Error in step 1 of virtual point calculate " + str(e))
            # sleep and continue the outer loop to reconnect the database
            time.sleep(60)
            continue
        finally:
            if cursor_system_db:
                cursor_system_db.close()
            if cnx_system_db:
                cnx_system_db.close()

        # shuffle the virtual point list for randomly calculating
        random.shuffle(virtual_point_list)

        print("Got all virtual points in MyEMS System Database")
        ################################################################################################################
        # Step 2: Create multiprocessing pool to call worker in parallel
        ################################################################################################################
        p = Pool(processes=config.pool_size)
        error_list = p.map(worker, virtual_point_list)
        p.close()
        p.join()

        for error in error_list:
            if error is not None and len(error) > 0:
                logger.error(error)

        print("go to sleep ")
        time.sleep(60)
        print("wake from sleep, and continue to work")


########################################################################################################################
# Step 1: get start datetime and end datetime
# Step 2: parse the expression and get all points in substitutions
# Step 3: query points type from system database
# Step 4: query points value from historical database
# Step 5: evaluate the equation with points values
########################################################################################################################

def worker(virtual_point):
    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:
        if cursor_historical_db:
            cursor_historical_db.close()
        if cnx_historical_db:
            cnx_historical_db.close()
        return "Error in step 1.1 of virtual point worker " + str(e) + " for '" + virtual_point['name'] + "'"

    print("Start to process virtual point: " + "'" + virtual_point['name'] + "'")

    ####################################################################################################################
    # step 1: get start datetime and end datetime
    ####################################################################################################################
    if virtual_point['object_type'] == 'ANALOG_VALUE':
        table_name = "tbl_analog_value"
    elif virtual_point['object_type'] == 'ENERGY_VALUE':
        table_name = "tbl_energy_value"
    else:
        if cursor_historical_db:
            cursor_historical_db.close()
        if cnx_historical_db:
            cnx_historical_db.close()
        return "variable point type should not be DIGITAL_VALUE " + " for '" + virtual_point['name'] + "'"

    try:
        query = (" SELECT MAX(utc_date_time) "
                 " FROM " + table_name +
                 " WHERE point_id = %s ")
        cursor_historical_db.execute(query, (virtual_point['id'],))
        row = cursor_historical_db.fetchone()
    except Exception as e:
        if cursor_historical_db:
            cursor_historical_db.close()
        if cnx_historical_db:
            cnx_historical_db.close()
        return "Error in step 1.2 of virtual point worker " + str(e) + " for '" + virtual_point['name'] + "'"

    start_datetime_utc = datetime.strptime(config.start_datetime_utc, '%Y-%m-%d %H:%M:%S').replace(tzinfo=None)

    if row is not None and len(row) > 0 and isinstance(row[0], datetime):
        start_datetime_utc = row[0].replace(tzinfo=None)

    end_datetime_utc = datetime.utcnow().replace(tzinfo=None)

    if end_datetime_utc <= start_datetime_utc:
        if cursor_historical_db:
            cursor_historical_db.close()
        if cnx_historical_db:
            cnx_historical_db.close()
        return "it isn't time to calculate" + " for '" + virtual_point['name'] + "'"

    print("start_datetime_utc: " + start_datetime_utc.isoformat()[0:19]
          + "end_datetime_utc: " + end_datetime_utc.isoformat()[0:19])

    ############################################################################################################
    # Step 2: parse the expression and get all points in substitutions
    ############################################################################################################
    point_list = list()
    try:
        ########################################################################################################
        # parse the expression and get all points in substitutions
        ########################################################################################################
        address = json.loads(virtual_point['address'])
        # algebraic expression example: '{"expression": "x1-x2", "substitutions": {"x1":1,"x2":2}}'
        # piecewise function example: '{"expression":"(1,x<200 ), (2,x>=500), (0,True)", "substitutions":{"x":101}}'
        if 'expression' not in address.keys() \
                or 'substitutions' not in address.keys() \
                or len(address['expression']) == 0 \
                or len(address['substitutions']) == 0:
            return "Error in step 2.1 of virtual point worker for '" + virtual_point['name'] + "'"
        expression = address['expression']
        substitutions = address['substitutions']
        for variable_name, point_id in substitutions.items():
            point_list.append({"variable_name": variable_name, "point_id": point_id})
    except Exception as e:
        if cursor_historical_db:
            cursor_historical_db.close()
        if cnx_historical_db:
            cnx_historical_db.close()
        return "Error in step 2.2 of virtual point worker " + str(e) + " for '" + virtual_point['name'] + "'"

    ############################################################################################################
    # Step 3: query points type from system database
    ############################################################################################################
    print("getting points type ")
    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:
        if cursor_system_db:
            cursor_system_db.close()
        if cnx_system_db:
            cnx_system_db.close()
        print("Error in step 3 of virtual point worker " + str(e))
        return "Error in step 3 of virtual point worker " + str(e)

    print("Connected to MyEMS System Database")

    all_point_dict = dict()
    try:
        cursor_system_db.execute(" SELECT id, object_type "
                                 " FROM tbl_points ")
        rows_points = cursor_system_db.fetchall()

        if rows_points is None or len(rows_points) == 0:
            return "Error in step 3.1 of virtual point worker for '" + virtual_point['name'] + "'"

        for row in rows_points:
            all_point_dict[row[0]] = row[1]
    except Exception as e:
        return "Error in step 3.2 virtual point worker " + str(e) + " for '" + virtual_point['name'] + "'"
    finally:
        if cursor_system_db:
            cursor_system_db.close()
        if cnx_system_db:
            cnx_system_db.close()
    ############################################################################################################
    # Step 4: query points value from historical database
    ############################################################################################################

    print("getting point values ")
    point_values_dict = dict()
    if point_list is not None and len(point_list) > 0:
        try:
            for point in point_list:
                point_object_type = all_point_dict.get(point['point_id'])
                if point_object_type is None:
                    return "variable point type should not be None " + " for '" + virtual_point['name'] + "'"
                if point_object_type == 'ANALOG_VALUE':
                    query = (" SELECT utc_date_time, actual_value "
                             " FROM tbl_analog_value "
                             " WHERE point_id = %s AND utc_date_time > %s AND utc_date_time < %s "
                             " ORDER BY utc_date_time ")
                    cursor_historical_db.execute(query, (point['point_id'], start_datetime_utc, end_datetime_utc,))
                    rows = cursor_historical_db.fetchall()
                    if rows is not None and len(rows) > 0:
                        point_values_dict[point['point_id']] = dict()
                        for row in rows:
                            point_values_dict[point['point_id']][row[0]] = row[1]
                elif point_object_type == 'ENERGY_VALUE':
                    query = (" SELECT utc_date_time, actual_value "
                             " FROM tbl_energy_value "
                             " WHERE point_id = %s AND utc_date_time > %s AND utc_date_time < %s "
                             " ORDER BY utc_date_time ")
                    cursor_historical_db.execute(query, (point['point_id'], start_datetime_utc, end_datetime_utc,))
                    rows = cursor_historical_db.fetchall()
                    if rows is not None and len(rows) > 0:
                        point_values_dict[point['point_id']] = dict()
                        for row in rows:
                            point_values_dict[point['point_id']][row[0]] = row[1]
                    else:
                        point_values_dict[point['point_id']] = None
                else:
                    # point type should not be DIGITAL_VALUE
                    return "variable point type should not be DIGITAL_VALUE " + " for '" + virtual_point['name'] + "'"
        except Exception as e:
            if cursor_historical_db:
                cursor_historical_db.close()
            if cnx_historical_db:
                cnx_historical_db.close()
            return "Error in step 4.1 virtual point worker " + str(e) + " for '" + virtual_point['name'] + "'"

    ############################################################################################################
    # Step 5: evaluate the equation with points values
    ############################################################################################################

    print("getting date time set for all points")
    utc_date_time_set = set()
    if point_values_dict is not None and len(point_values_dict) > 0:
        for point_id, point_values in point_values_dict.items():
            if point_values is not None and len(point_values) > 0:
                utc_date_time_set = utc_date_time_set.union(point_values.keys())

    print("evaluating the equation with SymPy")
    normalized_values = list()

    ############################################################################################################
    # Converting Strings to SymPy Expressions
    # The sympify function(that’s sympify, not to be confused with simplify) can be used to
    # convert strings into SymPy expressions.
    ############################################################################################################
    try:
        if re.search(',', expression):
            for item in substitutions.keys():
                locals()[item] = symbols(item)

            expr = eval(expression)
            print("the expression will be evaluated as piecewise function: " + str(expr))
        else:
            expr = sympify(expression)
            print("the expression will be evaluated as algebraic expression: " + str(expr))

        for utc_date_time in utc_date_time_set:
            meta_data = dict()
            meta_data['utc_date_time'] = utc_date_time

            ####################################################################################################
            # create a dictionary of Symbol: point pairs
            ####################################################################################################

            subs = dict()

            ####################################################################################################
            # Evaluating the expression at current_datetime_utc
            ####################################################################################################

            if point_list is not None and len(point_list) > 0:
                for point in point_list:
                    actual_value = point_values_dict[point['point_id']].get(utc_date_time, None)
                    if actual_value is None:
                        break
                    subs[point['variable_name']] = actual_value

            if len(subs) != len(point_list):
                continue

            ####################################################################################################
            # To numerically evaluate an expression with a Symbol at a point,
            # we might use subs followed by evalf,
            # but it is more efficient and numerically stable to pass the substitution to evalf
            # using the subs flag, which takes a dictionary of Symbol: point pairs.
            ####################################################################################################
            if re.search(',', expression):
                formula = Piecewise(*expr)
                meta_data['actual_value'] = Decimal(str(formula.subs(subs)))
                normalized_values.append(meta_data)
            else:
                meta_data['actual_value'] = Decimal(str(expr.evalf(subs=subs)))
                normalized_values.append(meta_data)
    except Exception as e:
        if cursor_historical_db:
            cursor_historical_db.close()
        if cnx_historical_db:
            cnx_historical_db.close()
        return "Error in step 5.1 virtual point worker " + str(e) + " for '" + virtual_point['name'] + "'"

    print("saving virtual points values to historical database")

    if len(normalized_values) > 0:
        latest_meta_data = normalized_values[0]

        while len(normalized_values) > 0:
            insert_100 = normalized_values[:100]
            normalized_values = normalized_values[100:]

            try:
                add_values = (" INSERT INTO " + table_name +
                              " (point_id, utc_date_time, actual_value) "
                              " VALUES  ")

                for meta_data in insert_100:
                    add_values += " (" + str(virtual_point['id']) + ","
                    add_values += "'" + meta_data['utc_date_time'].isoformat()[0:19] + "',"
                    add_values += str(meta_data['actual_value']) + "), "

                    if meta_data['utc_date_time'] > latest_meta_data['utc_date_time']:
                        latest_meta_data = meta_data

                # print("add_values:" + add_values)
                # trim ", " at the end of string and then execute
                cursor_historical_db.execute(add_values[:-2])
                cnx_historical_db.commit()
            except Exception as e:
                if cursor_historical_db:
                    cursor_historical_db.close()
                if cnx_historical_db:
                    cnx_historical_db.close()
                return "Error in step 5.2 virtual point worker " + str(e) + " for '" + virtual_point['name'] + "'"

        try:
            # update tbl_analog_value_latest or tbl_energy_value_latest
            delete_value = " DELETE FROM " + table_name + "_latest WHERE point_id = {} ".format(virtual_point['id'])
            # print("delete_value:" + delete_value)
            cursor_historical_db.execute(delete_value)
            cnx_historical_db.commit()

            latest_value = (" INSERT INTO " + table_name + "_latest (point_id, utc_date_time, actual_value) "
                            " VALUES ({}, '{}', {}) "
                            .format(virtual_point['id'],
                                    latest_meta_data['utc_date_time'].isoformat()[0:19],
                                    latest_meta_data['actual_value']))
            # print("latest_value:" + latest_value)
            cursor_historical_db.execute(latest_value)
            cnx_historical_db.commit()
        except Exception as e:
            if cursor_historical_db:
                cursor_historical_db.close()
            if cnx_historical_db:
                cnx_historical_db.close()
            return "Error in step 5.3 virtual point worker " + str(e) + " for '" + virtual_point['name'] + "'"

    if cursor_historical_db:
        cursor_historical_db.close()
    if cnx_historical_db:
        cnx_historical_db.close()

    return None


1		import json
2		import random
3		import re
4		import time
5		from datetime import datetime
6		from decimal import Decimal
7		from multiprocessing import Pool
8		import mysql.connector
9		from sympy import sympify, Piecewise, symbols
10		import config
11
12
13		########################################################################################################################
14		# PROCEDURES:
15		# Step 1: Query all virtual points
16		# Step 2: Create multiprocessing pool to call worker in parallel
17		########################################################################################################################
18
19	View Code Duplication	def calculate(logger):
		0 ignored issues – show Duplication introduced 2021-03-07 06:03 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
20		while True:
21		# the outermost while loop to reconnect server if there is a connection error
22		cnx_system_db = None
23		cursor_system_db = None
24		try:
25		cnx_system_db = mysql.connector.connect(**config.myems_system_db)
26		cursor_system_db = cnx_system_db.cursor()
27		except Exception as e:
28		logger.error("Error in step 0 of virtual point calculate " + str(e))
29		if cursor_system_db:
30		cursor_system_db.close()
31		if cnx_system_db:
32		cnx_system_db.close()
33		# sleep and continue the outer loop to reconnect the database
34		time.sleep(60)
35		continue
36
37		print("Connected to MyEMS System Database")
38
39		virtual_point_list = list()
40		try:
41		cursor_system_db.execute(" SELECT id, name, data_source_id, object_type, high_limit, low_limit, address "
42		" FROM tbl_points "
43		" WHERE is_virtual = 1 ")
44		rows_virtual_points = cursor_system_db.fetchall()
45
46		if rows_virtual_points is None or len(rows_virtual_points) == 0:
47		# sleep several minutes and continue the outer loop to reconnect the database
48		time.sleep(60)
49		continue
50
51		for row in rows_virtual_points:
52		meta_result = {"id": row[0],
53		"name": row[1],
54		"data_source_id": row[2],
55		"object_type": row[3],
56		"high_limit": row[4],
57		"low_limit": row[5],
58		"address": row[6]}
59		virtual_point_list.append(meta_result)
60
61		except Exception as e:
62		logger.error("Error in step 1 of virtual point calculate " + str(e))
63		# sleep and continue the outer loop to reconnect the database
64		time.sleep(60)
65		continue
66		finally:
67		if cursor_system_db:
68		cursor_system_db.close()
69		if cnx_system_db:
70		cnx_system_db.close()
71
72		# shuffle the virtual point list for randomly calculating
73		random.shuffle(virtual_point_list)
74
75		print("Got all virtual points in MyEMS System Database")
76		################################################################################################################
77		# Step 2: Create multiprocessing pool to call worker in parallel
78		################################################################################################################
79		p = Pool(processes=config.pool_size)
80		error_list = p.map(worker, virtual_point_list)
81		p.close()
82		p.join()
83
84		for error in error_list:
85		if error is not None and len(error) > 0:
86		logger.error(error)
87
88		print("go to sleep ")
89		time.sleep(60)
90		print("wake from sleep, and continue to work")
91
92
93		########################################################################################################################
94		# Step 1: get start datetime and end datetime
95		# Step 2: parse the expression and get all points in substitutions
96		# Step 3: query points type from system database
97		# Step 4: query points value from historical database
98		# Step 5: evaluate the equation with points values
99		########################################################################################################################
100
101		def worker(virtual_point):
102		cnx_historical_db = None
103		cursor_historical_db = None
104
105		try:
106		cnx_historical_db = mysql.connector.connect(**config.myems_historical_db)
107		cursor_historical_db = cnx_historical_db.cursor()
108		except Exception as e:
109		if cursor_historical_db:
110		cursor_historical_db.close()
111		if cnx_historical_db:
112		cnx_historical_db.close()
113		return "Error in step 1.1 of virtual point worker " + str(e) + " for '" + virtual_point['name'] + "'"
114
115		print("Start to process virtual point: " + "'" + virtual_point['name'] + "'")
116
117		####################################################################################################################
118		# step 1: get start datetime and end datetime
119		####################################################################################################################
120		if virtual_point['object_type'] == 'ANALOG_VALUE':
121		table_name = "tbl_analog_value"
122		elif virtual_point['object_type'] == 'ENERGY_VALUE':
123		table_name = "tbl_energy_value"
124		else:
125		if cursor_historical_db:
126		cursor_historical_db.close()
127		if cnx_historical_db:
128		cnx_historical_db.close()
129		return "variable point type should not be DIGITAL_VALUE " + " for '" + virtual_point['name'] + "'"
130
131		try:
132		query = (" SELECT MAX(utc_date_time) "
133		" FROM " + table_name +
134		" WHERE point_id = %s ")
135		cursor_historical_db.execute(query, (virtual_point['id'],))
136		row = cursor_historical_db.fetchone()
137		except Exception as e:
138		if cursor_historical_db:
139		cursor_historical_db.close()
140		if cnx_historical_db:
141		cnx_historical_db.close()
142		return "Error in step 1.2 of virtual point worker " + str(e) + " for '" + virtual_point['name'] + "'"
143
144		start_datetime_utc = datetime.strptime(config.start_datetime_utc, '%Y-%m-%d %H:%M:%S').replace(tzinfo=None)
145
146		if row is not None and len(row) > 0 and isinstance(row[0], datetime):
147		start_datetime_utc = row[0].replace(tzinfo=None)
148
149		end_datetime_utc = datetime.utcnow().replace(tzinfo=None)
150
151		if end_datetime_utc <= start_datetime_utc:
152		if cursor_historical_db:
153		cursor_historical_db.close()
154		if cnx_historical_db:
155		cnx_historical_db.close()
156		return "it isn't time to calculate" + " for '" + virtual_point['name'] + "'"
157
158		print("start_datetime_utc: " + start_datetime_utc.isoformat()[0:19]
159		+ "end_datetime_utc: " + end_datetime_utc.isoformat()[0:19])
160
161		############################################################################################################
162		# Step 2: parse the expression and get all points in substitutions
163		############################################################################################################
164		point_list = list()
165		try:
166		########################################################################################################
167		# parse the expression and get all points in substitutions
168		########################################################################################################
169		address = json.loads(virtual_point['address'])
170		# algebraic expression example: '{"expression": "x1-x2", "substitutions": {"x1":1,"x2":2}}'
171		# piecewise function example: '{"expression":"(1,x<200 ), (2,x>=500), (0,True)", "substitutions":{"x":101}}'
172		if 'expression' not in address.keys() \
173		or 'substitutions' not in address.keys() \
174		or len(address['expression']) == 0 \
175		or len(address['substitutions']) == 0:
176		return "Error in step 2.1 of virtual point worker for '" + virtual_point['name'] + "'"
177		expression = address['expression']
178		substitutions = address['substitutions']
179		for variable_name, point_id in substitutions.items():
180		point_list.append({"variable_name": variable_name, "point_id": point_id})
181		except Exception as e:
182		if cursor_historical_db:
183		cursor_historical_db.close()
184		if cnx_historical_db:
185		cnx_historical_db.close()
186		return "Error in step 2.2 of virtual point worker " + str(e) + " for '" + virtual_point['name'] + "'"
187
188		############################################################################################################
189		# Step 3: query points type from system database
190		############################################################################################################
191		print("getting points type ")
192		cnx_system_db = None
193		cursor_system_db = None
194		try:
195		cnx_system_db = mysql.connector.connect(**config.myems_system_db)
196		cursor_system_db = cnx_system_db.cursor()
197		except Exception as e:
198		if cursor_system_db:
199		cursor_system_db.close()
200		if cnx_system_db:
201		cnx_system_db.close()
202		print("Error in step 3 of virtual point worker " + str(e))
203		return "Error in step 3 of virtual point worker " + str(e)
204
205		print("Connected to MyEMS System Database")
206
207		all_point_dict = dict()
208		try:
209		cursor_system_db.execute(" SELECT id, object_type "
210		" FROM tbl_points ")
211		rows_points = cursor_system_db.fetchall()
212
213		if rows_points is None or len(rows_points) == 0:
214		return "Error in step 3.1 of virtual point worker for '" + virtual_point['name'] + "'"
215
216		for row in rows_points:
217		all_point_dict[row[0]] = row[1]
218		except Exception as e:
219		return "Error in step 3.2 virtual point worker " + str(e) + " for '" + virtual_point['name'] + "'"
220		finally:
221		if cursor_system_db:
222		cursor_system_db.close()
223		if cnx_system_db:
224		cnx_system_db.close()
225		############################################################################################################
226		# Step 4: query points value from historical database
227		############################################################################################################
228
229		print("getting point values ")
230		point_values_dict = dict()
231		if point_list is not None and len(point_list) > 0:
232		try:
233		for point in point_list:
234		point_object_type = all_point_dict.get(point['point_id'])
235		if point_object_type is None:
236		return "variable point type should not be None " + " for '" + virtual_point['name'] + "'"
237		if point_object_type == 'ANALOG_VALUE':
238		query = (" SELECT utc_date_time, actual_value "
239		" FROM tbl_analog_value "
240		" WHERE point_id = %s AND utc_date_time > %s AND utc_date_time < %s "
241		" ORDER BY utc_date_time ")
242		cursor_historical_db.execute(query, (point['point_id'], start_datetime_utc, end_datetime_utc,))
243		rows = cursor_historical_db.fetchall()
244		if rows is not None and len(rows) > 0:
245		point_values_dict[point['point_id']] = dict()
246		for row in rows:
247		point_values_dict[point['point_id']][row[0]] = row[1]
248		elif point_object_type == 'ENERGY_VALUE':
249		query = (" SELECT utc_date_time, actual_value "
250		" FROM tbl_energy_value "
251		" WHERE point_id = %s AND utc_date_time > %s AND utc_date_time < %s "
252		" ORDER BY utc_date_time ")
253		cursor_historical_db.execute(query, (point['point_id'], start_datetime_utc, end_datetime_utc,))
254		rows = cursor_historical_db.fetchall()
255		if rows is not None and len(rows) > 0:
256		point_values_dict[point['point_id']] = dict()
257		for row in rows:
258		point_values_dict[point['point_id']][row[0]] = row[1]
259		else:
260		point_values_dict[point['point_id']] = None
261		else:
262		# point type should not be DIGITAL_VALUE
263		return "variable point type should not be DIGITAL_VALUE " + " for '" + virtual_point['name'] + "'"
264		except Exception as e:
265		if cursor_historical_db:
266		cursor_historical_db.close()
267		if cnx_historical_db:
268		cnx_historical_db.close()
269		return "Error in step 4.1 virtual point worker " + str(e) + " for '" + virtual_point['name'] + "'"
270
271		############################################################################################################
272		# Step 5: evaluate the equation with points values
273		############################################################################################################
274
275		print("getting date time set for all points")
276		utc_date_time_set = set()
277		if point_values_dict is not None and len(point_values_dict) > 0:
278		for point_id, point_values in point_values_dict.items():
279		if point_values is not None and len(point_values) > 0:
280		utc_date_time_set = utc_date_time_set.union(point_values.keys())
281
282		print("evaluating the equation with SymPy")
283		normalized_values = list()
284
285		############################################################################################################
286		# Converting Strings to SymPy Expressions
287		# The sympify function(that’s sympify, not to be confused with simplify) can be used to
288		# convert strings into SymPy expressions.
289		############################################################################################################
290		try:
291		if re.search(',', expression):
292		for item in substitutions.keys():
293		locals()[item] = symbols(item)
		0 ignored issues – show Comprehensibility Best Practice introduced 2022-08-04 02:30 UTC by Report Bug Copy Issue Report The variable `locals` does not seem to be defined. Loading history...
294		expr = eval(expression)
295		print("the expression will be evaluated as piecewise function: " + str(expr))
296		else:
297		expr = sympify(expression)
298		print("the expression will be evaluated as algebraic expression: " + str(expr))
299
300		for utc_date_time in utc_date_time_set:
301		meta_data = dict()
302		meta_data['utc_date_time'] = utc_date_time
303
304		####################################################################################################
305		# create a dictionary of Symbol: point pairs
306		####################################################################################################
307
308		subs = dict()
309
310		####################################################################################################
311		# Evaluating the expression at current_datetime_utc
312		####################################################################################################
313
314		if point_list is not None and len(point_list) > 0:
315		for point in point_list:
316		actual_value = point_values_dict[point['point_id']].get(utc_date_time, None)
317		if actual_value is None:
318		break
319		subs[point['variable_name']] = actual_value
320
321		if len(subs) != len(point_list):
322		continue
323
324		####################################################################################################
325		# To numerically evaluate an expression with a Symbol at a point,
326		# we might use subs followed by evalf,
327		# but it is more efficient and numerically stable to pass the substitution to evalf
328		# using the subs flag, which takes a dictionary of Symbol: point pairs.
329		####################################################################################################
330		if re.search(',', expression):
331		formula = Piecewise(*expr)
332		meta_data['actual_value'] = Decimal(str(formula.subs(subs)))
333		normalized_values.append(meta_data)
334		else:
335		meta_data['actual_value'] = Decimal(str(expr.evalf(subs=subs)))
336		normalized_values.append(meta_data)
337		except Exception as e:
338		if cursor_historical_db:
339		cursor_historical_db.close()
340		if cnx_historical_db:
341		cnx_historical_db.close()
342		return "Error in step 5.1 virtual point worker " + str(e) + " for '" + virtual_point['name'] + "'"
343
344		print("saving virtual points values to historical database")
345
346		if len(normalized_values) > 0:
347		latest_meta_data = normalized_values[0]
348
349		while len(normalized_values) > 0:
350		insert_100 = normalized_values[:100]
351		normalized_values = normalized_values[100:]
352
353		try:
354		add_values = (" INSERT INTO " + table_name +
355		" (point_id, utc_date_time, actual_value) "
356		" VALUES ")
357
358		for meta_data in insert_100:
359		add_values += " (" + str(virtual_point['id']) + ","
360		add_values += "'" + meta_data['utc_date_time'].isoformat()[0:19] + "',"
361		add_values += str(meta_data['actual_value']) + "), "
362
363		if meta_data['utc_date_time'] > latest_meta_data['utc_date_time']:
364		latest_meta_data = meta_data
365
366		# print("add_values:" + add_values)
367		# trim ", " at the end of string and then execute
368		cursor_historical_db.execute(add_values[:-2])
369		cnx_historical_db.commit()
370		except Exception as e:
371		if cursor_historical_db:
372		cursor_historical_db.close()
373		if cnx_historical_db:
374		cnx_historical_db.close()
375		return "Error in step 5.2 virtual point worker " + str(e) + " for '" + virtual_point['name'] + "'"
376
377		try:
378		# update tbl_analog_value_latest or tbl_energy_value_latest
379		delete_value = " DELETE FROM " + table_name + "_latest WHERE point_id = {} ".format(virtual_point['id'])
380		# print("delete_value:" + delete_value)
381		cursor_historical_db.execute(delete_value)
382		cnx_historical_db.commit()
383
384		latest_value = (" INSERT INTO " + table_name + "_latest (point_id, utc_date_time, actual_value) "
385		" VALUES ({}, '{}', {}) "
386		.format(virtual_point['id'],
387		latest_meta_data['utc_date_time'].isoformat()[0:19],
388		latest_meta_data['actual_value']))
389		# print("latest_value:" + latest_value)
390		cursor_historical_db.execute(latest_value)
391		cnx_historical_db.commit()
392		except Exception as e:
393		if cursor_historical_db:
394		cursor_historical_db.close()
395		if cnx_historical_db:
396		cnx_historical_db.close()
397		return "Error in step 5.3 virtual point worker " + str(e) + " for '" + virtual_point['name'] + "'"
398
399		if cursor_historical_db:
400		cursor_historical_db.close()
401		if cnx_historical_db:
402		cnx_historical_db.close()
403
404		return None
405

MyEMS / myems

virtualpoint.calculate() F last analyzed 2025-08-28 05:12 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

virtualpoint.calculate() F
last analyzed 2025-08-28 05:12 UTC