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

Code Duplication Length = 111-115 lines in 2 locations

myems-api/core/energystoragepowerstation.py 1 location


        resp.status = falcon.HTTP_200


class EnergyStoragePowerStationContainerCollection:
    def __init__(self):
        """Initializes Class"""
        pass

    @staticmethod
    def on_options(req, resp, id_):
        _ = req
        resp.status = falcon.HTTP_200
        _ = id_

    @staticmethod
    def on_get(req, resp, id_):
        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_POWER_STATION_ID')

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

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

        query = (" SELECT s.id, s.name, s.uuid "
                 " FROM tbl_energy_storage_power_stations e, "
                 "      tbl_energy_storage_power_stations_containers es, tbl_energy_storage_containers s "
                 " WHERE es.energy_storage_power_station_id = e.id "
                 "       AND s.id = es.energy_storage_container_id "
                 "       AND e.id = %s "
                 " ORDER BY s.id ")
        cursor.execute(query, (id_,))
        rows = cursor.fetchall()

        result = list()
        if rows is not None and len(rows) > 0:
            for row in rows:
                meta_result = {"id": row[0], "name": row[1], "uuid": row[2]}
                result.append(meta_result)

        resp.text = json.dumps(result)

    @staticmethod
    @user_logger
    def on_post(req, resp, id_):
        """Handles POST requests"""
        admin_control(req)
        try:
            raw_json = req.stream.read().decode('utf-8')
        except Exception as ex:
            print(str(ex))
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.FAILED_TO_READ_REQUEST_STREAM')

        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_ENERGY_STORAGE_POWER_STATION_ID')

        new_values = json.loads(raw_json)

        if 'energy_storage_container_id' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['energy_storage_container_id'], int) or \
                new_values['data']['energy_storage_container_id'] <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_ENERGY_STORAGE_CONTAINER_ID')
        energy_storage_container_id = new_values['data']['energy_storage_container_id']

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

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

        cursor.execute(" SELECT name "
                       " FROM tbl_energy_storage_containers "
                       " WHERE id = %s ", (energy_storage_container_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 = (" SELECT id " 
                 " FROM tbl_energy_storage_power_stations_containers "
                 " WHERE energy_storage_power_station_id = %s AND energy_storage_container_id = %s")
        cursor.execute(query, (id_, energy_storage_container_id,))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
                                   description='API.ENERGY_STORAGE_CONTAINER_SENSOR_RELATION_EXISTS')

        add_row = (" INSERT INTO tbl_energy_storage_power_stations_containers "
                   "        (energy_storage_power_station_id, energy_storage_container_id) "
                   " VALUES (%s, %s) ")
        cursor.execute(add_row, (id_, energy_storage_container_id,))
        cnx.commit()
        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_201
        resp.location = '/energystoragepowerstationss/' + str(id_) + '/containers/' + str(energy_storage_container_id)


class EnergyStoragePowerStationContainerItem:

myems-api/core/microgrid.py 1 location


        resp.status = falcon.HTTP_200


class MicrogridSensorCollection:
    def __init__(self):
        """Initializes Class"""
        pass

    @staticmethod
    def on_options(req, resp, id_):
        _ = req
        resp.status = falcon.HTTP_200
        _ = id_

    @staticmethod
    def on_get(req, resp, id_):
        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')

        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 = (" SELECT s.id, s.name, s.uuid "
                 " FROM tbl_microgrids m, tbl_microgrids_sensors ms, tbl_sensors s "
                 " WHERE ms.microgrid_id = m.id AND s.id = ms.sensor_id AND m.id = %s "
                 " ORDER BY s.id ")
        cursor.execute(query, (id_,))
        rows = cursor.fetchall()

        result = list()
        if rows is not None and len(rows) > 0:
            for row in rows:
                meta_result = {"id": row[0], "name": row[1], "uuid": row[2]}
                result.append(meta_result)

        resp.text = json.dumps(result)

    @staticmethod
    @user_logger
    def on_post(req, resp, id_):
        """Handles POST requests"""
        admin_control(req)
        try:
            raw_json = req.stream.read().decode('utf-8')
        except Exception as ex:
            print(str(ex))
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.FAILED_TO_READ_REQUEST_STREAM')

        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_MICROGRID_ID')

        new_values = json.loads(raw_json)

        if 'sensor_id' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['sensor_id'], int) or \
                new_values['data']['sensor_id'] <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_SENSOR_ID')
        sensor_id = new_values['data']['sensor_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')

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

        query = (" SELECT id " 
                 " FROM tbl_microgrids_sensors "
                 " WHERE microgrid_id = %s AND sensor_id = %s")
        cursor.execute(query, (id_, sensor_id,))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
                                   description='API.MICROGRID_SENSOR_RELATION_EXISTS')

        add_row = (" INSERT INTO tbl_microgrids_sensors (microgrid_id, sensor_id) "
                   " VALUES (%s, %s) ")
        cursor.execute(add_row, (id_, sensor_id,))
        cnx.commit()
        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_201
        resp.location = '/microgrids/' + str(id_) + '/sensors/' + str(sensor_id)


