Code Duplication - MyEMS/myems - Measure and Improve Code Quality continuously with Scrutinizer

Code Duplication Length = 98-107 lines in 2 locations

myems-api/core/microgrid.py 1 location


        _ = req
        resp.status = falcon.HTTP_200

    @staticmethod
    def on_get(req, resp):
        """
        Handle GET requests to retrieve all microgrids

        Returns a list of all microgrids with their complete information including:
        - Microgrid ID, name, and UUID
        - Associated contact and cost center information
        - Microgrid specifications and parameters
        - Related equipment and meter associations
        - SVG diagram information for visualization

        Args:
            req: Falcon request object
            resp: Falcon response object
        """
        access_control(req)

        search_query = req.get_param('q', default=None)
        if search_query is not None and len(search_query.strip()) > 0:
            search_query = search_query.strip()
        else:
            search_query = ''

        # Connect to database
        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        # Query to retrieve all contacts for reference
        query = (" SELECT id, name, uuid "
                 " FROM tbl_contacts ")
        cursor.execute(query)
        rows_contacts = cursor.fetchall()

        # Build contact dictionary for quick lookup by ID
        contact_dict = dict()
        if rows_contacts is not None and len(rows_contacts) > 0:
            for row in rows_contacts:
                contact_dict[row[0]] = {"id": row[0],
                                        "name": row[1],
                                        "uuid": row[2]}

        # Query to retrieve all cost centers for reference
        query = (" SELECT id, name, uuid "
                 " FROM tbl_cost_centers ")
        cursor.execute(query)
        rows_cost_centers = cursor.fetchall()

        # Build cost center dictionary for quick lookup by ID
        cost_center_dict = dict()
        if rows_cost_centers is not None and len(rows_cost_centers) > 0:
            for row in rows_cost_centers:
                cost_center_dict[row[0]] = {"id": row[0],
                                            "name": row[1],
                                            "uuid": row[2]}

        # Initialize SVG dictionary for diagram references
        svg_dict = dict()

        query = (" SELECT id, name, uuid "
                 " FROM tbl_svgs ")
        cursor.execute(query)
        rows_svgs = cursor.fetchall()
        if rows_svgs is not None and len(rows_svgs) > 0:
            for row in rows_svgs:
                svg_dict[row[0]] = {"id": row[0],
                                    "name": row[1],
                                    "uuid": row[2]}

        query = (" SELECT id, name, uuid, "
                 "        address, postal_code, latitude, longitude, rated_capacity, rated_power, "
                 "        contact_id, cost_center_id, serial_number, svg_id, is_cost_data_displayed, "
                 "        phase_of_lifecycle, description "
                 " FROM tbl_microgrids ")
        params = []
        if search_query:
            query += " WHERE name LIKE %s OR address LIKE %s OR description LIKE %s "
            params = [f'%{search_query}%', f'%{search_query}%', f'%{search_query}%']
        query += " ORDER BY id "
        cursor.execute(query, params)
        rows_microgrids = cursor.fetchall()

        result = list()
        if rows_microgrids is not None and len(rows_microgrids) > 0:
            for row in rows_microgrids:
                meta_result = {"id": row[0],
                               "name": row[1],
                               "uuid": row[2],
                               "address": row[3],
                               "postal_code": row[4],
                               "latitude": row[5],
                               "longitude": row[6],
                               "rated_capacity": row[7],
                               "rated_power": row[8],
                               "contact": contact_dict.get(row[9], None),
                               "cost_center": cost_center_dict.get(row[10], None),
                               "serial_number": row[11],
                               "svg": svg_dict.get(row[12], None),
                               "is_cost_data_displayed": bool(row[13]),
                               "phase_of_lifecycle": row[14],
                               "description": row[15],
                               "qrcode": 'microgrid:' + row[2]}
                result.append(meta_result)

        cursor.close()
        cnx.close()
        resp.text = json.dumps(result)

    @staticmethod
    @user_logger

