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

Code Duplication Length = 86-87 lines in 2 locations

myems-api/core/energystoragecontainer.py 1 location


        resp.status = falcon.HTTP_200
        _ = id_

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

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

        cursor.execute(" SELECT name "
                       " FROM tbl_energy_storage_containers "
                       " 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.ENERGY_STORAGE_CONTAINER_NOT_FOUND')

        # query energy storage power station dict
        query = (" SELECT id, name, uuid "
                 " FROM tbl_energy_storage_containers ")
        cursor.execute(query)
        rows_energystoragecontainers = cursor.fetchall()

        energy_storage_container_dict = dict()
        if rows_energystoragecontainers is not None and len(rows_energystoragecontainers) > 0:
            for row in rows_energystoragecontainers:
                energy_storage_container_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, energy_storage_container_id, "
                 "       battery_state_point_id, soc_point_id, power_point_id, "
                 "       charge_meter_id, discharge_meter_id, rated_capacity, rated_power, nominal_voltage "
                 " FROM tbl_energy_storage_containers_batteries "
                 " WHERE id = %s ")
        cursor.execute(query, (bid,))
        row = cursor.fetchone()
        cursor.close()
        cnx.close()

        if row is None:
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.ENERGY_STORAGE_CONTAINER_BATTERY_NOT_FOUND')
        else:
            meta_result = {"id": row[0],
                           "name": row[1],
                           "uuid": row[2],
                           "energy_storage_container": energy_storage_container_dict.get(row[3]),
                           "battery_state_point": point_dict.get(row[4]),
                           "soc_point": point_dict.get(row[5]),
                           "power_point": point_dict.get(row[6]),
                           "charge_meter": meter_dict.get(row[7]),
                           "discharge_meter": meter_dict.get(row[8]),
                           "rated_capacity": row[9],
                           "rated_power": row[10],
                           "nominal_voltage": row[11]
                           }

        resp.text = json.dumps(meta_result)

    @staticmethod
    @user_logger

myems-api/core/microgrid.py 1 location


        resp.status = falcon.HTTP_200
        _ = id_

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

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

        cursor.execute(" SELECT name "
                       " FROM tbl_microgrids "
                       " 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.MICROGRID_NOT_FOUND')

        # query microgrid dict
        query = (" SELECT id, name, uuid "
                 " FROM tbl_microgrids ")
        cursor.execute(query)
        rows_microgrids = cursor.fetchall()

        microgrid_dict = dict()
        if rows_microgrids is not None and len(rows_microgrids) > 0:
            for row in rows_microgrids:
                microgrid_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, microgrid_id, "
                 "       battery_state_point_id, soc_point_id, power_point_id, "
                 "       charge_meter_id, discharge_meter_id, rated_capacity, rated_power, nominal_voltage "
                 " FROM tbl_microgrids_batteries "
                 " WHERE id = %s ")
        cursor.execute(query, (bid,))
        row = cursor.fetchone()
        cursor.close()
        cnx.close()

        if row is None:
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.MICROGRID_BATTERY_NOT_FOUND')
        else:
            meta_result = {"id": row[0],
                           "name": row[1],
                           "uuid": row[2],
                           "microgrid": microgrid_dict.get(row[3]),
                           "battery_state_point": point_dict.get(row[4]),
                           "soc_point": point_dict.get(row[5]),
                           "power_point": point_dict.get(row[6]),
                           "charge_meter": meter_dict.get(row[7]),
                           "discharge_meter": meter_dict.get(row[8]),
                           "rated_capacity": row[9],
                           "rated_power": row[10],
                           "nominal_voltage": row[11]}

        resp.text = json.dumps(meta_result)

    @staticmethod
    @user_logger

		@@ 819-905 (lines=87) @@
