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

Code Duplication Length = 113-117 lines in 2 locations

myems-api/core/photovoltaicpowerstation.py 2 locations


    def on_options(req, resp, id_, gid):
        resp.status = falcon.HTTP_200

    @staticmethod
    def on_get(req, resp, id_, gid):
        access_control(req)
        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')
        if not gid.isdigit() or int(gid) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_PHOTOVOLTAIC_POWER_STATION_GRID_ID')

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

        cursor.execute(" SELECT name "
                       " FROM tbl_photovoltaic_power_stations "
                       " WHERE id = %s ", (id_,))
        if cursor.fetchone() is None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.PHOTOVOLTAIC_POWER_STATION_NOT_FOUND')

        query = (" SELECT id, name, uuid "
                 " FROM tbl_photovoltaic_power_stations ")
        cursor.execute(query)
        rows_photovoltaic_power_stations = cursor.fetchall()

        photovoltaic_power_station_dict = dict()
        if rows_photovoltaic_power_stations is not None and len(rows_photovoltaic_power_stations) > 0:
            for row in rows_photovoltaic_power_stations:
                photovoltaic_power_station_dict[row[0]] = {"id": row[0],
                                                           "name": row[1],
                                                           "uuid": row[2]}
        # query meter dict
        query = (" SELECT id, name, uuid "
                 " FROM tbl_meters ")
        cursor.execute(query)
        rows_meters = cursor.fetchall()

        meter_dict = dict()
        if rows_meters is not None and len(rows_meters) > 0:
            for row in rows_meters:
                meter_dict[row[0]] = {"id": row[0],
                                      "name": row[1],
                                      "uuid": row[2]}
        # query point dict
        query = (" SELECT id, name "
                 " FROM tbl_points ")
        cursor.execute(query)
        rows_points = cursor.fetchall()

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

        query = (" SELECT id, name, uuid, "
                 "        photovoltaic_power_station_id, "
                 "        power_point_id, "
                 "        buy_meter_id, "
                 "        sell_meter_id, "
                 "        capacity, "
                 "        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 id = %s ")
        cursor.execute(query, (gid,))
        row = cursor.fetchone()
        cursor.close()
        cnx.close()

        if row is None:
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.PHOTOVOLTAIC_POWER_STATION_GRID_NOT_FOUND')
        else:
            meta_result = {"id": row[0],
                           "name": row[1],
                           "uuid": row[2],
                           "photovoltaic_power_station": photovoltaic_power_station_dict.get(row[3]),
                           "power_point": point_dict.get(row[4]),
                           "buy_meter": meter_dict.get(row[5]),
                           "sell_meter": meter_dict.get(row[6]),
                           "capacity": row[7],
                           "total_active_power_point": point_dict.get(row[8]),
                           "active_power_a_point": point_dict.get(row[9]),
                           "active_power_b_point": point_dict.get(row[10]),
                           "active_power_c_point": point_dict.get(row[11]),
                           "total_reactive_power_point": point_dict.get(row[12]),
                           "reactive_power_a_point": point_dict.get(row[13]),
                           "reactive_power_b_point": point_dict.get(row[14]),
                           "reactive_power_c_point": point_dict.get(row[15]),
                           "total_apparent_power_point": point_dict.get(row[16]),
                           "apparent_power_a_point": point_dict.get(row[17]),
                           "apparent_power_b_point": point_dict.get(row[18]),
                           "apparent_power_c_point": point_dict.get(row[19]),
                           "total_power_factor_point": point_dict.get(row[20]),
                           "active_energy_import_point": point_dict.get(row[21]),
                           "active_energy_export_point": point_dict.get(row[22]),
                           "active_energy_net_point_id": point_dict.get(row[23]),
                           }

        resp.text = json.dumps(meta_result)

    @staticmethod
    @user_logger

    def on_options(req, resp, id_, lid):
        resp.status = falcon.HTTP_200

    @staticmethod
    def on_get(req, resp, id_, lid):
        access_control(req)
        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')
        if not lid.isdigit() or int(lid) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_PHOTOVOLTAIC_POWER_STATION_LOAD_ID')

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

        cursor.execute(" SELECT name "
                       " FROM tbl_photovoltaic_power_stations "
                       " WHERE id = %s ", (id_,))
        if cursor.fetchone() is None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.PHOTOVOLTAIC_POWER_STATION_NOT_FOUND')

        query = (" SELECT id, name, uuid "
                 " FROM tbl_photovoltaic_power_stations ")
        cursor.execute(query)
        rows_photovoltaic_power_stations = cursor.fetchall()

        photovoltaic_power_station_dict = dict()
        if rows_photovoltaic_power_stations is not None and len(rows_photovoltaic_power_stations) > 0:
            for row in rows_photovoltaic_power_stations:
                photovoltaic_power_station_dict[row[0]] = {"id": row[0],
                                                           "name": row[1],
                                                           "uuid": row[2]}
        # query meter dict
        query = (" SELECT id, name, uuid "
                 " FROM tbl_meters ")
        cursor.execute(query)
        rows_meters = cursor.fetchall()

        meter_dict = dict()
        if rows_meters is not None and len(rows_meters) > 0:
            for row in rows_meters:
                meter_dict[row[0]] = {"id": row[0],
                                      "name": row[1],
                                      "uuid": row[2]}
        # query point dict
        query = (" SELECT id, name "
                 " FROM tbl_points ")
        cursor.execute(query)
        rows_points = cursor.fetchall()

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

        query = (" SELECT id, name, uuid, "
                 "        photovoltaic_power_station_id, "
                 "        power_point_id, meter_id, "
                 "        rated_input_power, "
                 "        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 id = %s ")
        cursor.execute(query, (lid,))
        row = cursor.fetchone()
        cursor.close()
        cnx.close()

        if row is None:
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.PHOTOVOLTAIC_POWER_STATION_LOAD_NOT_FOUND')
        else:
            meta_result = {"id": row[0],
                           "name": row[1],
                           "uuid": row[2],
                           "photovoltaic_power_station": photovoltaic_power_station_dict.get(row[3]),
                           "power_point": point_dict.get(row[4]),
                           "meter": meter_dict.get(row[5]),
                           "rated_input_power": row[6],
                           "total_active_power_point": point_dict.get(row[7]),
                           "active_power_a_point": point_dict.get(row[8]),
                           "active_power_b_point": point_dict.get(row[9]),
                           "active_power_c_point": point_dict.get(row[10]),
                           "total_reactive_power_point": point_dict.get(row[11]),
                           "reactive_power_a_point": point_dict.get(row[12]),
                           "reactive_power_b_point": point_dict.get(row[13]),
                           "reactive_power_c_point": point_dict.get(row[14]),
                           "total_apparent_power_point": point_dict.get(row[15]),
                           "apparent_power_a_point": point_dict.get(row[16]),
                           "apparent_power_b_point": point_dict.get(row[17]),
                           "apparent_power_c_point": point_dict.get(row[18]),
                           "total_power_factor_point": point_dict.get(row[19]),
                           "active_energy_import_point": point_dict.get(row[20]),
                           "active_energy_export_point": point_dict.get(row[21]),
                           "active_energy_net_point": point_dict.get(row[22])}

        resp.text = json.dumps(meta_result)

    @staticmethod
    @user_logger

		@@ 1343-1459 (lines=117) @@