class MicrogridSensorItem:

		@@ 744-858 (lines=115) @@
741		resp.status = falcon.HTTP_200
742
743
744		class EnergyStoragePowerStationContainerCollection:
745		def __init__(self):
746		"""Initializes Class"""
747		pass
748
749		@staticmethod
750		def on_options(req, resp, id_):
751		_ = req
752		resp.status = falcon.HTTP_200
753		_ = id_
754
755		@staticmethod
756		def on_get(req, resp, id_):
757		access_control(req)
758		if not id_.isdigit() or int(id_) <= 0:
759		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
760		description='API.INVALID_ENERGY_STORAGE_POWER_STATION_ID')
761
762		cnx = mysql.connector.connect(**config.myems_system_db)
763		cursor = cnx.cursor()
764
765		cursor.execute(" SELECT name "
766		" FROM tbl_energy_storage_power_stations "
767		" WHERE id = %s ", (id_,))
768		if cursor.fetchone() is None:
769		cursor.close()
770		cnx.close()
771		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
772		description='API.ENERGY_STORAGE_POWER_STATION_NOT_FOUND')
773
774		query = (" SELECT s.id, s.name, s.uuid "
775		" FROM tbl_energy_storage_power_stations e, "
776		" tbl_energy_storage_power_stations_containers es, tbl_energy_storage_containers s "
777		" WHERE es.energy_storage_power_station_id = e.id "
778		" AND s.id = es.energy_storage_container_id "
779		" AND e.id = %s "
780		" ORDER BY s.id ")
781		cursor.execute(query, (id_,))
782		rows = cursor.fetchall()
783
784		result = list()
785		if rows is not None and len(rows) > 0:
786		for row in rows:
787		meta_result = {"id": row[0], "name": row[1], "uuid": row[2]}
788		result.append(meta_result)
789
790		resp.text = json.dumps(result)
791
792		@staticmethod
793		@user_logger
794		def on_post(req, resp, id_):
795		"""Handles POST requests"""
796		admin_control(req)
797		try:
798		raw_json = req.stream.read().decode('utf-8')
799		except Exception as ex:
800		print(str(ex))
801		raise falcon.HTTPError(status=falcon.HTTP_400,
802		title='API.BAD_REQUEST',
803		description='API.FAILED_TO_READ_REQUEST_STREAM')
804
805		if not id_.isdigit() or int(id_) <= 0:
806		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
807		description='API.INVALID_ENERGY_STORAGE_POWER_STATION_ID')
808
809		new_values = json.loads(raw_json)
810
811		if 'energy_storage_container_id' not in new_values['data'].keys() or \
812		not isinstance(new_values['data']['energy_storage_container_id'], int) or \
813		new_values['data']['energy_storage_container_id'] <= 0:
814		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
815		description='API.INVALID_ENERGY_STORAGE_CONTAINER_ID')
816		energy_storage_container_id = new_values['data']['energy_storage_container_id']
817
818		cnx = mysql.connector.connect(**config.myems_system_db)
819		cursor = cnx.cursor()
820
821		cursor.execute(" SELECT name "
822		" FROM tbl_energy_storage_power_stations "
823		" WHERE id = %s ", (id_,))
824		if cursor.fetchone() is None:
825		cursor.close()
826		cnx.close()
827		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
828		description='API.ENERGY_STORAGE_POWER_STATION_NOT_FOUND')
829
830		cursor.execute(" SELECT name "
831		" FROM tbl_energy_storage_containers "
832		" WHERE id = %s ", (energy_storage_container_id,))
833		if cursor.fetchone() is None:
834		cursor.close()
835		cnx.close()
836		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
837		description='API.ENERGY_STORAGE_CONTAINER_NOT_FOUND')
838
839		query = (" SELECT id "
840		" FROM tbl_energy_storage_power_stations_containers "
841		" WHERE energy_storage_power_station_id = %s AND energy_storage_container_id = %s")
842		cursor.execute(query, (id_, energy_storage_container_id,))
843		if cursor.fetchone() is not None:
844		cursor.close()
845		cnx.close()
846		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
847		description='API.ENERGY_STORAGE_CONTAINER_SENSOR_RELATION_EXISTS')
848
849		add_row = (" INSERT INTO tbl_energy_storage_power_stations_containers "
850		" (energy_storage_power_station_id, energy_storage_container_id) "
851		" VALUES (%s, %s) ")
852		cursor.execute(add_row, (id_, energy_storage_container_id,))
853		cnx.commit()
854		cursor.close()
855		cnx.close()
856
857		resp.status = falcon.HTTP_201
858		resp.location = '/energystoragepowerstationss/' + str(id_) + '/containers/' + str(energy_storage_container_id)
859
860
861		class EnergyStoragePowerStationContainerItem:

		@@ 4855-4965 (lines=111) @@