myems-api/core/photovoltaicpowerstation.py 1 location


        _ = req
        resp.status = falcon.HTTP_200

    @staticmethod
    def on_get(req, resp):
        """
        Handle GET requests to retrieve all photovoltaic power stations

        Returns a list of all photovoltaic power stations with their complete information including:
        - Station ID, name, and UUID
        - Associated contact and cost center information
        - Station specifications and parameters
        - Related equipment and meter associations
        - SVG diagram information for visualization

        Args:
            req: Falcon request object
            resp: Falcon response object
        """
        access_control(req)
        search_query = req.get_param('q', default=None)
        if search_query is not None and len(search_query.strip()) > 0:
            search_query = search_query.strip()
        else:
            search_query = ''

        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        query = (" SELECT id, name, uuid "
                 " FROM tbl_contacts ")
        cursor.execute(query)
        rows_contacts = cursor.fetchall()

        contact_dict = dict()
        if rows_contacts is not None and len(rows_contacts) > 0:
            for row in rows_contacts:
                contact_dict[row[0]] = {"id": row[0],
                                        "name": row[1],
                                        "uuid": row[2]}

        query = (" SELECT id, name, uuid "
                 " FROM tbl_cost_centers ")
        cursor.execute(query)
        rows_cost_centers = cursor.fetchall()

        cost_center_dict = dict()
        if rows_cost_centers is not None and len(rows_cost_centers) > 0:
            for row in rows_cost_centers:
                cost_center_dict[row[0]] = {"id": row[0],
                                            "name": row[1],
                                            "uuid": row[2]}

        svg_dict = dict()

        query = (" SELECT id, name, uuid "
                 " FROM tbl_svgs ")
        cursor.execute(query)
        rows_svgs = cursor.fetchall()
        if rows_svgs is not None and len(rows_svgs) > 0:
            for row in rows_svgs:
                svg_dict[row[0]] = {"id": row[0],
                                    "name": row[1],
                                    "uuid": row[2]}

        query = (" SELECT id, name, uuid, "
                 "        station_code, address, latitude, longitude, rated_capacity, rated_power, "
                 "        contact_id, cost_center_id, svg_id, is_cost_data_displayed, phase_of_lifecycle, description "
                 " FROM tbl_photovoltaic_power_stations ")
        params = []
        if search_query:
            query += " WHERE name LIKE %s OR address LIKE %s OR description LIKE %s "
            params = [f'%{search_query}%', f'%{search_query}%', f'%{search_query}%']
        query += " ORDER BY id "
        cursor.execute(query, params)
        rows_photovoltaic_power_stations = cursor.fetchall()

        result = list()
        if rows_photovoltaic_power_stations is not None and len(rows_photovoltaic_power_stations) > 0:
            for row in rows_photovoltaic_power_stations:
                meta_result = {"id": row[0],
                               "name": row[1],
                               "uuid": row[2],
                               "station_code": row[3],
                               "address": row[4],
                               "latitude": row[5],
                               "longitude": row[6],
                               "rated_capacity": row[7],
                               "rated_power": row[8],
                               "contact": contact_dict.get(row[9], None),
                               "cost_center": cost_center_dict.get(row[10], None),
                               "svg": svg_dict.get(row[11], None),
                               "is_cost_data_displayed": bool(row[12]),
                               "phase_of_lifecycle": row[13],
                               "description": row[14],
                               "qrcode": 'photovoltaicpowerstation:' + row[2]}
                result.append(meta_result)

        cursor.close()
        cnx.close()
        resp.text = json.dumps(result)

    @staticmethod
    @user_logger

		@@ 35-141 (lines=107) @@
