reports.meterrealtime.Reporting.on_get() - Code Metrics - MyEMS/myems - Measure and Improve Code Quality continuously with Scrutinizer

reports.meterrealtime.Reporting.on_get() F
last analyzed 2025-06-17 15:42 UTC

↳ Parent: reports.meterrealtime

Complexity

Conditions

Size

Total Lines	211
Code Lines	148

Duplication

Lines	67
Ratio	31.75 %

Importance

Changes

Metric	Value
eloc	148
dl	67
loc	211
rs	0
c	0
b	0
f	0
cc	43
nop	2

How to fix Long Method Complexity

import re
from datetime import datetime, timedelta, timezone
import falcon
import mysql.connector
import simplejson as json
import config
from core.useractivity import access_control, api_key_control


class Reporting:
    def __init__(self):
        """"Initializes Reporting"""
        pass

    @staticmethod
    def on_options(req, resp):
        _ = req
        resp.status = falcon.HTTP_200

    ####################################################################################################################
    # PROCEDURES
    # Step 1: valid parameters
    # Step 2: query the meter and energy category
    # Step 3: query associated points
    # Step 4: query reporting period points trends
    # Step 5: query tariff data
    # Step 6: construct the report
    ####################################################################################################################
    @staticmethod
    def on_get(req, resp):
        if 'API-KEY' not in req.headers or \
                not isinstance(req.headers['API-KEY'], str) or \
                len(str.strip(req.headers['API-KEY'])) == 0:
            access_control(req)
        else:
            api_key_control(req)
        print(req.params)
        meter_id = req.params.get('meterid')
        meter_uuid = req.params.get('meteruuid')
        quick_mode = req.params.get('quickmode')

        ################################################################################################################
        # Step 1: valid parameters
        ################################################################################################################
        if meter_id is None and meter_uuid is None:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST', description='API.INVALID_METER_ID')

        if meter_id is not None:
            meter_id = str.strip(meter_id)
            if not meter_id.isdigit() or int(meter_id) <= 0:
                raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                       description='API.INVALID_METER_ID')

        if meter_uuid is not None:
            regex = re.compile(r'^[a-f0-9]{8}-?[a-f0-9]{4}-?4[a-f0-9]{3}-?[89ab][a-f0-9]{3}-?[a-f0-9]{12}\Z', re.I)
            match = regex.match(str.strip(meter_uuid))
            if not bool(match):
                raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                       description='API.INVALID_METER_UUID')

        # if turn quick mode on, do not return parameters data and excel file
        is_quick_mode = False
        if quick_mode is not None and \
                len(str.strip(quick_mode)) > 0 and \
                str.lower(str.strip(quick_mode)) in ('true', 't', 'on', 'yes', 'y'):
            is_quick_mode = True

        timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
        if config.utc_offset[0] == '-':
            timezone_offset = -timezone_offset

        reporting_end_datetime_utc = datetime.utcnow()
        reporting_start_datetime_utc = reporting_end_datetime_utc - timedelta(minutes=60)
        ################################################################################################################
        # Step 2: query the meter and energy category
        ################################################################################################################
        cnx_system = mysql.connector.connect(**config.myems_system_db)
        cursor_system = cnx_system.cursor()

        cnx_historical = mysql.connector.connect(**config.myems_historical_db)
        cursor_historical = cnx_historical.cursor()

        if meter_id is not None:
            cursor_system.execute(" SELECT m.id, m.name, m.cost_center_id, m.energy_category_id, "
                                  "        ec.name, ec.unit_of_measure, ec.kgce, ec.kgco2e "
                                  " FROM tbl_meters m, tbl_energy_categories ec "
                                  " WHERE m.id = %s AND m.energy_category_id = ec.id ", (meter_id,))
            row_meter = cursor_system.fetchone()
        elif meter_uuid is not None:
            cursor_system.execute(" SELECT m.id, m.name, m.cost_center_id, m.energy_category_id, "
                                  "        ec.name, ec.unit_of_measure, ec.kgce, ec.kgco2e "
                                  " FROM tbl_meters m, tbl_energy_categories ec "
                                  " WHERE m.uuid = %s AND m.energy_category_id = ec.id ", (meter_uuid,))
            row_meter = cursor_system.fetchone()

        if row_meter is None:

            if cursor_system:
                cursor_system.close()
            if cnx_system:
                cnx_system.close()

            if cursor_historical:
                cursor_historical.close()
            if cnx_historical:
                cnx_historical.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND', description='API.METER_NOT_FOUND')

        meter = dict()
        meter['id'] = row_meter[0]
        meter['name'] = row_meter[1]
        meter['cost_center_id'] = row_meter[2]
        meter['energy_category_id'] = row_meter[3]
        meter['energy_category_name'] = row_meter[4]
        meter['unit_of_measure'] = row_meter[5]

        ################################################################################################################
        # Step 3: query associated points
        ################################################################################################################
        point_list = list()
        cursor_system.execute(" SELECT p.id, p.name, p.units, p.object_type  "
                              " FROM tbl_meters m, tbl_meters_points mp, tbl_points p "
                              " WHERE m.id = %s AND m.id = mp.meter_id AND mp.point_id = p.id "
                              " ORDER BY p.id ", (meter['id'],))
        rows_points = cursor_system.fetchall()
        if rows_points is not None and len(rows_points) > 0:
            for row in rows_points:
                point_list.append({"id": row[0], "name": row[1], "units": row[2], "object_type": row[3]})

        ################################################################################################################
        # Step 7: query associated points data
        ################################################################################################################
        energy_value_data = dict()
        energy_value_data['name'] = None
        energy_value_data['timestamps'] = list()
        energy_value_data['values'] = list()

        parameters_data = dict()
        parameters_data['names'] = list()
        parameters_data['timestamps'] = list()
        parameters_data['values'] = list()

        if not is_quick_mode:

            for point in point_list:
                point_values = []
                point_timestamps = []
                if point['object_type'] == 'ENERGY_VALUE':
                    energy_value_data['name'] = point['name'] + ' (' + point['units'] + ')'
                    query = (" SELECT utc_date_time, actual_value "
                             " FROM tbl_energy_value "
                             " WHERE point_id = %s "
                             "       AND utc_date_time BETWEEN %s AND %s "
                             " ORDER BY utc_date_time ")
                    cursor_historical.execute(query, (point['id'],
                                                      reporting_start_datetime_utc,
                                                      reporting_end_datetime_utc))
                    rows = cursor_historical.fetchall()
                    if rows is not None and len(rows) > 0:
                        for row in rows:
                            current_datetime_local = row[0].replace(tzinfo=timezone.utc) + \
                                                     timedelta(minutes=timezone_offset)
                            current_datetime = current_datetime_local.isoformat()[0:19]
                            energy_value_data['timestamps'].append(current_datetime)
                            energy_value_data['values'].append(row[1])
                elif point['object_type'] == 'ANALOG_VALUE':

                    query = (" SELECT utc_date_time, actual_value "
                             " FROM tbl_analog_value "
                             " WHERE point_id = %s "
                             "       AND utc_date_time BETWEEN %s AND %s "
                             " ORDER BY utc_date_time ")
                    cursor_historical.execute(query, (point['id'],
                                                      reporting_start_datetime_utc,
                                                      reporting_end_datetime_utc))
                    rows = cursor_historical.fetchall()

                    if rows is not None and len(rows) > 0:
                        for row in rows:
                            current_datetime_local = row[0].replace(tzinfo=timezone.utc) + \
                                                     timedelta(minutes=timezone_offset)
                            current_datetime = current_datetime_local.isoformat()[0:19]
                            point_timestamps.append(current_datetime)
                            point_values.append(row[1])

                    parameters_data['names'].append(point['name'] + ' (' + point['units'] + ')')
                    parameters_data['timestamps'].append(point_timestamps)
                    parameters_data['values'].append(point_values)
                elif point['object_type'] == 'DIGITAL_VALUE':
                    query = (" SELECT utc_date_time, actual_value "
                             " FROM tbl_digital_value "
                             " WHERE point_id = %s "
                             "       AND utc_date_time BETWEEN %s AND %s "
                             " ORDER BY utc_date_time ")
                    cursor_historical.execute(query, (point['id'],
                                                      reporting_start_datetime_utc,
                                                      reporting_end_datetime_utc))
                    rows = cursor_historical.fetchall()

                    if rows is not None and len(rows) > 0:
                        for row in rows:
                            current_datetime_local = row[0].replace(tzinfo=timezone.utc) + \
                                                     timedelta(minutes=timezone_offset)
                            current_datetime = current_datetime_local.isoformat()[0:19]
                            point_timestamps.append(current_datetime)
                            point_values.append(row[1])

                    parameters_data['names'].append(point['name'] + ' (' + point['units'] + ')')
                    parameters_data['timestamps'].append(point_timestamps)
                    parameters_data['values'].append(point_values)

        ################################################################################################################
        # Step 6: construct the report
        ################################################################################################################
        if cursor_system:
            cursor_system.close()
        if cnx_system:
            cnx_system.close()

        if cursor_historical:
            cursor_historical.close()
        if cnx_historical:
            cnx_historical.close()

        result = {
            "meter": {
                "cost_center_id": meter['cost_center_id'],
                "energy_category_id": meter['energy_category_id'],
                "energy_category_name": meter['energy_category_name'],
                "unit_of_measure": meter['unit_of_measure'],
            },
            "energy_value": energy_value_data,
            "parameters": {
                "names": parameters_data['names'],
                "timestamps": parameters_data['timestamps'],
                "values": parameters_data['values']
            },

        }

        resp.text = json.dumps(result)


