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

meter.worker() F
last analyzed 2025-06-17 15:42 UTC

↳ Parent: meter

Complexity

Conditions

Size

Total Lines	325
Code Lines	168

Duplication

Lines	23
Ratio	7.08 %

Importance

Changes

Metric	Value
eloc	168
dl	23
loc	325
rs	0
c	0
b	0
f	0
cc	51
nop	1

How to fix Long Method Complexity

import random
import time
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from multiprocessing import Pool
import mysql.connector
import config


########################################################################################################################
# PROCEDURES:
# Step 1: Query all meters and associated energy value points
# Step 2: Create multiprocessing pool to call worker in parallel
########################################################################################################################


def calculate_hourly(logger):


    while True:
        ################################################################################################################
        # Step 1: Query all meters and associated energy value points
        ################################################################################################################
        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 meter.calculate_hourly process " + str(e))
            if cursor_system_db:
                cursor_system_db.close()
            if cnx_system_db:
                cnx_system_db.close()
            # sleep several minutes and continue the outer loop to reconnect the database
            time.sleep(60)
            continue

        print("Connected to the MyEMS System Database")

        try:
            cursor_system_db.execute(" SELECT m.id, m.name, m.hourly_low_limit, m.hourly_high_limit, "
                                     "        p.id as point_id, p.units "
                                     " FROM tbl_meters m, tbl_meters_points mp, tbl_points p "
                                     " WHERE m.id = mp.meter_id "
                                     "       AND mp.point_id = p.id "
                                     "       AND p.object_type = 'ENERGY_VALUE'")
            rows_meters = cursor_system_db.fetchall()

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

            meter_list = list()
            for row in rows_meters:
                meta_result = {"id": row[0],
                               "name": row[1],
                               "hourly_low_limit": row[2],
                               "hourly_high_limit": row[3],
                               "point_id": row[4],
                               "units": row[5]}

                meter_list.append(meta_result)

        except Exception as e:
            logger.error("Error in step 1.2 meter.calculate_hourly " + str(e))
            # sleep several minutes 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 meter list for randomly calculating the meter hourly value
        random.shuffle(meter_list)


        print("Got all meters 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, meter_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...")
    # end of outer while


########################################################################################################################
# PROCEDURES:
# Step 1: Determine the start datetime and end datetime
# Step 2: Get raw data from historical database between start_datetime_utc and end datetime
# Step 3: Normalize energy values by minutes_to_count
# Step 4: Insert into energy database
#
# NOTE: returns None or the error string because that the logger object cannot be passed in as parameter
########################################################################################################################