1340		def on_options(req, resp, id_, gid):
1341		resp.status = falcon.HTTP_200
1342
1343		@staticmethod
1344		def on_get(req, resp, id_, gid):
1345		access_control(req)
1346		if not id_.isdigit() or int(id_) <= 0:
1347		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
1348		description='API.INVALID_PHOTOVOLTAIC_POWER_STATION_ID')
1349		if not gid.isdigit() or int(gid) <= 0:
1350		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
1351		description='API.INVALID_PHOTOVOLTAIC_POWER_STATION_GRID_ID')
1352
1353		cnx = mysql.connector.connect(**config.myems_system_db)
1354		cursor = cnx.cursor()
1355
1356		cursor.execute(" SELECT name "
1357		" FROM tbl_photovoltaic_power_stations "
1358		" WHERE id = %s ", (id_,))
1359		if cursor.fetchone() is None:
1360		cursor.close()
1361		cnx.close()
1362		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1363		description='API.PHOTOVOLTAIC_POWER_STATION_NOT_FOUND')
1364
1365		query = (" SELECT id, name, uuid "
1366		" FROM tbl_photovoltaic_power_stations ")
1367		cursor.execute(query)
1368		rows_photovoltaic_power_stations = cursor.fetchall()
1369
1370		photovoltaic_power_station_dict = dict()
1371		if rows_photovoltaic_power_stations is not None and len(rows_photovoltaic_power_stations) > 0:
1372		for row in rows_photovoltaic_power_stations:
1373		photovoltaic_power_station_dict[row[0]] = {"id": row[0],
1374		"name": row[1],
1375		"uuid": row[2]}
1376		# query meter dict
1377		query = (" SELECT id, name, uuid "
1378		" FROM tbl_meters ")
1379		cursor.execute(query)
1380		rows_meters = cursor.fetchall()
1381
1382		meter_dict = dict()
1383		if rows_meters is not None and len(rows_meters) > 0:
1384		for row in rows_meters:
1385		meter_dict[row[0]] = {"id": row[0],
1386		"name": row[1],
1387		"uuid": row[2]}
1388		# query point dict
1389		query = (" SELECT id, name "
1390		" FROM tbl_points ")
1391		cursor.execute(query)
1392		rows_points = cursor.fetchall()
1393
1394		point_dict = dict()
1395		if rows_points is not None and len(rows_points) > 0:
1396		for row in rows_points:
1397		point_dict[row[0]] = {"id": row[0],
1398		"name": row[1]}
1399
1400		query = (" SELECT id, name, uuid, "
1401		" photovoltaic_power_station_id, "
1402		" power_point_id, "
1403		" buy_meter_id, "
1404		" sell_meter_id, "
1405		" capacity, "
1406		" total_active_power_point_id, "
1407		" active_power_a_point_id, "
1408		" active_power_b_point_id, "
1409		" active_power_c_point_id, "
1410		" total_reactive_power_point_id, "
1411		" reactive_power_a_point_id, "
1412		" reactive_power_b_point_id, "
1413		" reactive_power_c_point_id, "
1414		" total_apparent_power_point_id, "
1415		" apparent_power_a_point_id, "
1416		" apparent_power_b_point_id, "
1417		" apparent_power_c_point_id, "
1418		" total_power_factor_point_id, "
1419		" active_energy_import_point_id, "
1420		" active_energy_export_point_id, "
1421		" active_energy_net_point_id "
1422		" FROM tbl_photovoltaic_power_stations_grids "
1423		" WHERE id = %s ")
1424		cursor.execute(query, (gid,))
1425		row = cursor.fetchone()
1426		cursor.close()
1427		cnx.close()
1428
1429		if row is None:
1430		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1431		description='API.PHOTOVOLTAIC_POWER_STATION_GRID_NOT_FOUND')
1432		else:
1433		meta_result = {"id": row[0],
1434		"name": row[1],
1435		"uuid": row[2],
1436		"photovoltaic_power_station": photovoltaic_power_station_dict.get(row[3]),
1437		"power_point": point_dict.get(row[4]),
1438		"buy_meter": meter_dict.get(row[5]),
1439		"sell_meter": meter_dict.get(row[6]),
1440		"capacity": row[7],
1441		"total_active_power_point": point_dict.get(row[8]),
1442		"active_power_a_point": point_dict.get(row[9]),
1443		"active_power_b_point": point_dict.get(row[10]),
1444		"active_power_c_point": point_dict.get(row[11]),
1445		"total_reactive_power_point": point_dict.get(row[12]),
1446		"reactive_power_a_point": point_dict.get(row[13]),
1447		"reactive_power_b_point": point_dict.get(row[14]),
1448		"reactive_power_c_point": point_dict.get(row[15]),
1449		"total_apparent_power_point": point_dict.get(row[16]),
1450		"apparent_power_a_point": point_dict.get(row[17]),
1451		"apparent_power_b_point": point_dict.get(row[18]),
1452		"apparent_power_c_point": point_dict.get(row[19]),
1453		"total_power_factor_point": point_dict.get(row[20]),
1454		"active_energy_import_point": point_dict.get(row[21]),
1455		"active_energy_export_point": point_dict.get(row[22]),
1456		"active_energy_net_point_id": point_dict.get(row[23]),
1457		}
1458
1459		resp.text = json.dumps(meta_result)
1460
1461		@staticmethod
1462		@user_logger
		@@ 4368-4480 (lines=113) @@
