reports.energystoragepowerstationdetailsmeter.Reporting.on_options() - Code Metrics - Inspection of "Merge branch 'develop'" - MyEMS/myems - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 183c79...49504c )

unknown

created 2025-07-30 09:00 UTC

Reporting.on_options() A

↳ Parent: reports.energystoragepowerstationdetailsmeter

Complexity

Conditions

Size

Total Lines	5
Code Lines	5

Duplication

Lines	5
Ratio	100 %

Importance

Changes

Metric	Value
eloc	5
dl	5
loc	5
rs	10
c	0
b	0
f	0
cc	1
nop	3

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


class Reporting:

    def __init__(self):
        """Initializes Class"""
        pass

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

    ####################################################################################################################
    # PROCEDURES
    # Step 1: valid parameters
    # Step 2: query the energy storage power station
    # Step 3: query associated containers
    # Step 4: query analog points latest values
    # Step 5: query energy points latest values
    # Step 6: query digital points latest values
    # Step 7: query the points of meters
    # Step 8: construct the report
    ####################################################################################################################
    @staticmethod
    def on_get(req, resp, id_):
        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)

        ################################################################################################################
        # Step 1: valid parameters
        ################################################################################################################
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_ENERGY_STORAGE_POWER_STATION_ID')
        energy_storage_power_station_id = id_
        ################################################################################################################
        # Step 2: query the energy storage power station
        ################################################################################################################
        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 energy_storage_power_station_id is not None:
            query = (" SELECT id, name, uuid "
                     " FROM tbl_energy_storage_power_stations "
                     " WHERE id = %s ")
            cursor_system.execute(query, (energy_storage_power_station_id,))
            row = cursor_system.fetchone()
            if row is None:
                cursor_system.close()
                cnx_system.close()
                raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                       description='API.ENERGY_STORAGE_POWER_STATION_NOT_FOUND')

        # query all points
        query = (" SELECT id, name, units, description "
                 " FROM tbl_points ")
        cursor_system.execute(query)
        rows = cursor_system.fetchall()

        points_dict = dict()
        if rows is not None and len(rows) > 0:
            for row in rows:
                points_dict[row[0]] = [row[1], row[2], row[3]]
        ################################################################################################################
        # Step 3: query associated containers
        ################################################################################################################
        container_list = list()
        cursor_system.execute(" SELECT c.id, c.name, c.uuid "
                              " FROM tbl_energy_storage_power_stations_containers espsc, "
                              "      tbl_energy_storage_containers c "
                              " WHERE espsc.energy_storage_power_station_id = %s "
                              "      AND espsc.energy_storage_container_id = c.id ",
                              (energy_storage_power_station_id,))
        rows_containers = cursor_system.fetchall()
        if rows_containers is not None and len(rows_containers) > 0:
            for row_container in rows_containers:
                container_list.append({"id": row_container[0],
                                       "name": row_container[1],
                                       "uuid": row_container[2]})
        print('container_list:' + str(container_list))

        ################################################################################################################
        # Step 4: query analog points latest values
        ################################################################################################################
        latest_value_dict = dict()
        query = (" SELECT point_id, actual_value "
                 " FROM tbl_analog_value_latest "
                 " WHERE utc_date_time > %s ")
        cursor_historical.execute(query, (datetime.utcnow() - timedelta(minutes=60),))
        rows = cursor_historical.fetchall()
        if rows is not None and len(rows) > 0:
            for row in rows:
                latest_value_dict[row[0]] = [points_dict[row[0]][0],
                                             points_dict[row[0]][1],
                                             points_dict[row[0]][2],
                                             row[1]]

        ################################################################################################################
        # Step 5: query energy points latest values
        ################################################################################################################
        query = (" SELECT point_id, actual_value "
                 " FROM tbl_energy_value_latest "
                 " WHERE utc_date_time > %s ")
        cursor_historical.execute(query, (datetime.utcnow() - timedelta(minutes=60),))
        rows = cursor_historical.fetchall()
        if rows is not None and len(rows) > 0:
            for row in rows:
                latest_value_dict[row[0]] = [points_dict[row[0]][0],
                                             points_dict[row[0]][1],
                                             points_dict[row[0]][2],
                                             row[1]]

        ################################################################################################################
        # Step 6: query digital points latest values
        ################################################################################################################
        query = (" SELECT point_id, actual_value "
                 " FROM tbl_digital_value_latest "
                 " WHERE utc_date_time > %s ")
        cursor_historical.execute(query, (datetime.utcnow() - timedelta(minutes=60),))
        rows = cursor_historical.fetchall()
        if rows is not None and len(rows) > 0:
            for row in rows:
                latest_value_dict[row[0]] = [points_dict[row[0]][0],
                                             points_dict[row[0]][1],
                                             points_dict[row[0]][2],
                                             row[1]]

        ################################################################################################################
        # Step 7: query the points of meters
        ################################################################################################################
        # query all points with units
        query = (" SELECT id, units "
                 " FROM tbl_points ")
        cursor_system.execute(query)
        rows = cursor_system.fetchall()

        units_dict = dict()
        if rows is not None and len(rows) > 0:
            for row in rows:
                units_dict[row[0]] = row[1]

        # query meter parameters
        meter_list = list()
        for container in container_list:
            cursor_system.execute(" SELECT charge_meter_id, discharge_meter_id "
                                  " FROM tbl_energy_storage_containers_batteries "
                                  " WHERE energy_storage_container_id = %s "
                                  " ORDER BY id ",
                                  (container['id'],))
            rows_meters = cursor_system.fetchall()
            if rows_meters is not None and len(rows_meters) > 0:
                for row in rows_meters:
                    charge_meter = dict()
                    charge_meter['id'] = row[0]
                    charge_meter['points'] = list()
                    meter_list.append(charge_meter)
                    discharge_meter = dict()
                    discharge_meter['id'] = row[1]
                    discharge_meter['points'] = list()
                    meter_list.append(discharge_meter)

            for index, meter in enumerate(meter_list):
                cursor_system.execute(" SELECT p.id "
                                      " FROM tbl_meters_points mp, tbl_points p "
                                      " WHERE mp.meter_id = %s AND mp.point_id = p.id "
                                      " ORDER BY mp.id ",
                                      (meter['id'],))
                rows_points = cursor_system.fetchall()
                if rows_points is not None and len(rows_points) > 0:
                    point_list = list()
                    for row in rows_points:
                        point = latest_value_dict.get(row[0], None)
                        if point is not None:
                            point_list.append(point)
                    meter_list[index]['points'] = point_list

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

        if cursor_historical:
            cursor_historical.close()
        if cnx_historical:
            cnx_historical.close()
        ################################################################################################################
        # Step 8: construct the report
        ################################################################################################################
        resp.text = json.dumps(meter_list)