def worker(meter):
    print("Start to process meter: " + "'" + meter['name'] + "'")
    ####################################################################################################################
    # Step 1: Determine the start datetime and end datetime
    ####################################################################################################################
    cnx_energy_db = None
    cursor_energy_db = None
    try:
        cnx_energy_db = mysql.connector.connect(**config.myems_energy_db)
        cursor_energy_db = cnx_energy_db.cursor()
    except Exception as e:
        error_string = "Error in step 1.1 of meter.worker " + str(e) + " for '" + meter['name'] + "'"
        if cursor_energy_db:
            cursor_energy_db.close()
        if cnx_energy_db:
            cnx_energy_db.close()
        print(error_string)
        return error_string

    # get the initial start datetime from config file in case there is no energy data
    start_datetime_utc = datetime.strptime(config.start_datetime_utc, '%Y-%m-%d %H:%M:%S')
    start_datetime_utc = start_datetime_utc.replace(tzinfo=timezone.utc)
    start_datetime_utc = start_datetime_utc.replace(minute=0, second=0, microsecond=0)

    try:
        query = (" SELECT MAX(start_datetime_utc) "
                 " FROM tbl_meter_hourly "
                 " WHERE meter_id = %s ")
        cursor_energy_db.execute(query, (meter['id'],))
        row_datetime = cursor_energy_db.fetchone()
    except Exception as e:
        error_string = "Error in step 1.3 of meter.worker " + str(e) + " for '" + meter['name'] + "'"
        if cursor_energy_db:
            cursor_energy_db.close()
        if cnx_energy_db:
            cnx_energy_db.close()
        print(error_string)
        return error_string

    if row_datetime is not None and len(row_datetime) > 0 and isinstance(row_datetime[0], datetime):
        start_datetime_utc = row_datetime[0].replace(tzinfo=timezone.utc)
        # replace second and microsecond with 0
        # NOTE: DO NOT replace minute in case of calculating in half hourly
        start_datetime_utc = start_datetime_utc.replace(second=0, microsecond=0)
        # start from the next time slot
        start_datetime_utc += timedelta(minutes=config.minutes_to_count)

    end_datetime_utc = datetime.utcnow().replace(tzinfo=timezone.utc)
    # we should allow myems-cleaning service to take at most [minutes_to_clean] minutes to clean the data
    end_datetime_utc -= timedelta(minutes=config.minutes_to_clean)

    time_difference = end_datetime_utc - start_datetime_utc
    time_difference_in_minutes = time_difference / timedelta(minutes=1)
    if time_difference_in_minutes < config.minutes_to_count:
        error_string = "it's too early to calculate" + " for '" + meter['name'] + "'"
        print(error_string)
        return error_string

    # trim end_datetime_utc
    trimmed_end_datetime_utc = start_datetime_utc + timedelta(minutes=config.minutes_to_count)
    while trimmed_end_datetime_utc <= end_datetime_utc:
        trimmed_end_datetime_utc += timedelta(minutes=config.minutes_to_count)

    end_datetime_utc = trimmed_end_datetime_utc - timedelta(minutes=config.minutes_to_count)

    if end_datetime_utc <= start_datetime_utc:
        error_string = "it's too early to calculate" + " for '" + meter['name'] + "'"
        print(error_string)
        return error_string

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

    ####################################################################################################################
    # Step 2: Get raw data from historical database between start_datetime_utc and end_datetime_utc
    ####################################################################################################################

    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:
        error_string = "Error in step 1.2 of meter.worker " + str(e) + " for '" + meter['name'] + "'"
        if cursor_historical_db:
            cursor_historical_db.close()
        if cnx_historical_db:
            cnx_historical_db.close()

        if cursor_energy_db:
            cursor_energy_db.close()
        if cnx_energy_db:
            cnx_energy_db.close()

        print(error_string)
        return error_string

    # query latest record before start_datetime_utc
    energy_value_just_before_start = dict()
    try:
        query = (" SELECT utc_date_time, actual_value "
                 " FROM tbl_energy_value "
                 " WHERE point_id = %s AND utc_date_time < %s AND is_bad = 0 "
                 " ORDER BY utc_date_time DESC "
                 " LIMIT 1 ")
        cursor_historical_db.execute(query, (meter['point_id'], start_datetime_utc,))
        row_energy_value_before_start = cursor_historical_db.fetchone()

        if row_energy_value_before_start is not None and len(row_energy_value_before_start) > 0:
            energy_value_just_before_start = {"utc_date_time": row_energy_value_before_start[0],
                                              "actual_value": row_energy_value_before_start[1]}
    except Exception as e:
        error_string = "Error in step 2.2 of meter.worker " + str(e) + " for '" + meter['name'] + "'"
        if cursor_historical_db:
            cursor_historical_db.close()
        if cnx_historical_db:
            cnx_historical_db.close()

        if cursor_energy_db:
            cursor_energy_db.close()
        if cnx_energy_db:
            cnx_energy_db.close()

        print(error_string)
        return error_string

    # query energy values to be normalized
    try:
        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 AND is_bad = 0 "
                 " ORDER BY utc_date_time ")
        cursor_historical_db.execute(query, (meter['point_id'], start_datetime_utc, end_datetime_utc))
        rows_energy_values = cursor_historical_db.fetchall()
    except Exception as e:
        error_string = "Error in step 2.3 of meter.worker " + str(e) + " for '" + meter['name'] + "'"

        if cursor_energy_db:
            cursor_energy_db.close()
        if cnx_energy_db:
            cnx_energy_db.close()

        print(error_string)
        return error_string
    finally:
        if cursor_historical_db:
            cursor_historical_db.close()
        if cnx_historical_db:
            cnx_historical_db.close()

    ####################################################################################################################
    # Step 3: Normalize energy values by minutes_to_count
    ####################################################################################################################

    ####################################################################################################################
    # special test case 1 (disconnected)
    # id       point_id  utc_date_time        actual_value
    # '878152', '3315', '2016-12-05 23:58:46', '38312088'
    # '878183', '3315', '2016-12-05 23:59:48', '38312088'
    # '878205', '3315', '2016-12-06 06:14:49', '38315900'
    # '878281', '3315', '2016-12-06 06:15:50', '38315928'
    # '878357', '3315', '2016-12-06 06:16:52', '38315928'
    ####################################################################################################################

    ####################################################################################################################
    # special test case 2 (a new added used meter)
    # id,         point_id,  utc_date_time,      actual_value
    # '19070111', '1734', '2017-03-27 02:36:07', '56842220.77297248'
    # '19069943', '1734', '2017-03-27 02:35:04', '56842208.420127675'
    # '19069775', '1734', '2017-03-27 02:34:01', '56842195.95270827'
    # '19069608', '1734', '2017-03-27 02:32:58', '56842183.48610827'
    # '19069439', '1734', '2017-03-27 02:31:53', '56842170.812365524'
    # '19069270', '1734', '2017-03-27 02:30:48', '56842157.90797222'
    # null,       null,   null,                , null

    ####################################################################################################################

    ####################################################################################################################
    # special test case 3 (hi_limit exceeded)
    # id       point_id  utc_date_time        actual_value
    # '3230282', '3336', '2016-12-24 08:26:14', '999984.0625'
    # '3230401', '3336', '2016-12-24 08:27:15', '999984.0625'
    # '3230519', '3336', '2016-12-24 08:28:17', '999984.0625'
    # '3230638', '3336', '2016-12-24 08:29:18', '20'
    # '3230758', '3336', '2016-12-24 08:30:20', '20'
    # '3230878', '3336', '2016-12-24 08:31:21', '20'
    ####################################################################################################################

    ####################################################################################################################
    # test case 4 (recovered from bad zeroes)
    # id      point_id  utc_date_time       actual_value is_bad
    # 300366736	1003344	2019-03-14 02:03:20	1103860.625
    # 300366195	1003344	2019-03-14 02:02:19	1103845
    # 300365654	1003344	2019-03-14 02:01:19	1103825.5
    # 300365106	1003344	2019-03-14 02:00:18	1103804.25
    # 300364562	1003344	2019-03-14 01:59:17	1103785.625
    # 300364021	1003344	2019-03-14 01:58:17	1103770.875
    # 300363478	1003344	2019-03-14 01:57:16	1103755.125
    # 300362936	1003344	2019-03-14 01:56:16	1103739.375
    # 300362393	1003344	2019-03-14 01:55:15	1103720.625
    # 300361851	1003344	2019-03-14 01:54:15	1103698.125
    # 300361305	1003344	2019-03-14 01:53:14	1103674.75
    # 300360764	1003344	2019-03-14 01:52:14	1103649
    # 300360221	1003344	2019-03-14 01:51:13	1103628.25
    # 300359676	1003344	2019-03-14 01:50:13	1103608.625
    # 300359133	1003344	2019-03-14 01:49:12	1103586.75
    # 300358592	1003344	2019-03-14 01:48:12	1103564
    # 300358050	1003344	2019-03-14 01:47:11	1103542
    # 300357509	1003344	2019-03-14 01:46:11	1103520.625
    # 300356966	1003344	2019-03-14 01:45:10	1103499.375
    # 300356509	1003344	2019-03-14 01:44:10	1103478.25
    # 300355964	1003344	2019-03-14 01:43:09	1103456.25
    # 300355419	1003344	2019-03-14 01:42:09	1103435.5
    # 300354878	1003344	2019-03-14 01:41:08	1103414.625
    # 300354335	1003344	2019-03-14 01:40:08	1103391.875
    # 300353793	1003344	2019-03-14 01:39:07	1103373
    # 300353248	1003344	2019-03-14 01:38:07	1103349
    # 300352705	1003344	2019-03-14 01:37:06	1103325.75
    # 300352163	1003344	2019-03-14 01:36:06	0	            1
    # 300351621	1003344	2019-03-14 01:35:05	0	            1
    # 300351080	1003344	2019-03-14 01:34:05	0	            1
    # 300350532	1003344	2019-03-14 01:33:04	0	            1
    # 300349988	1003344	2019-03-14 01:32:04	0	            1
    # 300349446	1003344	2019-03-14 01:31:03	0	            1
    # 300348903	1003344	2019-03-14 01:30:02	0	            1
    # 300348359	1003344	2019-03-14 01:29:02	0	            1
    # 300347819	1003344	2019-03-14 01:28:01	0	            1
    # 300347277	1003344	2019-03-14 01:27:01	0	            1
    # 300346733	1003344	2019-03-14 01:26:00	0	            1
    # 300346191	1003344	2019-03-14 01:25:00	0	            1
    ####################################################################################################################

    normalized_values = list()
    if rows_energy_values is None or len(rows_energy_values) == 0:

        # NOTE: there isn't any value to be normalized
        # that means the meter is offline or all values are bad
        current_datetime_utc = start_datetime_utc
        while current_datetime_utc < end_datetime_utc:
            normalized_values.append({'start_datetime_utc': current_datetime_utc, 'actual_value': Decimal(0.0)})
            current_datetime_utc += timedelta(minutes=config.minutes_to_count)
    else:
        maximum = Decimal(0.0)
        if energy_value_just_before_start is not None and \
                len(energy_value_just_before_start) > 0 and \
                energy_value_just_before_start['actual_value'] > Decimal(0.0):
            maximum = energy_value_just_before_start['actual_value']

        current_datetime_utc = start_datetime_utc
        while current_datetime_utc < end_datetime_utc:
            initial_maximum = maximum
            # get all energy values in current time slot
            current_energy_values = list()
            while len(rows_energy_values) > 0:
                row_energy_value = rows_energy_values.pop(0)
                energy_value_datetime = row_energy_value[0].replace(tzinfo=timezone.utc)
                if energy_value_datetime < current_datetime_utc + timedelta(minutes=config.minutes_to_count):
                    current_energy_values.append(row_energy_value)
                else:
                    rows_energy_values.insert(0, row_energy_value)
                    break

            # get the energy increment one by one in current time slot
            increment = Decimal(0.0)
            # maximum should be equal to the maximum value of last time here
            for index in range(len(current_energy_values)):
                current_energy_value = current_energy_values[index]
                if maximum < current_energy_value[1]:
                    increment += current_energy_value[1] - maximum
                maximum = current_energy_value[1]

            # omit huge initial value for a new meter
            # or omit huge value for a recovered meter with zero values during failure
            # NOTE: this method may cause the lose of energy consumption in this time slot
            if initial_maximum <= Decimal(0.1):
                increment = Decimal(0.0)

            # check with hourly low limit
            if increment < meter['hourly_low_limit']:
                increment = Decimal(0.0)

            # check with hourly high limit
            # NOTE: this method may cause the lose of energy consumption in this time slot
            if increment > meter['hourly_high_limit']:
                increment = Decimal(0.0)

            meta_data = {'start_datetime_utc': current_datetime_utc,
                         'actual_value': increment}
            # append mete_data
            normalized_values.append(meta_data)
            current_datetime_utc += timedelta(minutes=config.minutes_to_count)

    ####################################################################################################################
    # Step 4: Insert into energy database
    ####################################################################################################################
    while len(normalized_values) > 0:

        insert_100 = normalized_values[:100]
        normalized_values = normalized_values[100:]
        try:
            add_values = (" INSERT INTO tbl_meter_hourly (meter_id, start_datetime_utc, actual_value) "
                          " VALUES  ")

            for meta_data in insert_100:
                add_values += " (" + str(meter['id']) + ","
                add_values += "'" + meta_data['start_datetime_utc'].isoformat()[0:19] + "',"
                add_values += str(meta_data['actual_value']) + "), "
            # trim ", " at the end of string and then execute
            cursor_energy_db.execute(add_values[:-2])
            cnx_energy_db.commit()
        except Exception as e:
            error_string = "Error in step 4.1 of meter.worker " + str(e) + " for '" + meter['name'] + "'"
            if cursor_energy_db:
                cursor_energy_db.close()
            if cnx_energy_db:
                cnx_energy_db.close()

            print(error_string)
            return error_string

    if cursor_energy_db:
        cursor_energy_db.close()
    if cnx_energy_db:
        cnx_energy_db.close()

    print("End of processing meter: " + "'" + meter['name'] + "'")
    return None


