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

Code Duplication Length = 86-87 lines in 2 locations

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

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

		@@ 934-1019 (lines=86) @@
931		resp.status = falcon.HTTP_200
932		_ = id_
933
934		@staticmethod
935		def on_get(req, resp, id_, bid):
936		access_control(req)
937		if not id_.isdigit() or int(id_) <= 0:
938		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
939		description='API.INVALID_MICROGRID_ID')
940		if not bid.isdigit() or int(bid) <= 0:
941		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
942		description='API.INVALID_MICROGRID_BATTERY_ID')
943
944		cnx = mysql.connector.connect(**config.myems_system_db)
945		cursor = cnx.cursor()
946
947		cursor.execute(" SELECT name "
948		" FROM tbl_microgrids "
949		" WHERE id = %s ", (id_,))
950		if cursor.fetchone() is None:
951		cursor.close()
952		cnx.close()
953		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
954		description='API.MICROGRID_NOT_FOUND')
955
956		# query microgrid dict
957		query = (" SELECT id, name, uuid "
958		" FROM tbl_microgrids ")
959		cursor.execute(query)
960		rows_microgrids = cursor.fetchall()
961
962		microgrid_dict = dict()
963		if rows_microgrids is not None and len(rows_microgrids) > 0:
964		for row in rows_microgrids:
965		microgrid_dict[row[0]] = {"id": row[0],
966		"name": row[1],
967		"uuid": row[2]}
968		# query meter dict
969		query = (" SELECT id, name, uuid "
970		" FROM tbl_meters ")
971		cursor.execute(query)
972		rows_meters = cursor.fetchall()
973
974		meter_dict = dict()
975		if rows_meters is not None and len(rows_meters) > 0:
976		for row in rows_meters:
977		meter_dict[row[0]] = {"id": row[0],
978		"name": row[1],
979		"uuid": row[2]}
980		# query point dict
981		query = (" SELECT id, name "
982		" FROM tbl_points ")
983		cursor.execute(query)
984		rows_points = cursor.fetchall()
985
986		point_dict = dict()
987		if rows_points is not None and len(rows_points) > 0:
988		for row in rows_points:
989		point_dict[row[0]] = {"id": row[0],
990		"name": row[1]}
991
992		query = (" SELECT id, name, uuid, microgrid_id, "
993		" battery_state_point_id, soc_point_id, power_point_id, "
994		" charge_meter_id, discharge_meter_id, rated_capacity, rated_power, nominal_voltage "
995		" FROM tbl_microgrids_batteries "
996		" WHERE id = %s ")
997		cursor.execute(query, (bid,))
998		row = cursor.fetchone()
999		cursor.close()
1000		cnx.close()
1001
1002		if row is None:
1003		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1004		description='API.MICROGRID_BATTERY_NOT_FOUND')
1005		else:
1006		meta_result = {"id": row[0],
1007		"name": row[1],
1008		"uuid": row[2],
1009		"microgrid": microgrid_dict.get(row[3]),
1010		"battery_state_point": point_dict.get(row[4]),
1011		"soc_point": point_dict.get(row[5]),
1012		"power_point": point_dict.get(row[6]),
1013		"charge_meter": meter_dict.get(row[7]),
1014		"discharge_meter": meter_dict.get(row[8]),
1015		"rated_capacity": row[9],
1016		"rated_power": row[10],
1017		"nominal_voltage": row[11]}
1018
1019		resp.text = json.dumps(meta_result)
1020
1021		@staticmethod
1022		@user_logger

		@@ 767-853 (lines=87) @@
764		resp.status = falcon.HTTP_200
765		_ = id_
766
767		@staticmethod
768		def on_get(req, resp, id_, bid):
769		access_control(req)
770		if not id_.isdigit() or int(id_) <= 0:
771		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
772		description='API.INVALID_ENERGY_STORAGE_CONTAINER_ID')
773		if not bid.isdigit() or int(bid) <= 0:
774		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
775		description='API.INVALID_ENERGY_STORAGE_CONTAINER_BATTERY_ID')
776
777		cnx = mysql.connector.connect(**config.myems_system_db)
778		cursor = cnx.cursor()
779
780		cursor.execute(" SELECT name "
781		" FROM tbl_energy_storage_containers "
782		" WHERE id = %s ", (id_,))
783		if cursor.fetchone() is None:
784		cursor.close()
785		cnx.close()
786		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
787		description='API.ENERGY_STORAGE_CONTAINER_NOT_FOUND')
788
789		# query energy storage power station dict
790		query = (" SELECT id, name, uuid "
791		" FROM tbl_energy_storage_containers ")
792		cursor.execute(query)
793		rows_energystoragecontainers = cursor.fetchall()
794
795		energy_storage_container_dict = dict()
796		if rows_energystoragecontainers is not None and len(rows_energystoragecontainers) > 0:
797		for row in rows_energystoragecontainers:
798		energy_storage_container_dict[row[0]] = {"id": row[0],
799		"name": row[1],
800		"uuid": row[2]}
801		# query meter dict
802		query = (" SELECT id, name, uuid "
803		" FROM tbl_meters ")
804		cursor.execute(query)
805		rows_meters = cursor.fetchall()
806
807		meter_dict = dict()
808		if rows_meters is not None and len(rows_meters) > 0:
809		for row in rows_meters:
810		meter_dict[row[0]] = {"id": row[0],
811		"name": row[1],
812		"uuid": row[2]}
813		# query point dict
814		query = (" SELECT id, name "
815		" FROM tbl_points ")
816		cursor.execute(query)
817		rows_points = cursor.fetchall()
818
819		point_dict = dict()
820		if rows_points is not None and len(rows_points) > 0:
821		for row in rows_points:
822		point_dict[row[0]] = {"id": row[0],
823		"name": row[1]}
824
825		query = (" SELECT id, name, uuid, energy_storage_container_id, "
826		" battery_state_point_id, soc_point_id, power_point_id, "
827		" charge_meter_id, discharge_meter_id, rated_capacity, rated_power, nominal_voltage "
828		" FROM tbl_energy_storage_containers_batteries "
829		" WHERE id = %s ")
830		cursor.execute(query, (bid,))
831		row = cursor.fetchone()
832		cursor.close()
833		cnx.close()
834
835		if row is None:
836		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
837		description='API.ENERGY_STORAGE_CONTAINER_BATTERY_NOT_FOUND')
838		else:
839		meta_result = {"id": row[0],
840		"name": row[1],
841		"uuid": row[2],
842		"energy_storage_container": energy_storage_container_dict.get(row[3]),
843		"battery_state_point": point_dict.get(row[4]),
844		"soc_point": point_dict.get(row[5]),
845		"power_point": point_dict.get(row[6]),
846		"charge_meter": meter_dict.get(row[7]),
847		"discharge_meter": meter_dict.get(row[8]),
848		"rated_capacity": row[9],
849		"rated_power": row[10],
850		"nominal_voltage": row[11]
851		}
852
853		resp.text = json.dumps(meta_result)
854
855		@staticmethod
856		@user_logger

MyEMS / myems

Code Duplication Length = 86-87 lines in 2 locations

myems-api/core/microgrid.py 1 location

myems-api/core/energystoragecontainer.py 1 location