1		from datetime import datetime, timedelta
2		import falcon
3		import mysql.connector
4		import simplejson as json
5		from core.useractivity import access_control, api_key_control
6		import config
7
8
9	View Code Duplication	class Reporting:
		0 ignored issues – show Duplication introduced 2025-03-02 07:57 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
10		def __init__(self):
11		"""Initializes Class"""
12		pass
13
14		@staticmethod
15		def on_options(req, resp, id_):
16		_ = req
17		resp.status = falcon.HTTP_200
18		_ = id_
19
20		####################################################################################################################
21		# PROCEDURES
22		# Step 1: valid parameters
23		# Step 2: query the energy storage power station
24		# Step 3: query associated containers
25		# Step 4: query analog points latest values
26		# Step 5: query energy points latest values
27		# Step 6: query digital points latest values
28		# Step 7: query the points of meters
29		# Step 8: construct the report
30		####################################################################################################################
31		@staticmethod
32		def on_get(req, resp, id_):
33		if 'API-KEY' not in req.headers or \
34		not isinstance(req.headers['API-KEY'], str) or \
35		len(str.strip(req.headers['API-KEY'])) == 0:
36		access_control(req)
37		else:
38		api_key_control(req)
39
40		################################################################################################################
41		# Step 1: valid parameters
42		################################################################################################################
43		if not id_.isdigit() or int(id_) <= 0:
44		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
45		description='API.INVALID_ENERGY_STORAGE_POWER_STATION_ID')
46		energy_storage_power_station_id = id_
47		################################################################################################################
48		# Step 2: query the energy storage power station
49		################################################################################################################
50		cnx_system = mysql.connector.connect(**config.myems_system_db)
51		cursor_system = cnx_system.cursor()
52
53		cnx_historical = mysql.connector.connect(**config.myems_historical_db)
54		cursor_historical = cnx_historical.cursor()
55
56		if energy_storage_power_station_id is not None:
57		query = (" SELECT id, name, uuid "
58		" FROM tbl_energy_storage_power_stations "
59		" WHERE id = %s ")
60		cursor_system.execute(query, (energy_storage_power_station_id,))
61		row = cursor_system.fetchone()
62		if row is None:
63		cursor_system.close()
64		cnx_system.close()
65		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
66		description='API.ENERGY_STORAGE_POWER_STATION_NOT_FOUND')
67
68		# query all points
69		query = (" SELECT id, name, units, description "
70		" FROM tbl_points ")
71		cursor_system.execute(query)
72		rows = cursor_system.fetchall()
73
74		points_dict = dict()
75		if rows is not None and len(rows) > 0:
76		for row in rows:
77		points_dict[row[0]] = [row[1], row[2], row[3]]
78		################################################################################################################
79		# Step 3: query associated containers
80		################################################################################################################
81		container_list = list()
82		cursor_system.execute(" SELECT c.id, c.name, c.uuid "
83		" FROM tbl_energy_storage_power_stations_containers espsc, "
84		" tbl_energy_storage_containers c "
85		" WHERE espsc.energy_storage_power_station_id = %s "
86		" AND espsc.energy_storage_container_id = c.id ",
87		(energy_storage_power_station_id,))
88		rows_containers = cursor_system.fetchall()
89		if rows_containers is not None and len(rows_containers) > 0:
90		for row_container in rows_containers:
91		container_list.append({"id": row_container[0],
92		"name": row_container[1],
93		"uuid": row_container[2]})
94		print('container_list:' + str(container_list))
95
96		################################################################################################################
97		# Step 4: query analog points latest values
98		################################################################################################################
99		latest_value_dict = dict()
100		query = (" SELECT point_id, actual_value "
101		" FROM tbl_analog_value_latest "
102		" WHERE utc_date_time > %s ")
103		cursor_historical.execute(query, (datetime.utcnow() - timedelta(minutes=60),))
104		rows = cursor_historical.fetchall()
105		if rows is not None and len(rows) > 0:
106		for row in rows:
107		latest_value_dict[row[0]] = [points_dict[row[0]][0],
108		points_dict[row[0]][1],
109		points_dict[row[0]][2],
110		row[1]]
111
112		################################################################################################################
113		# Step 5: query energy points latest values
114		################################################################################################################
115		query = (" SELECT point_id, actual_value "
116		" FROM tbl_energy_value_latest "
117		" WHERE utc_date_time > %s ")
118		cursor_historical.execute(query, (datetime.utcnow() - timedelta(minutes=60),))
119		rows = cursor_historical.fetchall()
120		if rows is not None and len(rows) > 0:
121		for row in rows:
122		latest_value_dict[row[0]] = [points_dict[row[0]][0],
123		points_dict[row[0]][1],
124		points_dict[row[0]][2],
125		row[1]]
126
127		################################################################################################################
128		# Step 6: query digital points latest values
129		################################################################################################################
130		query = (" SELECT point_id, actual_value "
131		" FROM tbl_digital_value_latest "
132		" WHERE utc_date_time > %s ")
133		cursor_historical.execute(query, (datetime.utcnow() - timedelta(minutes=60),))
134		rows = cursor_historical.fetchall()
135		if rows is not None and len(rows) > 0:
136		for row in rows:
137		latest_value_dict[row[0]] = [points_dict[row[0]][0],
138		points_dict[row[0]][1],
139		points_dict[row[0]][2],
140		row[1]]
141
142		################################################################################################################
143		# Step 7: query the points of meters
144		################################################################################################################
145		# query all points with units
146		query = (" SELECT id, units "
147		" FROM tbl_points ")
148		cursor_system.execute(query)
149		rows = cursor_system.fetchall()
150
151		units_dict = dict()
152		if rows is not None and len(rows) > 0:
153		for row in rows:
154		units_dict[row[0]] = row[1]
155
156		# query meter parameters
157		meter_list = list()
158		for container in container_list:
159		cursor_system.execute(" SELECT charge_meter_id, discharge_meter_id "
160		" FROM tbl_energy_storage_containers_batteries "
161		" WHERE energy_storage_container_id = %s "
162		" ORDER BY id ",
163		(container['id'],))
164		rows_meters = cursor_system.fetchall()
165		if rows_meters is not None and len(rows_meters) > 0:
166		for row in rows_meters:
167		charge_meter = dict()
168		charge_meter['id'] = row[0]
169		charge_meter['points'] = list()
170		meter_list.append(charge_meter)
171		discharge_meter = dict()
172		discharge_meter['id'] = row[1]
173		discharge_meter['points'] = list()
174		meter_list.append(discharge_meter)
175
176		for index, meter in enumerate(meter_list):
177		cursor_system.execute(" SELECT p.id "
178		" FROM tbl_meters_points mp, tbl_points p "
179		" WHERE mp.meter_id = %s AND mp.point_id = p.id "
180		" ORDER BY mp.id ",
181		(meter['id'],))
182		rows_points = cursor_system.fetchall()
183		if rows_points is not None and len(rows_points) > 0:
184		point_list = list()
185		for row in rows_points:
186		point = latest_value_dict.get(row[0], None)
187		if point is not None:
188		point_list.append(point)
189		meter_list[index]['points'] = point_list
190
191		if cursor_system:
192		cursor_system.close()
193		if cnx_system:
194		cnx_system.close()
195
196		if cursor_historical:
197		cursor_historical.close()
198		if cnx_historical:
199		cnx_historical.close()
200		################################################################################################################
201		# Step 8: construct the report
202		################################################################################################################
203		resp.text = json.dumps(meter_list)
204

MyEMS / myems

Push — master ( 183c79...49504c )

Reporting.on_options() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like