1		import random
2		import time
3		from datetime import datetime, timedelta, timezone
4		from decimal import Decimal
5		from multiprocessing import Pool
6		import mysql.connector
7		import config
8
9
10		########################################################################################################################
11		# PROCEDURES:
12		# Step 1: Query all meters and associated energy value points
13		# Step 2: Create multiprocessing pool to call worker in parallel
14		########################################################################################################################
15
16
17	View Code Duplication	def calculate_hourly(logger):
		0 ignored issues – show Duplication introduced 2021-05-08 05:56 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
18
19		while True:
20		################################################################################################################
21		# Step 1: Query all meters and associated energy value points
22		################################################################################################################
23		cnx_system_db = None
24		cursor_system_db = None
25		try:
26		cnx_system_db = mysql.connector.connect(**config.myems_system_db)
27		cursor_system_db = cnx_system_db.cursor()
28		except Exception as e:
29		logger.error("Error in step 1.1 of meter.calculate_hourly process " + str(e))
30		if cursor_system_db:
31		cursor_system_db.close()
32		if cnx_system_db:
33		cnx_system_db.close()
34		# sleep several minutes and continue the outer loop to reconnect the database
35		time.sleep(60)
36		continue
37
38		print("Connected to the MyEMS System Database")
39
40		try:
41		cursor_system_db.execute(" SELECT m.id, m.name, m.hourly_low_limit, m.hourly_high_limit, "
42		" p.id as point_id, p.units "
43		" FROM tbl_meters m, tbl_meters_points mp, tbl_points p "
44		" WHERE m.id = mp.meter_id "
45		" AND mp.point_id = p.id "
46		" AND p.object_type = 'ENERGY_VALUE'")
47		rows_meters = cursor_system_db.fetchall()
48
49		if rows_meters is None or len(rows_meters) == 0:
50		# sleep several minutes and continue the outer loop to reconnect the database
51		time.sleep(60)
52		continue
53
54		meter_list = list()
55		for row in rows_meters:
56		meta_result = {"id": row[0],
57		"name": row[1],
58		"hourly_low_limit": row[2],
59		"hourly_high_limit": row[3],
60		"point_id": row[4],
61		"units": row[5]}
62
63		meter_list.append(meta_result)
64
65		except Exception as e:
66		logger.error("Error in step 1.2 meter.calculate_hourly " + str(e))
67		# sleep several minutes and continue the outer loop to reconnect the database
68		time.sleep(60)
69		continue
70		finally:
71		if cursor_system_db:
72		cursor_system_db.close()
73		if cnx_system_db:
74		cnx_system_db.close()
75
76		# shuffle the meter list for randomly calculating the meter hourly value
77		random.shuffle(meter_list)
		0 ignored issues – show introduced 2021-03-02 03:54 UTC by Report Bug Copy Issue Report The variable `meter_list` does not seem to be defined for all execution paths. Loading history...
