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


    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

myems-api/core/energystoragecontainer.py 1 location


        _=req
        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

		@@ 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

		@@ 714-800 (lines=87) @@
711		_=req
712		resp.status = falcon.HTTP_200
713		_=id_
714		@staticmethod
715		def on_get(req, resp, id_, bid):
716		access_control(req)
717		if not id_.isdigit() or int(id_) <= 0:
718		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
719		description='API.INVALID_ENERGY_STORAGE_CONTAINER_ID')
720		if not bid.isdigit() or int(bid) <= 0:
721		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
722		description='API.INVALID_ENERGY_STORAGE_CONTAINER_BATTERY_ID')
723
724		cnx = mysql.connector.connect(**config.myems_system_db)
725		cursor = cnx.cursor()
726
727		cursor.execute(" SELECT name "
728		" FROM tbl_energy_storage_containers "
729		" WHERE id = %s ", (id_,))
730		if cursor.fetchone() is None:
731		cursor.close()
732		cnx.close()
733		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
734		description='API.ENERGY_STORAGE_CONTAINER_NOT_FOUND')
735
736		# query energy storage power station dict
737		query = (" SELECT id, name, uuid "
738		" FROM tbl_energy_storage_containers ")
739		cursor.execute(query)
740		rows_energystoragecontainers = cursor.fetchall()
741
742		energy_storage_container_dict = dict()
743		if rows_energystoragecontainers is not None and len(rows_energystoragecontainers) > 0:
744		for row in rows_energystoragecontainers:
745		energy_storage_container_dict[row[0]] = {"id": row[0],
746		"name": row[1],
747		"uuid": row[2]}
748		# query meter dict
749		query = (" SELECT id, name, uuid "
750		" FROM tbl_meters ")
751		cursor.execute(query)
752		rows_meters = cursor.fetchall()
753
754		meter_dict = dict()
755		if rows_meters is not None and len(rows_meters) > 0:
756		for row in rows_meters:
757		meter_dict[row[0]] = {"id": row[0],
758		"name": row[1],
759		"uuid": row[2]}
760		# query point dict
761		query = (" SELECT id, name "
762		" FROM tbl_points ")
763		cursor.execute(query)
764		rows_points = cursor.fetchall()
765
766		point_dict = dict()
767		if rows_points is not None and len(rows_points) > 0:
768		for row in rows_points:
769		point_dict[row[0]] = {"id": row[0],
770		"name": row[1]}
771
772		query = (" SELECT id, name, uuid, energy_storage_container_id, "
773		" battery_state_point_id, soc_point_id, power_point_id, "
774		" charge_meter_id, discharge_meter_id, rated_capacity, rated_power, nominal_voltage "
775		" FROM tbl_energy_storage_containers_batteries "
776		" WHERE id = %s ")
777		cursor.execute(query, (bid,))
778		row = cursor.fetchone()
779		cursor.close()
780		cnx.close()
781
782		if row is None:
783		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
784		description='API.ENERGY_STORAGE_CONTAINER_BATTERY_NOT_FOUND')
785		else:
786		meta_result = {"id": row[0],
787		"name": row[1],
788		"uuid": row[2],
789		"energy_storage_container": energy_storage_container_dict.get(row[3]),
790		"battery_state_point": point_dict.get(row[4]),
791		"soc_point": point_dict.get(row[5]),
792		"power_point": point_dict.get(row[6]),
793		"charge_meter": meter_dict.get(row[7]),
794		"discharge_meter": meter_dict.get(row[8]),
795		"rated_capacity": row[9],
796		"rated_power": row[10],
797		"nominal_voltage": row[11]
798		}
799
800		resp.text = json.dumps(meta_result)
801
802		@staticmethod
803		@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