4852		resp.status = falcon.HTTP_200
4853
4854
4855		class MicrogridSensorCollection:
4856		def __init__(self):
4857		"""Initializes Class"""
4858		pass
4859
4860		@staticmethod
4861		def on_options(req, resp, id_):
4862		_ = req
4863		resp.status = falcon.HTTP_200
4864		_ = id_
4865
4866		@staticmethod
4867		def on_get(req, resp, id_):
4868		access_control(req)
4869		if not id_.isdigit() or int(id_) <= 0:
4870		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
4871		description='API.INVALID_MICROGRID_ID')
4872
4873		cnx = mysql.connector.connect(**config.myems_system_db)
4874		cursor = cnx.cursor()
4875
4876		cursor.execute(" SELECT name "
4877		" FROM tbl_microgrids "
4878		" WHERE id = %s ", (id_,))
4879		if cursor.fetchone() is None:
4880		cursor.close()
4881		cnx.close()
4882		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
4883		description='API.MICROGRID_NOT_FOUND')
4884
4885		query = (" SELECT s.id, s.name, s.uuid "
4886		" FROM tbl_microgrids m, tbl_microgrids_sensors ms, tbl_sensors s "
4887		" WHERE ms.microgrid_id = m.id AND s.id = ms.sensor_id AND m.id = %s "
4888		" ORDER BY s.id ")
4889		cursor.execute(query, (id_,))
4890		rows = cursor.fetchall()
4891
4892		result = list()
4893		if rows is not None and len(rows) > 0:
4894		for row in rows:
4895		meta_result = {"id": row[0], "name": row[1], "uuid": row[2]}
4896		result.append(meta_result)
4897
4898		resp.text = json.dumps(result)
4899
4900		@staticmethod
4901		@user_logger
4902		def on_post(req, resp, id_):
4903		"""Handles POST requests"""
4904		admin_control(req)
4905		try:
4906		raw_json = req.stream.read().decode('utf-8')
4907		except Exception as ex:
4908		print(str(ex))
4909		raise falcon.HTTPError(status=falcon.HTTP_400,
4910		title='API.BAD_REQUEST',
4911		description='API.FAILED_TO_READ_REQUEST_STREAM')
4912
4913		if not id_.isdigit() or int(id_) <= 0:
4914		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
4915		description='API.INVALID_MICROGRID_ID')
4916
4917		new_values = json.loads(raw_json)
4918
4919		if 'sensor_id' not in new_values['data'].keys() or \
4920		not isinstance(new_values['data']['sensor_id'], int) or \
4921		new_values['data']['sensor_id'] <= 0:
4922		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
4923		description='API.INVALID_SENSOR_ID')
4924		sensor_id = new_values['data']['sensor_id']
4925
4926		cnx = mysql.connector.connect(**config.myems_system_db)
4927		cursor = cnx.cursor()
4928
4929		cursor.execute(" SELECT name "
4930		" from tbl_microgrids "
4931		" WHERE id = %s ", (id_,))
4932		if cursor.fetchone() is None:
4933		cursor.close()
4934		cnx.close()
4935		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
4936		description='API.MICROGRID_NOT_FOUND')
4937
4938		cursor.execute(" SELECT name "
4939		" FROM tbl_sensors "
4940		" WHERE id = %s ", (sensor_id,))
4941		if cursor.fetchone() is None:
4942		cursor.close()
4943		cnx.close()
4944		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
4945		description='API.SENSOR_NOT_FOUND')
4946
4947		query = (" SELECT id "
4948		" FROM tbl_microgrids_sensors "
4949		" WHERE microgrid_id = %s AND sensor_id = %s")
4950		cursor.execute(query, (id_, sensor_id,))
4951		if cursor.fetchone() is not None:
4952		cursor.close()
4953		cnx.close()
4954		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
4955		description='API.MICROGRID_SENSOR_RELATION_EXISTS')
4956
4957		add_row = (" INSERT INTO tbl_microgrids_sensors (microgrid_id, sensor_id) "
4958		" VALUES (%s, %s) ")
4959		cursor.execute(add_row, (id_, sensor_id,))
4960		cnx.commit()
4961		cursor.close()
4962		cnx.close()
4963
4964		resp.status = falcon.HTTP_201
4965		resp.location = '/microgrids/' + str(id_) + '/sensors/' + str(sensor_id)
4966
4967
4968		class MicrogridSensorItem:

MyEMS / myems

Code Duplication Length = 111-115 lines in 2 locations

myems-api/core/energystoragepowerstation.py 1 location

myems-api/core/microgrid.py 1 location