78
79		print("Got all meters in MyEMS System Database")
80
81		################################################################################################################
82		# Step 2: Create multiprocessing pool to call worker in parallel
83		################################################################################################################
84		p = Pool(processes=config.pool_size)
85		error_list = p.map(worker, meter_list)
86		p.close()
87		p.join()
88
89		for error in error_list:
90		if error is not None and len(error) > 0:
91		logger.error(error)
92
93		print("go to sleep ...")
94		time.sleep(60)
95		print("wake from sleep, and continue to work...")
96		# end of outer while
97
98
99		########################################################################################################################
100		# PROCEDURES:
101		# Step 1: Determine the start datetime and end datetime
102		# Step 2: Get raw data from historical database between start_datetime_utc and end datetime
103		# Step 3: Normalize energy values by minutes_to_count
104		# Step 4: Insert into energy database
105		#
106		# NOTE: returns None or the error string because that the logger object cannot be passed in as parameter
107		########################################################################################################################
108
109		def worker(meter):
110		print("Start to process meter: " + "'" + meter['name'] + "'")
111		####################################################################################################################
112		# Step 1: Determine the start datetime and end datetime
113		####################################################################################################################
114		cnx_energy_db = None
115		cursor_energy_db = None
116		try:
117		cnx_energy_db = mysql.connector.connect(**config.myems_energy_db)
118		cursor_energy_db = cnx_energy_db.cursor()
119		except Exception as e:
120		error_string = "Error in step 1.1 of meter.worker " + str(e) + " for '" + meter['name'] + "'"
121		if cursor_energy_db:
122		cursor_energy_db.close()
123		if cnx_energy_db:
124		cnx_energy_db.close()
125		print(error_string)
126		return error_string
127
128		# get the initial start datetime from config file in case there is no energy data
129		start_datetime_utc = datetime.strptime(config.start_datetime_utc, '%Y-%m-%d %H:%M:%S')
130		start_datetime_utc = start_datetime_utc.replace(tzinfo=timezone.utc)
131		start_datetime_utc = start_datetime_utc.replace(minute=0, second=0, microsecond=0)
132
133		try:
134		query = (" SELECT MAX(start_datetime_utc) "
135		" FROM tbl_meter_hourly "
136		" WHERE meter_id = %s ")
137		cursor_energy_db.execute(query, (meter['id'],))
138		row_datetime = cursor_energy_db.fetchone()
139		except Exception as e:
140		error_string = "Error in step 1.3 of meter.worker " + str(e) + " for '" + meter['name'] + "'"
141		if cursor_energy_db:
142		cursor_energy_db.close()
143		if cnx_energy_db:
144		cnx_energy_db.close()
145		print(error_string)
146		return error_string
147
148		if row_datetime is not None and len(row_datetime) > 0 and isinstance(row_datetime[0], datetime):
149		start_datetime_utc = row_datetime[0].replace(tzinfo=timezone.utc)
150		# replace second and microsecond with 0
151		# NOTE: DO NOT replace minute in case of calculating in half hourly
152		start_datetime_utc = start_datetime_utc.replace(second=0, microsecond=0)
153		# start from the next time slot
154		start_datetime_utc += timedelta(minutes=config.minutes_to_count)
155
156		end_datetime_utc = datetime.utcnow().replace(tzinfo=timezone.utc)
157		# we should allow myems-cleaning service to take at most [minutes_to_clean] minutes to clean the data
158		end_datetime_utc -= timedelta(minutes=config.minutes_to_clean)
159
160		time_difference = end_datetime_utc - start_datetime_utc
161		time_difference_in_minutes = time_difference / timedelta(minutes=1)
162		if time_difference_in_minutes < config.minutes_to_count:
163		error_string = "it's too early to calculate" + " for '" + meter['name'] + "'"
164		print(error_string)
165		return error_string
166
167		# trim end_datetime_utc
168		trimmed_end_datetime_utc = start_datetime_utc + timedelta(minutes=config.minutes_to_count)
169		while trimmed_end_datetime_utc <= end_datetime_utc:
170		trimmed_end_datetime_utc += timedelta(minutes=config.minutes_to_count)
171
172		end_datetime_utc = trimmed_end_datetime_utc - timedelta(minutes=config.minutes_to_count)
173
174		if end_datetime_utc <= start_datetime_utc:
175		error_string = "it's too early to calculate" + " for '" + meter['name'] + "'"
176		print(error_string)
177		return error_string
178
179		print("start_datetime_utc: " + start_datetime_utc.isoformat()[0:19]
180		+ "end_datetime_utc: " + end_datetime_utc.isoformat()[0:19])
181
182		####################################################################################################################
183		# Step 2: Get raw data from historical database between start_datetime_utc and end_datetime_utc
184		####################################################################################################################
185
186		cnx_historical_db = None
187		cursor_historical_db = None
188		try:
189		cnx_historical_db = mysql.connector.connect(**config.myems_historical_db)
190		cursor_historical_db = cnx_historical_db.cursor()
191		except Exception as e:
192		error_string = "Error in step 1.2 of meter.worker " + str(e) + " for '" + meter['name'] + "'"
193		if cursor_historical_db:
194		cursor_historical_db.close()
195		if cnx_historical_db:
196		cnx_historical_db.close()
197
198		if cursor_energy_db:
199		cursor_energy_db.close()
200		if cnx_energy_db:
201		cnx_energy_db.close()
202
203		print(error_string)
204		return error_string
205
206		# query latest record before start_datetime_utc
207		energy_value_just_before_start = dict()
208		try:
209		query = (" SELECT utc_date_time, actual_value "
210		" FROM tbl_energy_value "
211		" WHERE point_id = %s AND utc_date_time < %s AND is_bad = 0 "
212		" ORDER BY utc_date_time DESC "
213		" LIMIT 1 ")
214		cursor_historical_db.execute(query, (meter['point_id'], start_datetime_utc,))
215		row_energy_value_before_start = cursor_historical_db.fetchone()
216
217		if row_energy_value_before_start is not None and len(row_energy_value_before_start) > 0:
218		energy_value_just_before_start = {"utc_date_time": row_energy_value_before_start[0],
219		"actual_value": row_energy_value_before_start[1]}
220		except Exception as e:
221		error_string = "Error in step 2.2 of meter.worker " + str(e) + " for '" + meter['name'] + "'"
222		if cursor_historical_db:
223		cursor_historical_db.close()
224		if cnx_historical_db:
225		cnx_historical_db.close()
226
227		if cursor_energy_db:
228		cursor_energy_db.close()
229		if cnx_energy_db:
230		cnx_energy_db.close()
231
232		print(error_string)
233		return error_string
234
235		# query energy values to be normalized
236		try:
237		query = (" SELECT utc_date_time, actual_value "
238		" FROM tbl_energy_value "
239		" WHERE point_id = %s AND utc_date_time >= %s AND utc_date_time < %s AND is_bad = 0 "
240		" ORDER BY utc_date_time ")
241		cursor_historical_db.execute(query, (meter['point_id'], start_datetime_utc, end_datetime_utc))
242		rows_energy_values = cursor_historical_db.fetchall()
243		except Exception as e:
244		error_string = "Error in step 2.3 of meter.worker " + str(e) + " for '" + meter['name'] + "'"
245
246		if cursor_energy_db:
247		cursor_energy_db.close()
248		if cnx_energy_db:
249		cnx_energy_db.close()
250
251		print(error_string)
252		return error_string
253		finally:
254		if cursor_historical_db:
255		cursor_historical_db.close()
256		if cnx_historical_db:
257		cnx_historical_db.close()
258
259		####################################################################################################################
260		# Step 3: Normalize energy values by minutes_to_count
261		####################################################################################################################
262
263		####################################################################################################################
264		# special test case 1 (disconnected)
265		# id point_id utc_date_time actual_value
266		# '878152', '3315', '2016-12-05 23:58:46', '38312088'
267		# '878183', '3315', '2016-12-05 23:59:48', '38312088'
268		# '878205', '3315', '2016-12-06 06:14:49', '38315900'
269		# '878281', '3315', '2016-12-06 06:15:50', '38315928'
270		# '878357', '3315', '2016-12-06 06:16:52', '38315928'
271		####################################################################################################################
272
273		####################################################################################################################
274		# special test case 2 (a new added used meter)
275		# id, point_id, utc_date_time, actual_value
276		# '19070111', '1734', '2017-03-27 02:36:07', '56842220.77297248'
277		# '19069943', '1734', '2017-03-27 02:35:04', '56842208.420127675'
278		# '19069775', '1734', '2017-03-27 02:34:01', '56842195.95270827'
279		# '19069608', '1734', '2017-03-27 02:32:58', '56842183.48610827'
280		# '19069439', '1734', '2017-03-27 02:31:53', '56842170.812365524'
281		# '19069270', '1734', '2017-03-27 02:30:48', '56842157.90797222'
282		# null, null, null, , null
283
284		####################################################################################################################
285
286		####################################################################################################################
287		# special test case 3 (hi_limit exceeded)
288		# id point_id utc_date_time actual_value
289		# '3230282', '3336', '2016-12-24 08:26:14', '999984.0625'
290		# '3230401', '3336', '2016-12-24 08:27:15', '999984.0625'
291		# '3230519', '3336', '2016-12-24 08:28:17', '999984.0625'
292		# '3230638', '3336', '2016-12-24 08:29:18', '20'
293		# '3230758', '3336', '2016-12-24 08:30:20', '20'
294		# '3230878', '3336', '2016-12-24 08:31:21', '20'
295		####################################################################################################################
296
297		####################################################################################################################
298		# test case 4 (recovered from bad zeroes)
299		# id point_id utc_date_time actual_value is_bad
300		# 300366736 1003344 2019-03-14 02:03:20 1103860.625
301		# 300366195 1003344 2019-03-14 02:02:19 1103845
302		# 300365654 1003344 2019-03-14 02:01:19 1103825.5
303		# 300365106 1003344 2019-03-14 02:00:18 1103804.25
304		# 300364562 1003344 2019-03-14 01:59:17 1103785.625
305		# 300364021 1003344 2019-03-14 01:58:17 1103770.875
306		# 300363478 1003344 2019-03-14 01:57:16 1103755.125
307		# 300362936 1003344 2019-03-14 01:56:16 1103739.375
308		# 300362393 1003344 2019-03-14 01:55:15 1103720.625
309		# 300361851 1003344 2019-03-14 01:54:15 1103698.125
310		# 300361305 1003344 2019-03-14 01:53:14 1103674.75
311		# 300360764 1003344 2019-03-14 01:52:14 1103649
312		# 300360221 1003344 2019-03-14 01:51:13 1103628.25
313		# 300359676 1003344 2019-03-14 01:50:13 1103608.625
314		# 300359133 1003344 2019-03-14 01:49:12 1103586.75
315		# 300358592 1003344 2019-03-14 01:48:12 1103564
316		# 300358050 1003344 2019-03-14 01:47:11 1103542
317		# 300357509 1003344 2019-03-14 01:46:11 1103520.625
318		# 300356966 1003344 2019-03-14 01:45:10 1103499.375
319		# 300356509 1003344 2019-03-14 01:44:10 1103478.25
320		# 300355964 1003344 2019-03-14 01:43:09 1103456.25
321		# 300355419 1003344 2019-03-14 01:42:09 1103435.5
322		# 300354878 1003344 2019-03-14 01:41:08 1103414.625
323		# 300354335 1003344 2019-03-14 01:40:08 1103391.875
324		# 300353793 1003344 2019-03-14 01:39:07 1103373
325		# 300353248 1003344 2019-03-14 01:38:07 1103349
326		# 300352705 1003344 2019-03-14 01:37:06 1103325.75
327		# 300352163 1003344 2019-03-14 01:36:06 0 1
328		# 300351621 1003344 2019-03-14 01:35:05 0 1
329		# 300351080 1003344 2019-03-14 01:34:05 0 1
330		# 300350532 1003344 2019-03-14 01:33:04 0 1
331		# 300349988 1003344 2019-03-14 01:32:04 0 1
332		# 300349446 1003344 2019-03-14 01:31:03 0 1
333		# 300348903 1003344 2019-03-14 01:30:02 0 1
334		# 300348359 1003344 2019-03-14 01:29:02 0 1
335		# 300347819 1003344 2019-03-14 01:28:01 0 1
336		# 300347277 1003344 2019-03-14 01:27:01 0 1
337		# 300346733 1003344 2019-03-14 01:26:00 0 1
338		# 300346191 1003344 2019-03-14 01:25:00 0 1
339		####################################################################################################################
340
341		normalized_values = list()
342		if rows_energy_values is None or len(rows_energy_values) == 0:
		0 ignored issues – show introduced 2021-03-02 03:54 UTC by Report Bug Copy Issue Report The variable `rows_energy_values` does not seem to be defined for all execution paths. Loading history...
