Code Duplication - Syncleus/apex - Measure and Improve Code Quality continuously with Scrutinizer

Code Duplication Length = 42-45 lines in 3 locations

src/apex/plugins/status/init.py 1 location


    plugin.stop()


class StatusPlugin(object):

    def __init__(self, config, port_map, packet_cache, aprsis):
        self.port_map = port_map
        self.packet_cache = packet_cache
        self.aprsis = aprsis
        self.running = False

        for section in config.sections():
            if section.startswith('TNC '):
                tnc_name = section.split(' ')[1]
                for port_id in range(1, 1+int(config.get(section, 'port_count'))):
                    port_name = tnc_name + '-' + str(port_id)
                    port = port_map[port_name]
                    port_section = 'PORT ' + port_name
                    port['status_text'] = config.get(port_section, 'status_text')
                    port['status_path'] = config.get(port_section, 'status_path')

    def stop(self):
        self.running = False

    def run(self):
        self.running = True

        # Don't do anything in the first 60 seconds
        last_trigger = time.time()
        while self.running and time.time() - last_trigger < 60:
            time.sleep(1)

        # run the id every 600 seconds
        last_trigger = time.time()
        while self.running:
            if time.time() - last_trigger >= 600:
                last_trigger = time.time()
                for port_name in self.port_map.keys():
                    port = self.port_map[port_name]

                    status_frame = {
                        'source': port['identifier'],
                        'destination': 'APRS',
                        'path': port['status_path'].split(','),
                        'text': port['status_text']}
                    port['tnc'].write(status_frame, port['tnc_port'])
            else:
                time.sleep(1)


src/apex/plugins/beacon/init.py 1 location


    plugin.stop()


class BeaconPlugin(object):

    def __init__(self, config, port_map, packet_cache, aprsis):
        self.port_map = port_map
        self.packet_cache = packet_cache
        self.aprsis = aprsis
        self.running = False

        for section in config.sections():
            if section.startswith('TNC '):
                tnc_name = section.split(' ')[1]
                for port_id in range(1, 1+int(config.get(section, 'port_count'))):
                    port_name = tnc_name + '-' + str(port_id)
                    port = port_map[port_name]
                    port_section = 'PORT ' + port_name
                    port['beacon_text'] = config.get(port_section, 'beacon_text')
                    port['beacon_path'] = config.get(port_section, 'beacon_path')

    def stop(self):
        self.running = False

    def run(self):
        self.running = True

        # Don't do anything in the first 30 seconds
        last_trigger = time.time()
        while self.running and time.time() - last_trigger < 30:
            time.sleep(1)

        # run every 600 second
        last_trigger = time.time()
        while self.running:
            if time.time() - last_trigger >= 600:
                last_trigger = time.time()
                for port_name in self.port_map.keys():
                    port = self.port_map[port_name]

                    beacon_frame = {'source': port['identifier'], 'destination': 'APRS',
                                    'path': port['beacon_path'].split(','), 'text': port['beacon_text']}
                    port['tnc'].write(beacon_frame, port['tnc_port'])
            else:
                time.sleep(1)


src/apex/plugins/id/init.py 1 location


    plugin.stop()


class IdPlugin(object):

    def __init__(self, config, port_map, packet_cache, aprsis):
        self.port_map = port_map
        self.packet_cache = packet_cache
        self.aprsis = aprsis
        self.running = False

        for section in config.sections():
            if section.startswith('TNC '):
                tnc_name = section.split(' ')[1]
                for port_id in range(1, 1+int(config.get(section, 'port_count'))):
                    port_name = tnc_name + '-' + str(port_id)
                    port = port_map[port_name]
                    port_section = 'PORT ' + port_name
                    port['id_text'] = config.get(port_section, 'id_text')
                    port['id_path'] = config.get(port_section, 'id_path')

    def stop(self):
        self.running = False

    def run(self):
        self.running = True

        # Don't do anything in the first 30 seconds
        last_trigger = time.time()
        while self.running and time.time() - last_trigger < 30:
            time.sleep(1)

        # run every 600 second
        last_trigger = time.time()
        while self.running:
            if time.time() - last_trigger >= 600:
                last_trigger = time.time()
                for port_name in self.port_map.keys():
                    port = self.port_map[port_name]

                    id_frame = {'source': port['identifier'], 'destination': 'ID', 'path': port['id_path'].split(','),
                                'text': port['id_text']}
                    port['tnc'].write(id_frame, port['tnc_port'])
            else:
                time.sleep(1)


		@@ 33-77 (lines=45) @@