32		_ = req
33		resp.status = falcon.HTTP_200
34
35		@staticmethod
36		def on_get(req, resp):
37		"""
38		Handle GET requests to retrieve all microgrids
39
40		Returns a list of all microgrids with their complete information including:
41		- Microgrid ID, name, and UUID
42		- Associated contact and cost center information
43		- Microgrid specifications and parameters
44		- Related equipment and meter associations
45		- SVG diagram information for visualization
46
47		Args:
48		req: Falcon request object
49		resp: Falcon response object
50		"""
51		access_control(req)
52
53		search_query = req.get_param('q', default=None)
54		if search_query is not None and len(search_query.strip()) > 0:
55		search_query = search_query.strip()
56		else:
57		search_query = ''
58
59		# Connect to database
60		cnx = mysql.connector.connect(**config.myems_system_db)
61		cursor = cnx.cursor()
62
63		# Query to retrieve all contacts for reference
64		query = (" SELECT id, name, uuid "
65		" FROM tbl_contacts ")
66		cursor.execute(query)
67		rows_contacts = cursor.fetchall()
68
69		# Build contact dictionary for quick lookup by ID
70		contact_dict = dict()
71		if rows_contacts is not None and len(rows_contacts) > 0:
72		for row in rows_contacts:
73		contact_dict[row[0]] = {"id": row[0],
74		"name": row[1],
75		"uuid": row[2]}
76
77		# Query to retrieve all cost centers for reference
78		query = (" SELECT id, name, uuid "
79		" FROM tbl_cost_centers ")
80		cursor.execute(query)
81		rows_cost_centers = cursor.fetchall()
82
83		# Build cost center dictionary for quick lookup by ID
84		cost_center_dict = dict()
85		if rows_cost_centers is not None and len(rows_cost_centers) > 0:
86		for row in rows_cost_centers:
87		cost_center_dict[row[0]] = {"id": row[0],
88		"name": row[1],
89		"uuid": row[2]}
90
91		# Initialize SVG dictionary for diagram references
92		svg_dict = dict()
93
94		query = (" SELECT id, name, uuid "
95		" FROM tbl_svgs ")
96		cursor.execute(query)
97		rows_svgs = cursor.fetchall()
98		if rows_svgs is not None and len(rows_svgs) > 0:
99		for row in rows_svgs:
100		svg_dict[row[0]] = {"id": row[0],
101		"name": row[1],
102		"uuid": row[2]}
103
104		query = (" SELECT id, name, uuid, "
105		" address, postal_code, latitude, longitude, rated_capacity, rated_power, "
106		" contact_id, cost_center_id, serial_number, svg_id, is_cost_data_displayed, "
107		" phase_of_lifecycle, description "
108		" FROM tbl_microgrids ")
109		params = []
110		if search_query:
111		query += " WHERE name LIKE %s OR address LIKE %s OR description LIKE %s "
112		params = [f'%{search_query}%', f'%{search_query}%', f'%{search_query}%']
113		query += " ORDER BY id "
114		cursor.execute(query, params)
115		rows_microgrids = cursor.fetchall()
116
117		result = list()
118		if rows_microgrids is not None and len(rows_microgrids) > 0:
119		for row in rows_microgrids:
120		meta_result = {"id": row[0],
121		"name": row[1],
122		"uuid": row[2],
123		"address": row[3],
124		"postal_code": row[4],
125		"latitude": row[5],
126		"longitude": row[6],
127		"rated_capacity": row[7],
128		"rated_power": row[8],
129		"contact": contact_dict.get(row[9], None),
130		"cost_center": cost_center_dict.get(row[10], None),
131		"serial_number": row[11],
132		"svg": svg_dict.get(row[12], None),
133		"is_cost_data_displayed": bool(row[13]),
134		"phase_of_lifecycle": row[14],
135		"description": row[15],
136		"qrcode": 'microgrid:' + row[2]}
137		result.append(meta_result)
138
139		cursor.close()
140		cnx.close()
141		resp.text = json.dumps(result)
142
143		@staticmethod
144		@user_logger

		@@ 35-132 (lines=98) @@