343		# NOTE: there isn't any value to be normalized
344		# that means the meter is offline or all values are bad
345		current_datetime_utc = start_datetime_utc
346		while current_datetime_utc < end_datetime_utc:
347		normalized_values.append({'start_datetime_utc': current_datetime_utc, 'actual_value': Decimal(0.0)})
348		current_datetime_utc += timedelta(minutes=config.minutes_to_count)
349		else:
350		maximum = Decimal(0.0)
351		if energy_value_just_before_start is not None and \
352		len(energy_value_just_before_start) > 0 and \
353		energy_value_just_before_start['actual_value'] > Decimal(0.0):
354		maximum = energy_value_just_before_start['actual_value']
355
356		current_datetime_utc = start_datetime_utc
357		while current_datetime_utc < end_datetime_utc:
358		initial_maximum = maximum
359		# get all energy values in current time slot
360		current_energy_values = list()
361		while len(rows_energy_values) > 0:
362		row_energy_value = rows_energy_values.pop(0)
363		energy_value_datetime = row_energy_value[0].replace(tzinfo=timezone.utc)
364		if energy_value_datetime < current_datetime_utc + timedelta(minutes=config.minutes_to_count):
365		current_energy_values.append(row_energy_value)
366		else:
367		rows_energy_values.insert(0, row_energy_value)
368		break
369
370		# get the energy increment one by one in current time slot
371		increment = Decimal(0.0)
372		# maximum should be equal to the maximum value of last time here
373		for index in range(len(current_energy_values)):
374		current_energy_value = current_energy_values[index]
375		if maximum < current_energy_value[1]:
376		increment += current_energy_value[1] - maximum
377		maximum = current_energy_value[1]
378
379		# omit huge initial value for a new meter
380		# or omit huge value for a recovered meter with zero values during failure
381		# NOTE: this method may cause the lose of energy consumption in this time slot
382		if initial_maximum <= Decimal(0.1):
383		increment = Decimal(0.0)
384
385		# check with hourly low limit
386		if increment < meter['hourly_low_limit']:
387		increment = Decimal(0.0)
388
389		# check with hourly high limit
390		# NOTE: this method may cause the lose of energy consumption in this time slot
391		if increment > meter['hourly_high_limit']:
392		increment = Decimal(0.0)
393
394		meta_data = {'start_datetime_utc': current_datetime_utc,
395		'actual_value': increment}
396		# append mete_data
397		normalized_values.append(meta_data)
398		current_datetime_utc += timedelta(minutes=config.minutes_to_count)
399
400		####################################################################################################################
401		# Step 4: Insert into energy database
402		####################################################################################################################
403	View Code Duplication	while len(normalized_values) > 0:
		0 ignored issues – show Duplication introduced 2021-03-02 03:54 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
