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

Code Duplication Length = 107-107 lines in 2 locations

myems-api/core/microgrid.py 1 location


        resp.status = falcon.HTTP_200


class MicrogridScheduleCollection:
    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 id, start_time_of_day, end_time_of_day, peak_type, power "
                 " FROM tbl_microgrids_schedules "
                 " WHERE microgrid_id = %s "
                 " ORDER BY start_time_of_day ")
        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],
                               "start_time_of_day": str(row[1]),
                               "end_time_of_day": str(row[2]),
                               "peak_type": row[3],
                               "power": row[4],
                               }
                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')

        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')

        new_values = json.loads(raw_json)

        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')

        add_schedule = (" INSERT INTO tbl_microgrids_schedules "
                        "     (microgrid_id, start_time_of_day, end_time_of_day, peak_type, power) "
                        " VALUES (%s, %s, %s, %s, %s) ")
        cursor.execute(add_schedule, (id_,
                                      new_values['data']['start_time_of_day'],
                                      new_values['data']['end_time_of_day'],
                                      new_values['data']['peak_type'],
                                      new_values['data']['power']))
        new_id = cursor.lastrowid
        cnx.commit()
        cursor.close()
        cnx.close()
        resp.status = falcon.HTTP_201
        resp.location = '/energystoragecontainerschedules/' + str(new_id)


class MicrogridScheduleItem:

myems-api/core/energystoragecontainer.py 1 location


        resp.status = falcon.HTTP_204


class EnergyStorageContainerScheduleCollection:
    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_CONTAINER_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 = (" SELECT id, start_time_of_day, end_time_of_day, peak_type, power "
                 " FROM tbl_energy_storage_containers_schedules "
                 " WHERE energy_storage_container_id = %s "
                 " ORDER BY start_time_of_day ")
        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],
                               "start_time_of_day": str(row[1]),
                               "end_time_of_day": str(row[2]),
                               "peak_type": row[3],
                               "power": row[4],
                               }
                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_CONTAINER_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')

        new_values = json.loads(raw_json)

        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')

        add_schedule = (" INSERT INTO tbl_energy_storage_containers_schedules "
                        "     (energy_storage_container_id, start_time_of_day, end_time_of_day, peak_type, power) "
                        " VALUES (%s, %s, %s, %s, %s) ")
        cursor.execute(add_schedule, (id_,
                                      new_values['data']['start_time_of_day'],
                                      new_values['data']['end_time_of_day'],
                                      new_values['data']['peak_type'],
                                      new_values['data']['power']))
        new_id = cursor.lastrowid
        cnx.commit()
        cursor.close()
        cnx.close()
        resp.status = falcon.HTTP_201
        resp.location = '/energystoragecontainerschedules/' + str(new_id)


class EnergyStorageContainerScheduleItem:

		@@ 4554-4660 (lines=107) @@
4551		resp.status = falcon.HTTP_200
4552
4553
4554		class MicrogridScheduleCollection:
4555		def __init__(self):
4556		"""Initializes Class"""
4557		pass
4558
4559		@staticmethod
4560		def on_options(req, resp, id_):
4561		_ = req
4562		resp.status = falcon.HTTP_200
4563		_ = id_
4564
4565		@staticmethod
4566		def on_get(req, resp, id_):
4567		access_control(req)
4568		if not id_.isdigit() or int(id_) <= 0:
4569		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
4570		description='API.INVALID_MICROGRID_ID')
4571
4572		cnx = mysql.connector.connect(**config.myems_system_db)
4573		cursor = cnx.cursor()
4574
4575		cursor.execute(" SELECT name "
4576		" FROM tbl_microgrids "
4577		" WHERE id = %s ", (id_,))
4578		if cursor.fetchone() is None:
4579		cursor.close()
4580		cnx.close()
4581		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
4582		description='API.MICROGRID_NOT_FOUND')
4583
4584		query = (" SELECT id, start_time_of_day, end_time_of_day, peak_type, power "
4585		" FROM tbl_microgrids_schedules "
4586		" WHERE microgrid_id = %s "
4587		" ORDER BY start_time_of_day ")
4588		cursor.execute(query, (id_,))
4589		rows = cursor.fetchall()
4590
4591		result = list()
4592		if rows is not None and len(rows) > 0:
4593		for row in rows:
4594		meta_result = {"id": row[0],
4595		"start_time_of_day": str(row[1]),
4596		"end_time_of_day": str(row[2]),
4597		"peak_type": row[3],
4598		"power": row[4],
4599		}
4600		result.append(meta_result)
4601
4602		resp.text = json.dumps(result)
4603
4604		@staticmethod
4605		@user_logger
4606		def on_post(req, resp, id_):
4607		"""Handles POST requests"""
4608		admin_control(req)
4609		try:
4610		raw_json = req.stream.read().decode('utf-8')
4611		except Exception as ex:
4612		print(str(ex))
4613		raise falcon.HTTPError(status=falcon.HTTP_400,
4614		title='API.BAD_REQUEST',
4615		description='API.FAILED_TO_READ_REQUEST_STREAM')
4616		if not id_.isdigit() or int(id_) <= 0:
4617		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
4618		description='API.INVALID_MICROGRID_ID')
4619
4620		cnx = mysql.connector.connect(**config.myems_system_db)
4621		cursor = cnx.cursor()
4622
4623		cursor.execute(" SELECT name "
4624		" FROM tbl_microgrids "
4625		" WHERE id = %s ", (id_,))
4626		if cursor.fetchone() is None:
4627		cursor.close()
4628		cnx.close()
4629		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
4630		description='API.MICROGRID_NOT_FOUND')
4631
4632		new_values = json.loads(raw_json)
4633
4634		cnx = mysql.connector.connect(**config.myems_system_db)
4635		cursor = cnx.cursor()
4636
4637		cursor.execute(" SELECT name "
4638		" FROM tbl_microgrids "
4639		" WHERE id = %s ",
4640		(id_,))
4641		if cursor.fetchone() is None:
4642		cursor.close()
4643		cnx.close()
4644		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
4645		description='API.MICROGRID_NOT_FOUND')
4646
4647		add_schedule = (" INSERT INTO tbl_microgrids_schedules "
4648		" (microgrid_id, start_time_of_day, end_time_of_day, peak_type, power) "
4649		" VALUES (%s, %s, %s, %s, %s) ")
4650		cursor.execute(add_schedule, (id_,
4651		new_values['data']['start_time_of_day'],
4652		new_values['data']['end_time_of_day'],
4653		new_values['data']['peak_type'],
4654		new_values['data']['power']))
4655		new_id = cursor.lastrowid
4656		cnx.commit()
4657		cursor.close()
4658		cnx.close()
4659		resp.status = falcon.HTTP_201
4660		resp.location = '/energystoragecontainerschedules/' + str(new_id)
4661
4662
4663		class MicrogridScheduleItem:

		@@ 5050-5156 (lines=107) @@
