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

Reporting.on_options() A
last analyzed 2025-06-17 15:42 UTC

↳ Parent: reports.photovoltaicpowerstationdetailsmeter

Complexity

Conditions

Size

Total Lines	5
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	5
dl	0
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 photovoltaic 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_PHOTOVOLTAIC_POWER_STATION_ID')
        photovoltaic_power_station_id = id_
        ################################################################################################################
        # Step 2: query the photovoltaic 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 photovoltaic_power_station_id is not None:
            query = (" SELECT name "
                     " FROM tbl_photovoltaic_power_stations "
                     " WHERE id = %s ")
            cursor_system.execute(query, (photovoltaic_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.PHOTOVOLTAIC_POWER_STATION_NOT_FOUND')
        else:
            photovoltaic_power_station_name = row[0]
        ################################################################################################################
        # 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]] = 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]] = 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]] = row[1]

        ################################################################################################################
        # Step 7: query the points of meters
        ################################################################################################################

        # query grid meter parameters
        meter_list = list()

        cursor_system.execute(" SELECT id, name, uuid, "
                              "        total_active_power_point_id, "
                              "        active_power_a_point_id, "
                              "        active_power_b_point_id, "
                              "        active_power_c_point_id, "
                              "        total_reactive_power_point_id, "
                              "        reactive_power_a_point_id, "
                              "        reactive_power_b_point_id, "
                              "        reactive_power_c_point_id, "
                              "        total_apparent_power_point_id, "
                              "        apparent_power_a_point_id, "
                              "        apparent_power_b_point_id, "
                              "        apparent_power_c_point_id, "
                              "        total_power_factor_point_id, "
                              "        active_energy_import_point_id, "
                              "        active_energy_export_point_id, "
                              "        active_energy_net_point_id "
                              " FROM tbl_photovoltaic_power_stations_grids "
                              " WHERE photovoltaic_power_station_id = %s "
                              " ORDER BY id ",
                              (photovoltaic_power_station_id,))
        rows_grid_meters = cursor_system.fetchall()
        if rows_grid_meters is not None and len(rows_grid_meters) > 0:
            for row in rows_grid_meters:
                current_grid_meter = dict()
                current_grid_meter['id'] = row[0]
                current_grid_meter['name'] = photovoltaic_power_station_name + '-' + row[1]
                current_grid_meter['uuid'] = row[2]
                current_grid_meter['total_active_power_point'] = latest_value_dict.get(row[3], None)
                current_grid_meter['active_power_a_point'] = latest_value_dict.get(row[4], None)
                current_grid_meter['active_power_b_point'] = latest_value_dict.get(row[5], None)
                current_grid_meter['active_power_c_point'] = latest_value_dict.get(row[6], None)
                current_grid_meter['total_reactive_power_point'] = latest_value_dict.get(row[7], None)
                current_grid_meter['reactive_power_a_point'] = latest_value_dict.get(row[8], None)
                current_grid_meter['reactive_power_b_point'] = latest_value_dict.get(row[9], None)
                current_grid_meter['reactive_power_c_point'] = latest_value_dict.get(row[10], None)
                current_grid_meter['total_apparent_power_point'] = latest_value_dict.get(row[11], None)
                current_grid_meter['apparent_power_a_point'] = latest_value_dict.get(row[12], None)
                current_grid_meter['apparent_power_b_point'] = latest_value_dict.get(row[13], None)
                current_grid_meter['apparent_power_c_point'] = latest_value_dict.get(row[14], None)
                current_grid_meter['total_power_factor_point'] = latest_value_dict.get(row[15], None)
                current_grid_meter['active_energy_import_point'] = latest_value_dict.get(row[16], None)
                current_grid_meter['active_energy_export_point'] = latest_value_dict.get(row[17], None)
                current_grid_meter['active_energy_net_point'] = latest_value_dict.get(row[18], None)
                meter_list.append(current_grid_meter)

        # query load meter parameters
        cursor_system.execute(" SELECT id, name, uuid, "
                              "        total_active_power_point_id, "
                              "        active_power_a_point_id, "
                              "        active_power_b_point_id, "
                              "        active_power_c_point_id, "
                              "        total_reactive_power_point_id, "
                              "        reactive_power_a_point_id, "
                              "        reactive_power_b_point_id, "
                              "        reactive_power_c_point_id, "
                              "        total_apparent_power_point_id, "
                              "        apparent_power_a_point_id, "
                              "        apparent_power_b_point_id, "
                              "        apparent_power_c_point_id, "
                              "        total_power_factor_point_id, "
                              "        active_energy_import_point_id, "
                              "        active_energy_export_point_id, "
                              "        active_energy_net_point_id "
                              " FROM tbl_photovoltaic_power_stations_loads "
                              " WHERE photovoltaic_power_station_id = %s "
                              " ORDER BY id ",
                              (photovoltaic_power_station_id,))
        rows_load_meters = cursor_system.fetchall()
        if rows_load_meters is not None and len(rows_load_meters) > 0:
            for row in rows_load_meters:
                current_load_meter = dict()
                current_load_meter['id'] = row[0]
                current_load_meter['name'] = photovoltaic_power_station_name + '-' + row[1]
                current_load_meter['uuid'] = row[2]
                current_load_meter['total_active_power_point'] = latest_value_dict.get(row[3], None)
                current_load_meter['active_power_a_point'] = latest_value_dict.get(row[4], None)
                current_load_meter['active_power_b_point'] = latest_value_dict.get(row[5], None)
                current_load_meter['active_power_c_point'] = latest_value_dict.get(row[6], None)
                current_load_meter['total_reactive_power_point'] = latest_value_dict.get(row[7], None)
                current_load_meter['reactive_power_a_point'] = latest_value_dict.get(row[8], None)
                current_load_meter['reactive_power_b_point'] = latest_value_dict.get(row[9], None)
                current_load_meter['reactive_power_c_point'] = latest_value_dict.get(row[10], None)
                current_load_meter['total_apparent_power_point'] = latest_value_dict.get(row[11], None)
                current_load_meter['apparent_power_a_point'] = latest_value_dict.get(row[12], None)
                current_load_meter['apparent_power_b_point'] = latest_value_dict.get(row[13], None)
                current_load_meter['apparent_power_c_point'] = latest_value_dict.get(row[14], None)
                current_load_meter['total_power_factor_point'] = latest_value_dict.get(row[15], None)
                current_load_meter['active_energy_import_point'] = latest_value_dict.get(row[16], None)
                current_load_meter['active_energy_export_point'] = latest_value_dict.get(row[17], None)
                current_load_meter['active_energy_net_point'] = latest_value_dict.get(row[18], None)
                meter_list.append(current_load_meter)

        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			class Reporting:
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 photovoltaic 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_PHOTOVOLTAIC_POWER_STATION_ID')
46			photovoltaic_power_station_id = id_
47			################################################################################################################
48			# Step 2: query the photovoltaic 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 photovoltaic_power_station_id is not None:
57			query = (" SELECT name "
58			" FROM tbl_photovoltaic_power_stations "
59			" WHERE id = %s ")
60			cursor_system.execute(query, (photovoltaic_power_station_id,))
61			row = cursor_system.fetchone()
62
63			if row is None:
			0 ignored issues – show introduced 2024-11-16 14:16 UTC by Report Bug Copy Issue Report The variable `row` does not seem to be defined in case `photovoltaic_power_station_id is not None` on line `56` is `False`. Are you sure this can never be the case? Loading history...
64			cursor_system.close()
65			cnx_system.close()
66			raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
67			description='API.PHOTOVOLTAIC_POWER_STATION_NOT_FOUND')
68			else:
69			photovoltaic_power_station_name = row[0]
70			################################################################################################################
71			# Step 4: query analog points latest values
72			################################################################################################################
73			latest_value_dict = dict()
74			query = (" SELECT point_id, actual_value "
75			" FROM tbl_analog_value_latest "
76			" WHERE utc_date_time > %s ")
77			cursor_historical.execute(query, (datetime.utcnow() - timedelta(minutes=60),))
78			rows = cursor_historical.fetchall()
79			if rows is not None and len(rows) > 0:
80			for row in rows:
81			latest_value_dict[row[0]] = row[1]
82
83			################################################################################################################
84			# Step 5: query energy points latest values
85			################################################################################################################
86			query = (" SELECT point_id, actual_value "
87			" FROM tbl_energy_value_latest "
88			" WHERE utc_date_time > %s ")
89			cursor_historical.execute(query, (datetime.utcnow() - timedelta(minutes=60),))
90			rows = cursor_historical.fetchall()
91			if rows is not None and len(rows) > 0:
92			for row in rows:
93			latest_value_dict[row[0]] = row[1]
94
95			################################################################################################################
96			# Step 6: query digital points latest values
97			################################################################################################################
98			query = (" SELECT point_id, actual_value "
99			" FROM tbl_digital_value_latest "
100			" WHERE utc_date_time > %s ")
101			cursor_historical.execute(query, (datetime.utcnow() - timedelta(minutes=60),))
102			rows = cursor_historical.fetchall()
103			if rows is not None and len(rows) > 0:
104			for row in rows:
105			latest_value_dict[row[0]] = row[1]
106
107			################################################################################################################
108			# Step 7: query the points of meters
109			################################################################################################################
110
111			# query grid meter parameters
112			meter_list = list()
113
114			cursor_system.execute(" SELECT id, name, uuid, "
115			" total_active_power_point_id, "
116			" active_power_a_point_id, "
117			" active_power_b_point_id, "
118			" active_power_c_point_id, "
119			" total_reactive_power_point_id, "
120			" reactive_power_a_point_id, "
121			" reactive_power_b_point_id, "
122			" reactive_power_c_point_id, "
123			" total_apparent_power_point_id, "
124			" apparent_power_a_point_id, "
125			" apparent_power_b_point_id, "
126			" apparent_power_c_point_id, "
127			" total_power_factor_point_id, "
128			" active_energy_import_point_id, "
129			" active_energy_export_point_id, "
130			" active_energy_net_point_id "
131			" FROM tbl_photovoltaic_power_stations_grids "
132			" WHERE photovoltaic_power_station_id = %s "
133			" ORDER BY id ",
134			(photovoltaic_power_station_id,))
135			rows_grid_meters = cursor_system.fetchall()
136			if rows_grid_meters is not None and len(rows_grid_meters) > 0:
137			for row in rows_grid_meters:
138			current_grid_meter = dict()
139			current_grid_meter['id'] = row[0]
140			current_grid_meter['name'] = photovoltaic_power_station_name + '-' + row[1]
141			current_grid_meter['uuid'] = row[2]
142			current_grid_meter['total_active_power_point'] = latest_value_dict.get(row[3], None)
143			current_grid_meter['active_power_a_point'] = latest_value_dict.get(row[4], None)
144			current_grid_meter['active_power_b_point'] = latest_value_dict.get(row[5], None)
145			current_grid_meter['active_power_c_point'] = latest_value_dict.get(row[6], None)
146			current_grid_meter['total_reactive_power_point'] = latest_value_dict.get(row[7], None)
147			current_grid_meter['reactive_power_a_point'] = latest_value_dict.get(row[8], None)
148			current_grid_meter['reactive_power_b_point'] = latest_value_dict.get(row[9], None)
149			current_grid_meter['reactive_power_c_point'] = latest_value_dict.get(row[10], None)
150			current_grid_meter['total_apparent_power_point'] = latest_value_dict.get(row[11], None)
151			current_grid_meter['apparent_power_a_point'] = latest_value_dict.get(row[12], None)
152			current_grid_meter['apparent_power_b_point'] = latest_value_dict.get(row[13], None)
153			current_grid_meter['apparent_power_c_point'] = latest_value_dict.get(row[14], None)
154			current_grid_meter['total_power_factor_point'] = latest_value_dict.get(row[15], None)
155			current_grid_meter['active_energy_import_point'] = latest_value_dict.get(row[16], None)
156			current_grid_meter['active_energy_export_point'] = latest_value_dict.get(row[17], None)
157			current_grid_meter['active_energy_net_point'] = latest_value_dict.get(row[18], None)
158			meter_list.append(current_grid_meter)
159
160			# query load meter parameters
161			cursor_system.execute(" SELECT id, name, uuid, "
162			" total_active_power_point_id, "
163			" active_power_a_point_id, "
164			" active_power_b_point_id, "
165			" active_power_c_point_id, "
166			" total_reactive_power_point_id, "
167			" reactive_power_a_point_id, "
168			" reactive_power_b_point_id, "
169			" reactive_power_c_point_id, "
170			" total_apparent_power_point_id, "
171			" apparent_power_a_point_id, "
172			" apparent_power_b_point_id, "
173			" apparent_power_c_point_id, "
174			" total_power_factor_point_id, "
175			" active_energy_import_point_id, "
176			" active_energy_export_point_id, "
177			" active_energy_net_point_id "
178			" FROM tbl_photovoltaic_power_stations_loads "
179			" WHERE photovoltaic_power_station_id = %s "
180			" ORDER BY id ",
181			(photovoltaic_power_station_id,))
182			rows_load_meters = cursor_system.fetchall()
183			if rows_load_meters is not None and len(rows_load_meters) > 0:
184			for row in rows_load_meters:
185			current_load_meter = dict()
186			current_load_meter['id'] = row[0]
187			current_load_meter['name'] = photovoltaic_power_station_name + '-' + row[1]
188			current_load_meter['uuid'] = row[2]
189			current_load_meter['total_active_power_point'] = latest_value_dict.get(row[3], None)
190			current_load_meter['active_power_a_point'] = latest_value_dict.get(row[4], None)
191			current_load_meter['active_power_b_point'] = latest_value_dict.get(row[5], None)
192			current_load_meter['active_power_c_point'] = latest_value_dict.get(row[6], None)
193			current_load_meter['total_reactive_power_point'] = latest_value_dict.get(row[7], None)
194			current_load_meter['reactive_power_a_point'] = latest_value_dict.get(row[8], None)
195			current_load_meter['reactive_power_b_point'] = latest_value_dict.get(row[9], None)
196			current_load_meter['reactive_power_c_point'] = latest_value_dict.get(row[10], None)
197			current_load_meter['total_apparent_power_point'] = latest_value_dict.get(row[11], None)
198			current_load_meter['apparent_power_a_point'] = latest_value_dict.get(row[12], None)
199			current_load_meter['apparent_power_b_point'] = latest_value_dict.get(row[13], None)
200			current_load_meter['apparent_power_c_point'] = latest_value_dict.get(row[14], None)
201			current_load_meter['total_power_factor_point'] = latest_value_dict.get(row[15], None)
202			current_load_meter['active_energy_import_point'] = latest_value_dict.get(row[16], None)
203			current_load_meter['active_energy_export_point'] = latest_value_dict.get(row[17], None)
204			current_load_meter['active_energy_net_point'] = latest_value_dict.get(row[18], None)
205			meter_list.append(current_load_meter)
206
207			if cursor_system:
208			cursor_system.close()
209			if cnx_system:
210			cnx_system.close()
211
212			if cursor_historical:
213			cursor_historical.close()
214			if cnx_historical:
215			cnx_historical.close()
216			################################################################################################################
217			# Step 8: construct the report
218			################################################################################################################
219			resp.text = json.dumps(meter_list)
220

MyEMS / myems

Reporting.on_options() A last analyzed 2025-06-17 15:42 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

Reporting.on_options() A
last analyzed 2025-06-17 15:42 UTC