1		import re
2		from datetime import datetime, timedelta, timezone
3		import falcon
4		import mysql.connector
5		import simplejson as json
6		import config
7		from core.useractivity import access_control, api_key_control
8
9
10		class Reporting:
11		def __init__(self):
12		""""Initializes Reporting"""
13		pass
14
15		@staticmethod
16		def on_options(req, resp):
17		_ = req
18		resp.status = falcon.HTTP_200
19
20		####################################################################################################################
21		# PROCEDURES
22		# Step 1: valid parameters
23		# Step 2: query the meter and energy category
24		# Step 3: query associated points
25		# Step 4: query reporting period points trends
26		# Step 5: query tariff data
27		# Step 6: construct the report
28		####################################################################################################################
29		@staticmethod
30		def on_get(req, resp):
31		if 'API-KEY' not in req.headers or \
32		not isinstance(req.headers['API-KEY'], str) or \
33		len(str.strip(req.headers['API-KEY'])) == 0:
34		access_control(req)
35		else:
36		api_key_control(req)
37		print(req.params)
38		meter_id = req.params.get('meterid')
39		meter_uuid = req.params.get('meteruuid')
40		quick_mode = req.params.get('quickmode')
41
42		################################################################################################################
43		# Step 1: valid parameters
44		################################################################################################################
45		if meter_id is None and meter_uuid is None:
46		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST', description='API.INVALID_METER_ID')
47
48		if meter_id is not None:
49		meter_id = str.strip(meter_id)
50		if not meter_id.isdigit() or int(meter_id) <= 0:
51		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
52		description='API.INVALID_METER_ID')
53
54		if meter_uuid is not None:
55		regex = re.compile(r'^[a-f0-9]{8}-?[a-f0-9]{4}-?4[a-f0-9]{3}-?[89ab][a-f0-9]{3}-?[a-f0-9]{12}\Z', re.I)
56		match = regex.match(str.strip(meter_uuid))
57		if not bool(match):
58		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
59		description='API.INVALID_METER_UUID')
60
61		# if turn quick mode on, do not return parameters data and excel file
62		is_quick_mode = False
63		if quick_mode is not None and \
64		len(str.strip(quick_mode)) > 0 and \
65		str.lower(str.strip(quick_mode)) in ('true', 't', 'on', 'yes', 'y'):
66		is_quick_mode = True
67
68		timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
69		if config.utc_offset[0] == '-':
70		timezone_offset = -timezone_offset
71
72		reporting_end_datetime_utc = datetime.utcnow()
73		reporting_start_datetime_utc = reporting_end_datetime_utc - timedelta(minutes=60)
74		################################################################################################################
75		# Step 2: query the meter and energy category
76		################################################################################################################
77		cnx_system = mysql.connector.connect(**config.myems_system_db)
78		cursor_system = cnx_system.cursor()
79
80		cnx_historical = mysql.connector.connect(**config.myems_historical_db)
81		cursor_historical = cnx_historical.cursor()
82
83		if meter_id is not None:
84		cursor_system.execute(" SELECT m.id, m.name, m.cost_center_id, m.energy_category_id, "
85		" ec.name, ec.unit_of_measure, ec.kgce, ec.kgco2e "
86		" FROM tbl_meters m, tbl_energy_categories ec "
87		" WHERE m.id = %s AND m.energy_category_id = ec.id ", (meter_id,))
88		row_meter = cursor_system.fetchone()
89		elif meter_uuid is not None:
90		cursor_system.execute(" SELECT m.id, m.name, m.cost_center_id, m.energy_category_id, "
91		" ec.name, ec.unit_of_measure, ec.kgce, ec.kgco2e "
92		" FROM tbl_meters m, tbl_energy_categories ec "
93		" WHERE m.uuid = %s AND m.energy_category_id = ec.id ", (meter_uuid,))
94		row_meter = cursor_system.fetchone()
95
96		if row_meter is None:
		0 ignored issues – show introduced 2022-01-18 14:12 UTC by Report Bug Copy Issue Report The variable `row_meter` does not seem to be defined for all execution paths. Loading history...
