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

meter.worker() F
last analyzed 2020-12-23 04:06 UTC

↳ Parent: meter

Complexity

Conditions

Size

Total Lines	323
Code Lines	166

Duplication

Lines	21
Ratio	6.5 %

Importance

Changes

Metric	Value
eloc	166
dl	21
loc	323
rs	0
c	0
b	0
f	0
cc	51
nop	1

How to fix Long Method Complexity

import time
from datetime import datetime, timedelta, timezone
import mysql.connector
from multiprocessing import Pool
import random
from decimal import Decimal
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 = FALSE "
                 " 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 = FALSE "
                 " 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
    ####################################################################################################################
    if len(normalized_values) > 0:

        try:
            add_values = (" INSERT INTO tbl_meter_hourly (meter_id, start_datetime_utc, actual_value) "
                          " VALUES  ")

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

myems / myems-normalization

meter.worker() F last analyzed 2020-12-23 04:06 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 2020-12-23 04:06 UTC