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

Code Duplication Length = 41-44 lines in 3 locations

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




def stop():
    plugin.stop()


class StatusPlugin(object):

    def __init__(self, config, port_map, aprsis):
        self.port_map = port_map
        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/id/init.py 1 location




def stop():
    plugin.stop()


class IdPlugin(object):

    def __init__(self, config, port_map, aprsis):
        self.port_map = port_map
        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)

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




def stop():
    plugin.stop()


class BeaconPlugin(object):

    def __init__(self, config, port_map, aprsis):
        self.port_map = port_map
        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)

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

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

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

Syncleus / apex

Code Duplication Length = 41-44 lines in 3 locations

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

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

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

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

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

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