97		if cursor_system:
98		cursor_system.close()
99		if cnx_system:
100		cnx_system.close()
101
102		if cursor_historical:
103		cursor_historical.close()
104		if cnx_historical:
105		cnx_historical.close()
106		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND', description='API.METER_NOT_FOUND')
107
108		meter = dict()
109		meter['id'] = row_meter[0]
110		meter['name'] = row_meter[1]
111		meter['cost_center_id'] = row_meter[2]
112		meter['energy_category_id'] = row_meter[3]
113		meter['energy_category_name'] = row_meter[4]
114		meter['unit_of_measure'] = row_meter[5]
115
116		################################################################################################################
117		# Step 3: query associated points
118		################################################################################################################
119		point_list = list()
120		cursor_system.execute(" SELECT p.id, p.name, p.units, p.object_type "
121		" FROM tbl_meters m, tbl_meters_points mp, tbl_points p "
122		" WHERE m.id = %s AND m.id = mp.meter_id AND mp.point_id = p.id "
123		" ORDER BY p.id ", (meter['id'],))
124		rows_points = cursor_system.fetchall()
125		if rows_points is not None and len(rows_points) > 0:
126		for row in rows_points:
127		point_list.append({"id": row[0], "name": row[1], "units": row[2], "object_type": row[3]})
128
129		################################################################################################################
130		# Step 7: query associated points data
131		################################################################################################################
132		energy_value_data = dict()
133		energy_value_data['name'] = None
134		energy_value_data['timestamps'] = list()
135		energy_value_data['values'] = list()
136
137		parameters_data = dict()
138		parameters_data['names'] = list()
139		parameters_data['timestamps'] = list()
140		parameters_data['values'] = list()
141
142	View Code Duplication	if not is_quick_mode:
		0 ignored issues – show Duplication introduced 2022-09-03 03:44 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