30		plugin.stop()
31
32
33		class StatusPlugin(object):
34
35		def __init__(self, config, port_map, packet_cache, aprsis):
36		self.port_map = port_map
37		self.packet_cache = packet_cache
38		self.aprsis = aprsis
39		self.running = False
40
41		for section in config.sections():
42		if section.startswith('TNC '):
43		tnc_name = section.split(' ')[1]
44		for port_id in range(1, 1+int(config.get(section, 'port_count'))):
45		port_name = tnc_name + '-' + str(port_id)
46		port = port_map[port_name]
47		port_section = 'PORT ' + port_name
48		port['status_text'] = config.get(port_section, 'status_text')
49		port['status_path'] = config.get(port_section, 'status_path')
50
51		def stop(self):
52		self.running = False
53
54		def run(self):
55		self.running = True
56
57		# Don't do anything in the first 60 seconds
58		last_trigger = time.time()
59		while self.running and time.time() - last_trigger < 60:
60		time.sleep(1)
61
62		# run the id every 600 seconds
63		last_trigger = time.time()
64		while self.running:
65		if time.time() - last_trigger >= 600:
66		last_trigger = time.time()
67		for port_name in self.port_map.keys():
68		port = self.port_map[port_name]
69
70		status_frame = {
71		'source': port['identifier'],
72		'destination': 'APRS',
73		'path': port['status_path'].split(','),
74		'text': port['status_text']}
75		port['tnc'].write(status_frame, port['tnc_port'])
76		else:
77		time.sleep(1)
78

		@@ 33-74 (lines=42) @@
30		plugin.stop()
31
32
33		class BeaconPlugin(object):
34
35		def __init__(self, config, port_map, packet_cache, aprsis):
36		self.port_map = port_map
37		self.packet_cache = packet_cache
38		self.aprsis = aprsis
39		self.running = False
40
41		for section in config.sections():
42		if section.startswith('TNC '):
43		tnc_name = section.split(' ')[1]
44		for port_id in range(1, 1+int(config.get(section, 'port_count'))):
45		port_name = tnc_name + '-' + str(port_id)
46		port = port_map[port_name]
47		port_section = 'PORT ' + port_name
48		port['beacon_text'] = config.get(port_section, 'beacon_text')
49		port['beacon_path'] = config.get(port_section, 'beacon_path')
50
51		def stop(self):
52		self.running = False
53
54		def run(self):
55		self.running = True
56
57		# Don't do anything in the first 30 seconds
58		last_trigger = time.time()
59		while self.running and time.time() - last_trigger < 30:
60		time.sleep(1)
61
62		# run every 600 second
63		last_trigger = time.time()
64		while self.running:
65		if time.time() - last_trigger >= 600:
66		last_trigger = time.time()
67		for port_name in self.port_map.keys():
68		port = self.port_map[port_name]
69
70		beacon_frame = {'source': port['identifier'], 'destination': 'APRS',
71		'path': port['beacon_path'].split(','), 'text': port['beacon_text']}
72		port['tnc'].write(beacon_frame, port['tnc_port'])
73		else:
74		time.sleep(1)
75

		@@ 33-74 (lines=42) @@
30		plugin.stop()
31
32
33		class IdPlugin(object):
34
35		def __init__(self, config, port_map, packet_cache, aprsis):
36		self.port_map = port_map
37		self.packet_cache = packet_cache
38		self.aprsis = aprsis
39		self.running = False
40
41		for section in config.sections():
42		if section.startswith('TNC '):
43		tnc_name = section.split(' ')[1]
44		for port_id in range(1, 1+int(config.get(section, 'port_count'))):
45		port_name = tnc_name + '-' + str(port_id)
46		port = port_map[port_name]
47		port_section = 'PORT ' + port_name
48		port['id_text'] = config.get(port_section, 'id_text')
49		port['id_path'] = config.get(port_section, 'id_path')
50
51		def stop(self):
52		self.running = False
53
54		def run(self):
55		self.running = True
56
57		# Don't do anything in the first 30 seconds
58		last_trigger = time.time()
59		while self.running and time.time() - last_trigger < 30:
60		time.sleep(1)
61
62		# run every 600 second
63		last_trigger = time.time()
64		while self.running:
65		if time.time() - last_trigger >= 600:
66		last_trigger = time.time()
67		for port_name in self.port_map.keys():
68		port = self.port_map[port_name]
69
70		id_frame = {'source': port['identifier'], 'destination': 'ID', 'path': port['id_path'].split(','),
71		'text': port['id_text']}
72		port['tnc'].write(id_frame, port['tnc_port'])
73		else:
74		time.sleep(1)
75

Syncleus / apex

Code Duplication Length = 42-45 lines in 3 locations

src/apex/plugins/status/__init__.py 1 location

src/apex/plugins/beacon/__init__.py 1 location

src/apex/plugins/id/__init__.py 1 location

src/apex/plugins/status/init.py 1 location

src/apex/plugins/beacon/init.py 1 location

src/apex/plugins/id/init.py 1 location