32		_ = req
33		resp.status = falcon.HTTP_200
34
35		@staticmethod
36		def on_get(req, resp):
37		"""
38		Handle GET requests to retrieve all photovoltaic power stations
39
40		Returns a list of all photovoltaic power stations with their complete information including:
41		- Station ID, name, and UUID
42		- Associated contact and cost center information
43		- Station specifications and parameters
44		- Related equipment and meter associations
45		- SVG diagram information for visualization
46
47		Args:
48		req: Falcon request object
49		resp: Falcon response object
50		"""
51		access_control(req)
52		search_query = req.get_param('q', default=None)
53		if search_query is not None and len(search_query.strip()) > 0:
54		search_query = search_query.strip()
55		else:
56		search_query = ''
57
58		cnx = mysql.connector.connect(**config.myems_system_db)
59		cursor = cnx.cursor()
60
61		query = (" SELECT id, name, uuid "
62		" FROM tbl_contacts ")
63		cursor.execute(query)
64		rows_contacts = cursor.fetchall()
65
66		contact_dict = dict()
67		if rows_contacts is not None and len(rows_contacts) > 0:
68		for row in rows_contacts:
69		contact_dict[row[0]] = {"id": row[0],
70		"name": row[1],
71		"uuid": row[2]}
72
73		query = (" SELECT id, name, uuid "
74		" FROM tbl_cost_centers ")
75		cursor.execute(query)
76		rows_cost_centers = cursor.fetchall()
77
78		cost_center_dict = dict()
79		if rows_cost_centers is not None and len(rows_cost_centers) > 0:
80		for row in rows_cost_centers:
81		cost_center_dict[row[0]] = {"id": row[0],
82		"name": row[1],
83		"uuid": row[2]}
84
85		svg_dict = dict()
86
87		query = (" SELECT id, name, uuid "
88		" FROM tbl_svgs ")
89		cursor.execute(query)
90		rows_svgs = cursor.fetchall()
91		if rows_svgs is not None and len(rows_svgs) > 0:
92		for row in rows_svgs:
93		svg_dict[row[0]] = {"id": row[0],
94		"name": row[1],
95		"uuid": row[2]}
96
97		query = (" SELECT id, name, uuid, "
98		" station_code, address, latitude, longitude, rated_capacity, rated_power, "
99		" contact_id, cost_center_id, svg_id, is_cost_data_displayed, phase_of_lifecycle, description "
100		" FROM tbl_photovoltaic_power_stations ")
101		params = []
102		if search_query:
103		query += " WHERE name LIKE %s OR address LIKE %s OR description LIKE %s "
104		params = [f'%{search_query}%', f'%{search_query}%', f'%{search_query}%']
105		query += " ORDER BY id "
106		cursor.execute(query, params)
107		rows_photovoltaic_power_stations = cursor.fetchall()
108
109		result = list()
110		if rows_photovoltaic_power_stations is not None and len(rows_photovoltaic_power_stations) > 0:
111		for row in rows_photovoltaic_power_stations:
112		meta_result = {"id": row[0],
113		"name": row[1],
114		"uuid": row[2],
115		"station_code": row[3],
116		"address": row[4],
117		"latitude": row[5],
118		"longitude": row[6],
119		"rated_capacity": row[7],
120		"rated_power": row[8],
121		"contact": contact_dict.get(row[9], None),
122		"cost_center": cost_center_dict.get(row[10], None),
123		"svg": svg_dict.get(row[11], None),
124		"is_cost_data_displayed": bool(row[12]),
125		"phase_of_lifecycle": row[13],
126		"description": row[14],
127		"qrcode": 'photovoltaicpowerstation:' + row[2]}
128		result.append(meta_result)
129
130		cursor.close()
131		cnx.close()
132		resp.text = json.dumps(result)
133
134		@staticmethod
135		@user_logger

MyEMS / myems

Code Duplication Length = 98-107 lines in 2 locations

myems-api/core/microgrid.py 1 location

myems-api/core/photovoltaicpowerstation.py 1 location