4365		def on_options(req, resp, id_, lid):
4366		resp.status = falcon.HTTP_200
4367
4368		@staticmethod
4369		def on_get(req, resp, id_, lid):
4370		access_control(req)
4371		if not id_.isdigit() or int(id_) <= 0:
4372		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
4373		description='API.INVALID_PHOTOVOLTAIC_POWER_STATION_ID')
4374		if not lid.isdigit() or int(lid) <= 0:
4375		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
4376		description='API.INVALID_PHOTOVOLTAIC_POWER_STATION_LOAD_ID')
4377
4378		cnx = mysql.connector.connect(**config.myems_system_db)
4379		cursor = cnx.cursor()
4380
4381		cursor.execute(" SELECT name "
4382		" FROM tbl_photovoltaic_power_stations "
4383		" WHERE id = %s ", (id_,))
4384		if cursor.fetchone() is None:
4385		cursor.close()
4386		cnx.close()
4387		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
4388		description='API.PHOTOVOLTAIC_POWER_STATION_NOT_FOUND')
4389
4390		query = (" SELECT id, name, uuid "
4391		" FROM tbl_photovoltaic_power_stations ")
4392		cursor.execute(query)
4393		rows_photovoltaic_power_stations = cursor.fetchall()
4394
4395		photovoltaic_power_station_dict = dict()
4396		if rows_photovoltaic_power_stations is not None and len(rows_photovoltaic_power_stations) > 0:
4397		for row in rows_photovoltaic_power_stations:
4398		photovoltaic_power_station_dict[row[0]] = {"id": row[0],
4399		"name": row[1],
4400		"uuid": row[2]}
4401		# query meter dict
4402		query = (" SELECT id, name, uuid "
4403		" FROM tbl_meters ")
4404		cursor.execute(query)
4405		rows_meters = cursor.fetchall()
4406
4407		meter_dict = dict()
4408		if rows_meters is not None and len(rows_meters) > 0:
4409		for row in rows_meters:
4410		meter_dict[row[0]] = {"id": row[0],
4411		"name": row[1],
4412		"uuid": row[2]}
4413		# query point dict
4414		query = (" SELECT id, name "
4415		" FROM tbl_points ")
4416		cursor.execute(query)
4417		rows_points = cursor.fetchall()
4418
4419		point_dict = dict()
4420		if rows_points is not None and len(rows_points) > 0:
4421		for row in rows_points:
4422		point_dict[row[0]] = {"id": row[0],
4423		"name": row[1]}
4424
4425		query = (" SELECT id, name, uuid, "
4426		" photovoltaic_power_station_id, "
4427		" power_point_id, meter_id, "
4428		" rated_input_power, "
4429		" total_active_power_point_id, "
4430		" active_power_a_point_id, "
4431		" active_power_b_point_id, "
4432		" active_power_c_point_id, "
4433		" total_reactive_power_point_id, "
4434		" reactive_power_a_point_id, "
4435		" reactive_power_b_point_id, "
4436		" reactive_power_c_point_id, "
4437		" total_apparent_power_point_id, "
4438		" apparent_power_a_point_id, "
4439		" apparent_power_b_point_id, "
4440		" apparent_power_c_point_id, "
4441		" total_power_factor_point_id, "
4442		" active_energy_import_point_id, "
4443		" active_energy_export_point_id, "
4444		" active_energy_net_point_id "
4445		" FROM tbl_photovoltaic_power_stations_loads "
4446		" WHERE id = %s ")
4447		cursor.execute(query, (lid,))
4448		row = cursor.fetchone()
4449		cursor.close()
4450		cnx.close()
4451
4452		if row is None:
4453		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
4454		description='API.PHOTOVOLTAIC_POWER_STATION_LOAD_NOT_FOUND')
4455		else:
4456		meta_result = {"id": row[0],
4457		"name": row[1],
4458		"uuid": row[2],
4459		"photovoltaic_power_station": photovoltaic_power_station_dict.get(row[3]),
4460		"power_point": point_dict.get(row[4]),
4461		"meter": meter_dict.get(row[5]),
4462		"rated_input_power": row[6],
4463		"total_active_power_point": point_dict.get(row[7]),
4464		"active_power_a_point": point_dict.get(row[8]),
4465		"active_power_b_point": point_dict.get(row[9]),
4466		"active_power_c_point": point_dict.get(row[10]),
4467		"total_reactive_power_point": point_dict.get(row[11]),
4468		"reactive_power_a_point": point_dict.get(row[12]),
4469		"reactive_power_b_point": point_dict.get(row[13]),
4470		"reactive_power_c_point": point_dict.get(row[14]),
4471		"total_apparent_power_point": point_dict.get(row[15]),
4472		"apparent_power_a_point": point_dict.get(row[16]),
4473		"apparent_power_b_point": point_dict.get(row[17]),
4474		"apparent_power_c_point": point_dict.get(row[18]),
4475		"total_power_factor_point": point_dict.get(row[19]),
4476		"active_energy_import_point": point_dict.get(row[20]),
4477		"active_energy_export_point": point_dict.get(row[21]),
4478		"active_energy_net_point": point_dict.get(row[22])}
4479
4480		resp.text = json.dumps(meta_result)
4481
4482		@staticmethod
4483		@user_logger

MyEMS / myems

Code Duplication Length = 113-117 lines in 2 locations

myems-api/core/photovoltaicpowerstation.py 2 locations