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


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

    @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


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

    @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

		@@ 845-931 (lines=87) @@
842		def on_options(req, resp, id_, bid):
843		resp.status = falcon.HTTP_200
844
845		@staticmethod
846		def on_get(req, resp, id_, bid):
847		access_control(req)
848		if not id_.isdigit() or int(id_) <= 0:
849		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
850		description='API.INVALID_ENERGY_STORAGE_CONTAINER_ID')
851		if not bid.isdigit() or int(bid) <= 0:
852		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
853		description='API.INVALID_ENERGY_STORAGE_CONTAINER_BATTERY_ID')
854
855		cnx = mysql.connector.connect(**config.myems_system_db)
856		cursor = cnx.cursor()
857
858		cursor.execute(" SELECT name "
859		" FROM tbl_energy_storage_containers "
860		" WHERE id = %s ", (id_,))
861		if cursor.fetchone() is None:
862		cursor.close()
863		cnx.close()
864		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
865		description='API.ENERGY_STORAGE_CONTAINER_NOT_FOUND')
866
867		# query energy storage power station dict
868		query = (" SELECT id, name, uuid "
869		" FROM tbl_energy_storage_containers ")
870		cursor.execute(query)
871		rows_energystoragecontainers = cursor.fetchall()
872
873		energy_storage_container_dict = dict()
874		if rows_energystoragecontainers is not None and len(rows_energystoragecontainers) > 0:
875		for row in rows_energystoragecontainers:
876		energy_storage_container_dict[row[0]] = {"id": row[0],
877		"name": row[1],
878		"uuid": row[2]}
879		# query meter dict
880		query = (" SELECT id, name, uuid "
881		" FROM tbl_meters ")
882		cursor.execute(query)
883		rows_meters = cursor.fetchall()
884
885		meter_dict = dict()
886		if rows_meters is not None and len(rows_meters) > 0:
887		for row in rows_meters:
888		meter_dict[row[0]] = {"id": row[0],
889		"name": row[1],
890		"uuid": row[2]}
891		# query point dict
892		query = (" SELECT id, name "
893		" FROM tbl_points ")
894		cursor.execute(query)
895		rows_points = cursor.fetchall()
896
897		point_dict = dict()
898		if rows_points is not None and len(rows_points) > 0:
899		for row in rows_points:
900		point_dict[row[0]] = {"id": row[0],
901		"name": row[1]}
902
903		query = (" SELECT id, name, uuid, energy_storage_container_id, "
904		" battery_state_point_id, soc_point_id, power_point_id, "
905		" charge_meter_id, discharge_meter_id, rated_capacity, rated_power, nominal_voltage "
906		" FROM tbl_energy_storage_containers_batteries "
907		" WHERE id = %s ")
908		cursor.execute(query, (bid,))
909		row = cursor.fetchone()
910		cursor.close()
911		cnx.close()
912
913		if row is None:
914		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
915		description='API.ENERGY_STORAGE_CONTAINER_BATTERY_NOT_FOUND')
916		else:
917		meta_result = {"id": row[0],
918		"name": row[1],
919		"uuid": row[2],
920		"energy_storage_container": energy_storage_container_dict.get(row[3]),
921		"battery_state_point": point_dict.get(row[4]),
922		"soc_point": point_dict.get(row[5]),
923		"power_point": point_dict.get(row[6]),
924		"charge_meter": meter_dict.get(row[7]),
925		"discharge_meter": meter_dict.get(row[8]),
926		"rated_capacity": row[9],
927		"rated_power": row[10],
928		"nominal_voltage": row[11]
929		}
930
931		resp.text = json.dumps(meta_result)
932
933		@staticmethod
934		@user_logger

		@@ 912-997 (lines=86) @@
909		def on_options(req, resp, id_, bid):
910		resp.status = falcon.HTTP_200
911
912		@staticmethod
913		def on_get(req, resp, id_, bid):
914		access_control(req)
915		if not id_.isdigit() or int(id_) <= 0:
916		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
917		description='API.INVALID_MICROGRID_ID')
918		if not bid.isdigit() or int(bid) <= 0:
919		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
920		description='API.INVALID_MICROGRID_BATTERY_ID')
921
922		cnx = mysql.connector.connect(**config.myems_system_db)
923		cursor = cnx.cursor()
924
925		cursor.execute(" SELECT name "
926		" FROM tbl_microgrids "
927		" WHERE id = %s ", (id_,))
928		if cursor.fetchone() is None:
929		cursor.close()
930		cnx.close()
931		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
932		description='API.MICROGRID_NOT_FOUND')
933
934		# query microgrid dict
935		query = (" SELECT id, name, uuid "
936		" FROM tbl_microgrids ")
937		cursor.execute(query)
938		rows_microgrids = cursor.fetchall()
939
940		microgrid_dict = dict()
941		if rows_microgrids is not None and len(rows_microgrids) > 0:
942		for row in rows_microgrids:
943		microgrid_dict[row[0]] = {"id": row[0],
944		"name": row[1],
945		"uuid": row[2]}
946		# query meter dict
947		query = (" SELECT id, name, uuid "
948		" FROM tbl_meters ")
949		cursor.execute(query)
950		rows_meters = cursor.fetchall()
951
952		meter_dict = dict()
953		if rows_meters is not None and len(rows_meters) > 0:
954		for row in rows_meters:
955		meter_dict[row[0]] = {"id": row[0],
956		"name": row[1],
957		"uuid": row[2]}
958		# query point dict
959		query = (" SELECT id, name "
960		" FROM tbl_points ")
961		cursor.execute(query)
962		rows_points = cursor.fetchall()
963
964		point_dict = dict()
965		if rows_points is not None and len(rows_points) > 0:
966		for row in rows_points:
967		point_dict[row[0]] = {"id": row[0],
968		"name": row[1]}
969
970		query = (" SELECT id, name, uuid, microgrid_id, "
971		" battery_state_point_id, soc_point_id, power_point_id, "
972		" charge_meter_id, discharge_meter_id, rated_capacity, rated_power, nominal_voltage "
973		" FROM tbl_microgrids_batteries "
974		" WHERE id = %s ")
975		cursor.execute(query, (bid,))
976		row = cursor.fetchone()
977		cursor.close()
978		cnx.close()
979
980		if row is None:
981		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
982		description='API.MICROGRID_BATTERY_NOT_FOUND')
983		else:
984		meta_result = {"id": row[0],
985		"name": row[1],
986		"uuid": row[2],
987		"microgrid": microgrid_dict.get(row[3]),
988		"battery_state_point": point_dict.get(row[4]),
989		"soc_point": point_dict.get(row[5]),
990		"power_point": point_dict.get(row[6]),
991		"charge_meter": meter_dict.get(row[7]),
992		"discharge_meter": meter_dict.get(row[8]),
993		"rated_capacity": row[9],
994		"rated_power": row[10],
995		"nominal_voltage": row[11]}
996
997		resp.text = json.dumps(meta_result)
998
999		@staticmethod
1000		@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