143		for point in point_list:
144		point_values = []
145		point_timestamps = []
146		if point['object_type'] == 'ENERGY_VALUE':
147		energy_value_data['name'] = point['name'] + ' (' + point['units'] + ')'
148		query = (" SELECT utc_date_time, actual_value "
149		" FROM tbl_energy_value "
150		" WHERE point_id = %s "
151		" AND utc_date_time BETWEEN %s AND %s "
152		" ORDER BY utc_date_time ")
153		cursor_historical.execute(query, (point['id'],
154		reporting_start_datetime_utc,
155		reporting_end_datetime_utc))
156		rows = cursor_historical.fetchall()
157		if rows is not None and len(rows) > 0:
158		for row in rows:
159		current_datetime_local = row[0].replace(tzinfo=timezone.utc) + \
160		timedelta(minutes=timezone_offset)
161		current_datetime = current_datetime_local.isoformat()[0:19]
162		energy_value_data['timestamps'].append(current_datetime)
163		energy_value_data['values'].append(row[1])
164		elif point['object_type'] == 'ANALOG_VALUE':
165
166		query = (" SELECT utc_date_time, actual_value "
167		" FROM tbl_analog_value "
168		" WHERE point_id = %s "
169		" AND utc_date_time BETWEEN %s AND %s "
170		" ORDER BY utc_date_time ")
171		cursor_historical.execute(query, (point['id'],
172		reporting_start_datetime_utc,
173		reporting_end_datetime_utc))
174		rows = cursor_historical.fetchall()
175
176		if rows is not None and len(rows) > 0:
177		for row in rows:
178		current_datetime_local = row[0].replace(tzinfo=timezone.utc) + \
179		timedelta(minutes=timezone_offset)
180		current_datetime = current_datetime_local.isoformat()[0:19]
181		point_timestamps.append(current_datetime)
182		point_values.append(row[1])
183
184		parameters_data['names'].append(point['name'] + ' (' + point['units'] + ')')
185		parameters_data['timestamps'].append(point_timestamps)
186		parameters_data['values'].append(point_values)
187		elif point['object_type'] == 'DIGITAL_VALUE':
188		query = (" SELECT utc_date_time, actual_value "
189		" FROM tbl_digital_value "
190		" WHERE point_id = %s "
191		" AND utc_date_time BETWEEN %s AND %s "
192		" ORDER BY utc_date_time ")
193		cursor_historical.execute(query, (point['id'],
194		reporting_start_datetime_utc,
195		reporting_end_datetime_utc))
196		rows = cursor_historical.fetchall()
197
198		if rows is not None and len(rows) > 0:
199		for row in rows:
200		current_datetime_local = row[0].replace(tzinfo=timezone.utc) + \
201		timedelta(minutes=timezone_offset)
202		current_datetime = current_datetime_local.isoformat()[0:19]
203		point_timestamps.append(current_datetime)
204		point_values.append(row[1])
205
206		parameters_data['names'].append(point['name'] + ' (' + point['units'] + ')')
207		parameters_data['timestamps'].append(point_timestamps)
208		parameters_data['values'].append(point_values)
209
210		################################################################################################################
211		# Step 6: construct the report
212		################################################################################################################
213		if cursor_system:
214		cursor_system.close()
215		if cnx_system:
216		cnx_system.close()
217
218		if cursor_historical:
219		cursor_historical.close()
220		if cnx_historical:
221		cnx_historical.close()
222
223		result = {
224		"meter": {
225		"cost_center_id": meter['cost_center_id'],
226		"energy_category_id": meter['energy_category_id'],
227		"energy_category_name": meter['energy_category_name'],
228		"unit_of_measure": meter['unit_of_measure'],
229		},
230		"energy_value": energy_value_data,
231		"parameters": {
232		"names": parameters_data['names'],
233		"timestamps": parameters_data['timestamps'],
234		"values": parameters_data['values']
235		},
236
237		}
238
239		resp.text = json.dumps(result)
240

MyEMS / myems

reports.meterrealtime.Reporting.on_get() 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

reports.meterrealtime.Reporting.on_get() F
last analyzed 2025-06-17 15:42 UTC