816		resp.status = falcon.HTTP_200
817		_ = id_
818
819		@staticmethod
820		def on_get(req, resp, id_, bid):
821		access_control(req)
822		if not id_.isdigit() or int(id_) <= 0:
823		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
824		description='API.INVALID_ENERGY_STORAGE_CONTAINER_ID')
825		if not bid.isdigit() or int(bid) <= 0:
826		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
827		description='API.INVALID_ENERGY_STORAGE_CONTAINER_BATTERY_ID')
828
829		cnx = mysql.connector.connect(**config.myems_system_db)
830		cursor = cnx.cursor()
831
832		cursor.execute(" SELECT name "
833		" FROM tbl_energy_storage_containers "
834		" WHERE id = %s ", (id_,))
835		if cursor.fetchone() is None:
836		cursor.close()
837		cnx.close()
838		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
839		description='API.ENERGY_STORAGE_CONTAINER_NOT_FOUND')
840
841		# query energy storage power station dict
842		query = (" SELECT id, name, uuid "
843		" FROM tbl_energy_storage_containers ")
844		cursor.execute(query)
845		rows_energystoragecontainers = cursor.fetchall()
846
847		energy_storage_container_dict = dict()
848		if rows_energystoragecontainers is not None and len(rows_energystoragecontainers) > 0:
849		for row in rows_energystoragecontainers:
850		energy_storage_container_dict[row[0]] = {"id": row[0],
851		"name": row[1],
852		"uuid": row[2]}
853		# query meter dict
854		query = (" SELECT id, name, uuid "
855		" FROM tbl_meters ")
856		cursor.execute(query)
857		rows_meters = cursor.fetchall()
858
859		meter_dict = dict()
860		if rows_meters is not None and len(rows_meters) > 0:
861		for row in rows_meters:
862		meter_dict[row[0]] = {"id": row[0],
863		"name": row[1],
864		"uuid": row[2]}
865		# query point dict
866		query = (" SELECT id, name "
867		" FROM tbl_points ")
868		cursor.execute(query)
869		rows_points = cursor.fetchall()
870
871		point_dict = dict()
872		if rows_points is not None and len(rows_points) > 0:
873		for row in rows_points:
874		point_dict[row[0]] = {"id": row[0],
875		"name": row[1]}
876
877		query = (" SELECT id, name, uuid, energy_storage_container_id, "
878		" battery_state_point_id, soc_point_id, power_point_id, "
879		" charge_meter_id, discharge_meter_id, rated_capacity, rated_power, nominal_voltage "
880		" FROM tbl_energy_storage_containers_batteries "
881		" WHERE id = %s ")
882		cursor.execute(query, (bid,))
883		row = cursor.fetchone()
884		cursor.close()
885		cnx.close()
886
887		if row is None:
888		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
889		description='API.ENERGY_STORAGE_CONTAINER_BATTERY_NOT_FOUND')
890		else:
891		meta_result = {"id": row[0],
892		"name": row[1],
893		"uuid": row[2],
894		"energy_storage_container": energy_storage_container_dict.get(row[3]),
895		"battery_state_point": point_dict.get(row[4]),
896		"soc_point": point_dict.get(row[5]),
897		"power_point": point_dict.get(row[6]),
898		"charge_meter": meter_dict.get(row[7]),
899		"discharge_meter": meter_dict.get(row[8]),
900		"rated_capacity": row[9],
901		"rated_power": row[10],
902		"nominal_voltage": row[11]
903		}
904
905		resp.text = json.dumps(meta_result)
906
907		@staticmethod
908		@user_logger

		@@ 978-1063 (lines=86) @@
975		resp.status = falcon.HTTP_200
976		_ = id_
977
978		@staticmethod
979		def on_get(req, resp, id_, bid):
980		access_control(req)
981		if not id_.isdigit() or int(id_) <= 0:
982		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
983		description='API.INVALID_MICROGRID_ID')
984		if not bid.isdigit() or int(bid) <= 0:
985		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
986		description='API.INVALID_MICROGRID_BATTERY_ID')
987
988		cnx = mysql.connector.connect(**config.myems_system_db)
989		cursor = cnx.cursor()
990
991		cursor.execute(" SELECT name "
992		" FROM tbl_microgrids "
993		" WHERE id = %s ", (id_,))
994		if cursor.fetchone() is None:
995		cursor.close()
996		cnx.close()
997		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
998		description='API.MICROGRID_NOT_FOUND')
999
1000		# query microgrid dict
1001		query = (" SELECT id, name, uuid "
1002		" FROM tbl_microgrids ")
1003		cursor.execute(query)
1004		rows_microgrids = cursor.fetchall()
1005
1006		microgrid_dict = dict()
1007		if rows_microgrids is not None and len(rows_microgrids) > 0:
1008		for row in rows_microgrids:
1009		microgrid_dict[row[0]] = {"id": row[0],
1010		"name": row[1],
1011		"uuid": row[2]}
1012		# query meter dict
1013		query = (" SELECT id, name, uuid "
1014		" FROM tbl_meters ")
1015		cursor.execute(query)
1016		rows_meters = cursor.fetchall()
1017
1018		meter_dict = dict()
1019		if rows_meters is not None and len(rows_meters) > 0:
1020		for row in rows_meters:
1021		meter_dict[row[0]] = {"id": row[0],
1022		"name": row[1],
1023		"uuid": row[2]}
1024		# query point dict
1025		query = (" SELECT id, name "
1026		" FROM tbl_points ")
1027		cursor.execute(query)
1028		rows_points = cursor.fetchall()
1029
1030		point_dict = dict()
1031		if rows_points is not None and len(rows_points) > 0:
1032		for row in rows_points:
1033		point_dict[row[0]] = {"id": row[0],
1034		"name": row[1]}
1035
1036		query = (" SELECT id, name, uuid, microgrid_id, "
1037		" battery_state_point_id, soc_point_id, power_point_id, "
1038		" charge_meter_id, discharge_meter_id, rated_capacity, rated_power, nominal_voltage "
1039		" FROM tbl_microgrids_batteries "
1040		" WHERE id = %s ")
1041		cursor.execute(query, (bid,))
1042		row = cursor.fetchone()
1043		cursor.close()
1044		cnx.close()
1045
1046		if row is None:
1047		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1048		description='API.MICROGRID_BATTERY_NOT_FOUND')
1049		else:
1050		meta_result = {"id": row[0],
1051		"name": row[1],
1052		"uuid": row[2],
1053		"microgrid": microgrid_dict.get(row[3]),
1054		"battery_state_point": point_dict.get(row[4]),
1055		"soc_point": point_dict.get(row[5]),
1056		"power_point": point_dict.get(row[6]),
1057		"charge_meter": meter_dict.get(row[7]),
1058		"discharge_meter": meter_dict.get(row[8]),
1059		"rated_capacity": row[9],
1060		"rated_power": row[10],
1061		"nominal_voltage": row[11]}
1062
1063		resp.text = json.dumps(meta_result)
1064
1065		@staticmethod
1066		@user_logger

MyEMS / myems

Code Duplication Length = 86-87 lines in 2 locations

myems-api/core/energystoragecontainer.py 1 location

myems-api/core/microgrid.py 1 location