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

Code Duplication Length = 107-107 lines in 2 locations

myems-api/core/energystoragecontainer.py 1 location


        resp.status = falcon.HTTP_204


class EnergyStorageContainerScheduleCollection:
    def __init__(self):
        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:
    def __init__(self):

myems-api/core/microgrid.py 1 location


        resp.status = falcon.HTTP_200


class MicrogridScheduleCollection:
    def __init__(self):
        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:
    def __init__(self):

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

		@@ 4583-4689 (lines=107) @@
4580		resp.status = falcon.HTTP_200
4581
4582
4583		class MicrogridScheduleCollection:
4584		def __init__(self):
4585		pass
4586
4587		@staticmethod
4588		def on_options(req, resp, id_):
4589		_ = req
4590		resp.status = falcon.HTTP_200
4591		_ = id_
4592
4593		@staticmethod
4594		def on_get(req, resp, id_):
4595		access_control(req)
4596		if not id_.isdigit() or int(id_) <= 0:
4597		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
4598		description='API.INVALID_MICROGRID_ID')
4599
4600		cnx = mysql.connector.connect(**config.myems_system_db)
4601		cursor = cnx.cursor()
4602
4603		cursor.execute(" SELECT name "
4604		" FROM tbl_microgrids "
4605		" WHERE id = %s ", (id_,))
4606		if cursor.fetchone() is None:
4607		cursor.close()
4608		cnx.close()
4609		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
4610		description='API.MICROGRID_NOT_FOUND')
4611
4612		query = (" SELECT id, start_time_of_day, end_time_of_day, peak_type, power "
4613		" FROM tbl_microgrids_schedules "
4614		" WHERE microgrid_id = %s "
4615		" ORDER BY start_time_of_day ")
4616		cursor.execute(query, (id_,))
4617		rows = cursor.fetchall()
4618
4619		result = list()
4620		if rows is not None and len(rows) > 0:
4621		for row in rows:
4622		meta_result = {"id": row[0],
4623		"start_time_of_day": str(row[1]),
4624		"end_time_of_day": str(row[2]),
4625		"peak_type": row[3],
4626		"power": row[4],
4627		}
4628		result.append(meta_result)
4629
4630		resp.text = json.dumps(result)
4631
4632		@staticmethod
4633		@user_logger
4634		def on_post(req, resp, id_):
4635		"""Handles POST requests"""
4636		admin_control(req)
4637		try:
4638		raw_json = req.stream.read().decode('utf-8')
4639		except Exception as ex:
4640		print(str(ex))
4641		raise falcon.HTTPError(status=falcon.HTTP_400,
4642		title='API.BAD_REQUEST',
4643		description='API.FAILED_TO_READ_REQUEST_STREAM')
4644		if not id_.isdigit() or int(id_) <= 0:
4645		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
4646		description='API.INVALID_MICROGRID_ID')
4647
4648		cnx = mysql.connector.connect(**config.myems_system_db)
4649		cursor = cnx.cursor()
4650
4651		cursor.execute(" SELECT name "
4652		" FROM tbl_microgrids "
4653		" WHERE id = %s ", (id_,))
4654		if cursor.fetchone() is None:
4655		cursor.close()
4656		cnx.close()
4657		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
4658		description='API.MICROGRID_NOT_FOUND')
4659
4660		new_values = json.loads(raw_json)
4661
4662		cnx = mysql.connector.connect(**config.myems_system_db)
4663		cursor = cnx.cursor()
4664
4665		cursor.execute(" SELECT name "
4666		" FROM tbl_microgrids "
4667		" WHERE id = %s ",
4668		(id_,))
4669		if cursor.fetchone() is None:
4670		cursor.close()
4671		cnx.close()
4672		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
4673		description='API.MICROGRID_NOT_FOUND')
4674
4675		add_schedule = (" INSERT INTO tbl_microgrids_schedules "
4676		" (microgrid_id, start_time_of_day, end_time_of_day, peak_type, power) "
4677		" VALUES (%s, %s, %s, %s, %s) ")
4678		cursor.execute(add_schedule, (id_,
4679		new_values['data']['start_time_of_day'],
4680		new_values['data']['end_time_of_day'],
4681		new_values['data']['peak_type'],
4682		new_values['data']['power']))
4683		new_id = cursor.lastrowid
4684		cnx.commit()
4685		cursor.close()
4686		cnx.close()
4687		resp.status = falcon.HTTP_201
4688		resp.location = '/energystoragecontainerschedules/' + str(new_id)
4689
4690
4691		class MicrogridScheduleItem:
4692		def __init__(self):

MyEMS / myems

Code Duplication Length = 107-107 lines in 2 locations

myems-api/core/energystoragecontainer.py 1 location

myems-api/core/microgrid.py 1 location