5047		resp.status = falcon.HTTP_204
5048
5049
5050		class EnergyStorageContainerScheduleCollection:
5051		def __init__(self):
5052		"""Initializes Class"""
5053		pass
5054
5055		@staticmethod
5056		def on_options(req, resp, id_):
5057		_ = req
5058		resp.status = falcon.HTTP_200
5059		_ = id_
5060
5061		@staticmethod
5062		def on_get(req, resp, id_):
5063		access_control(req)
5064		if not id_.isdigit() or int(id_) <= 0:
5065		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
5066		description='API.INVALID_ENERGY_STORAGE_CONTAINER_ID')
5067
5068		cnx = mysql.connector.connect(**config.myems_system_db)
5069		cursor = cnx.cursor()
5070
5071		cursor.execute(" SELECT name "
5072		" FROM tbl_energy_storage_containers "
5073		" WHERE id = %s ", (id_,))
5074		if cursor.fetchone() is None:
5075		cursor.close()
5076		cnx.close()
5077		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
5078		description='API.ENERGY_STORAGE_CONTAINER_NOT_FOUND')
5079
5080		query = (" SELECT id, start_time_of_day, end_time_of_day, peak_type, power "
5081		" FROM tbl_energy_storage_containers_schedules "
5082		" WHERE energy_storage_container_id = %s "
5083		" ORDER BY start_time_of_day ")
5084		cursor.execute(query, (id_,))
5085		rows = cursor.fetchall()
5086
5087		result = list()
5088		if rows is not None and len(rows) > 0:
5089		for row in rows:
5090		meta_result = {"id": row[0],
5091		"start_time_of_day": str(row[1]),
5092		"end_time_of_day": str(row[2]),
5093		"peak_type": row[3],
5094		"power": row[4],
5095		}
5096		result.append(meta_result)
5097
5098		resp.text = json.dumps(result)
5099
5100		@staticmethod
5101		@user_logger
5102		def on_post(req, resp, id_):
5103		"""Handles POST requests"""
5104		admin_control(req)
5105		try:
5106		raw_json = req.stream.read().decode('utf-8')
5107		except Exception as ex:
5108		print(str(ex))
5109		raise falcon.HTTPError(status=falcon.HTTP_400,
5110		title='API.BAD_REQUEST',
5111		description='API.FAILED_TO_READ_REQUEST_STREAM')
5112		if not id_.isdigit() or int(id_) <= 0:
5113		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
5114		description='API.INVALID_ENERGY_STORAGE_CONTAINER_ID')
5115
5116		cnx = mysql.connector.connect(**config.myems_system_db)
5117		cursor = cnx.cursor()
5118
5119		cursor.execute(" SELECT name "
5120		" FROM tbl_energy_storage_containers "
5121		" WHERE id = %s ", (id_,))
5122		if cursor.fetchone() is None:
5123		cursor.close()
5124		cnx.close()
5125		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
5126		description='API.ENERGY_STORAGE_CONTAINER_NOT_FOUND')
5127
5128		new_values = json.loads(raw_json)
5129
5130		cnx = mysql.connector.connect(**config.myems_system_db)
5131		cursor = cnx.cursor()
5132
5133		cursor.execute(" SELECT name "
5134		" FROM tbl_energy_storage_containers "
5135		" WHERE id = %s ",
5136		(id_,))
5137		if cursor.fetchone() is None:
5138		cursor.close()
5139		cnx.close()
5140		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
5141		description='API.ENERGY_STORAGE_CONTAINER_NOT_FOUND')
5142
5143		add_schedule = (" INSERT INTO tbl_energy_storage_containers_schedules "
5144		" (energy_storage_container_id, start_time_of_day, end_time_of_day, peak_type, power) "
5145		" VALUES (%s, %s, %s, %s, %s) ")
5146		cursor.execute(add_schedule, (id_,
5147		new_values['data']['start_time_of_day'],
5148		new_values['data']['end_time_of_day'],
5149		new_values['data']['peak_type'],
5150		new_values['data']['power']))
5151		new_id = cursor.lastrowid
5152		cnx.commit()
5153		cursor.close()
5154		cnx.close()
5155		resp.status = falcon.HTTP_201
5156		resp.location = '/energystoragecontainerschedules/' + str(new_id)
5157
5158
5159		class EnergyStorageContainerScheduleItem:

MyEMS / myems

Code Duplication Length = 107-107 lines in 2 locations

myems-api/core/microgrid.py 1 location

myems-api/core/energystoragecontainer.py 1 location