404		insert_100 = normalized_values[:100]
405		normalized_values = normalized_values[100:]
406		try:
407		add_values = (" INSERT INTO tbl_meter_hourly (meter_id, start_datetime_utc, actual_value) "
408		" VALUES ")
409
410		for meta_data in insert_100:
411		add_values += " (" + str(meter['id']) + ","
412		add_values += "'" + meta_data['start_datetime_utc'].isoformat()[0:19] + "',"
413		add_values += str(meta_data['actual_value']) + "), "
414		# trim ", " at the end of string and then execute
415		cursor_energy_db.execute(add_values[:-2])
416		cnx_energy_db.commit()
417		except Exception as e:
418		error_string = "Error in step 4.1 of meter.worker " + str(e) + " for '" + meter['name'] + "'"
419		if cursor_energy_db:
420		cursor_energy_db.close()
421		if cnx_energy_db:
422		cnx_energy_db.close()
423
424		print(error_string)
425		return error_string
426
427		if cursor_energy_db:
428		cursor_energy_db.close()
429		if cnx_energy_db:
430		cnx_energy_db.close()
431
432		print("End of processing meter: " + "'" + meter['name'] + "'")
433		return None
434

MyEMS / myems

meter.worker() F last analyzed 2025-06-17 15:42 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

meter.worker() F
last analyzed 2025-06-17 15:42 UTC