find_config() - Code Metrics - Inspection of "Changed the config directory list to properly hand..." - Syncleus/apex - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 244a24...befce1 )

by Jeffrey

created 2016-09-27 08:33 UTC

find_config() F

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	1

Metric	Value
cc	9
c	2
b	0
f	1
dl	0
loc	28
rs	3

"""
Module that contains the command line app.

Why does this file exist, and why not put this in __main__?

  You might be tempted to import things from __main__ later, but that will cause
  problems: the code will get executed twice:

  - When you run `python -mapex` python will execute
    ``__main__.py`` as a script. That means there won't be any
    ``apex.__main__`` in ``sys.modules``.
  - When you import __main__ it will get executed again (as a module) because
    there's no ``apex.__main__`` in ``sys.modules``.

  Also see (1) from http://click.pocoo.org/5/setuptools/#setuptools-integration
"""

# These imports are for python3 compatibility inside python2
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import os
import signal
import sys
import threading
import time
import traceback
import cachetools
import click

import apex.aprs
from apex.kiss import constants as kissConstants
from apex.plugin_loader import get_plugins
from apex.plugin_loader import load_plugin

configparser = None
if sys.version_info < (3, 0):
    import ConfigParser  # noqa: F401
    if configparser is None:
        configparser = ConfigParser
elif sys.version_info >= (3, 0):
    import configparser

__author__ = 'Jeffrey Phillips Freeman (WI2ARD)'
__maintainer__ = 'Jeffrey Phillips Freeman (WI2ARD)'
__email__ = '[email protected]'
__license__ = 'Apache License, Version 2.0'
__copyright__ = 'Copyright 2016, Syncleus, Inc. and contributors'
__credits__ = []

port_map = {}
running = True
plugin_modules = []
plugin_threads = []


def find_config(config_paths, verbose):
    config_file = 'apex.conf'
    rc_file = '.apexrc'
    cur_path = os.path.join(os.curdir, config_file)
    home_path = os.path.join(os.path.expanduser('~'), rc_file)
    etc_path = os.pathsep + os.path.join('etc', config_file)
    if config_paths is None:
        if os.name == 'posix':
            config_paths = [cur_path, home_path, etc_path]
        else:
            config_paths = [cur_path, home_path]
    elif isinstance(config_paths, str):
        config_paths = [config_paths]
    elif not isinstance(config_paths, list):
        raise TypeError('config_paths argument was neither a string nor a list')

    if verbose:
        click.echo('Searching for configuration file in the following locations: %s' % repr(config_paths))

    config = configparser.ConfigParser()
    for config_path in config_paths:
        try:
            if len(config.read(config_path)) > 0:
                return config
        except IOError:
            pass

    return None


@click.command(context_settings=dict(auto_envvar_prefix='APEX'))
@click.option('-c',
              '--configfile',
              type=click.Path(exists=True, file_okay=True, dir_okay=False, readable=True, resolve_path=True),
              help='Configuration file for APEX.')
@click.option('-v', '--verbose', is_flag=True, help='Enables verbose mode.')
def main(verbose, configfile):

    config = find_config(configfile, verbose)
    if config is None:
        click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
                   click.style('No apex configuration found, can not continue.', bold=True))
        return
    for section in config.sections():
        if section.startswith("TNC "):
            tnc_name = section.split(" ")[1]
            if config.has_option(section, 'com_port') and config.has_option(section, 'baud'):
                com_port = config.get(section, 'com_port')
                baud = config.get(section, 'baud')
                kiss_tnc = apex.kiss.KissSerial(com_port=com_port, baud=baud)
            elif config.has_option(section, 'tcp_host') and config.has_option(section, 'tcp_port'):
                tcp_host = config.get(section, 'tcp_host')
                tcp_port = config.get(section, 'tcp_port')
                kiss_tnc = apex.kiss.KissTcp(host=tcp_host, tcp_port=tcp_port)
            else:
                click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
                           click.style("""Invalid configuration, must have both com_port and baud set or tcp_host and
                           tcp_port set in TNC sections of configuration file""", bold=True))
                return

            if not config.has_option(section, 'kiss_init'):
                click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
                           click.style("""Invalid configuration, must have kiss_init set in TNC sections of
                           configuration file""", bold=True))
                return
            kiss_init_string = config.get(section, 'kiss_init')
            if kiss_init_string == 'MODE_INIT_W8DED':
                kiss_tnc.connect(kissConstants.MODE_INIT_W8DED)
            elif kiss_init_string == 'MODE_INIT_KENWOOD_D710':
                kiss_tnc.connect(kissConstants.MODE_INIT_KENWOOD_D710)
            elif kiss_init_string == 'NONE':
                kiss_tnc.connect()
            else:
                click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
                           click.style('Invalid configuration, value assigned to kiss_init was not recognized: %s'
                                       % kiss_init_string, bold=True))
                return

            aprs_tnc = apex.aprs.Aprs(data_stream=kiss_tnc)

            for port in range(1, 1 + int(config.get(section, 'port_count'))):
                port_name = tnc_name + '-' + str(port)
                port_section = 'PORT ' + port_name
                port_identifier = config.get(port_section, 'identifier')
                port_net = config.get(port_section, 'net')
                tnc_port = int(config.get(port_section, 'tnc_port'))
                port_map[port_name] = {'identifier': port_identifier, 'net': port_net, 'tnc': aprs_tnc,
                                       'tnc_port': tnc_port}

    aprsis = None
    if config.has_section('APRS-IS'):
        aprsis_callsign = config.get('APRS-IS', 'callsign')
        if config.has_option('APRS-IS', 'password'):
            aprsis_password = config.get('APRS-IS', 'password')
        else:
            aprsis_password = -1
        aprsis_server = config.get('APRS-IS', 'server')
        aprsis_server_port = config.get('APRS-IS', 'server_port')
        aprsis = apex.aprs.ReconnectingPacketBuffer(apex.aprs.IGate(aprsis_callsign, aprsis_password))
        aprsis.connect(aprsis_server, int(aprsis_server_port))

    click.echo("Press ctrl + c at any time to exit")

    packet_cache = cachetools.TTLCache(10000, 5)
    # start the plugins
    try:
        plugin_loaders = get_plugins()
        if not len(plugin_loaders):
            click.echo(click.style('Warning: ', fg='yellow') +
                       click.style('No plugins were able to be discovered, will only display incoming messages.'))
        for plugin_loader in plugin_loaders:
            if verbose:
                click.echo('Plugin found at the following location: %s' % repr(plugin_loader))
            loaded_plugin = load_plugin(plugin_loader)
            plugin_modules.append(loaded_plugin)
            new_thread = threading.Thread(target=loaded_plugin.start, args=(config, port_map, packet_cache, aprsis))
            new_thread.start()
            plugin_threads.append(new_thread)
    except IOError:
        click.echo(click.style('Warning: ', fg='yellow') +
                   click.style('plugin directory not found, will only display incoming messages.'))

    def sigint_handler(signal, frame):
        global running

        running = False

        click.echo()
        click.echo('SIGINT caught, shutting down..')

        for plugin_module in plugin_modules:
            plugin_module.stop()
        if aprsis:
            aprsis.close()
        # Lets wait until all the plugins successfully end
        for plugin_thread in plugin_threads:
            plugin_thread.join()
        for port in port_map.values():
            port['tnc'].data_stream.close()

        click.echo('APEX successfully shutdown.')
        sys.exit(0)

    signal.signal(signal.SIGINT, sigint_handler)

    if verbose:
        click.echo('Starting packet processing...')
    while running:
        something_read = False
        try:
            for port_name in port_map.keys():
                port = port_map[port_name]
                frame = port['tnc'].read()
                if frame:
                    formatted_aprs = '>'.join([click.style(frame['source'], fg='green'), click.style(frame['destination'], fg='blue')])
                    paths = []
                    for path in frame['path']:
                        paths.append(click.style(path, fg='cyan'))
                    paths = ','.join(paths)
                    if frame['path']:
                        formatted_aprs = ','.join([formatted_aprs, paths])
                    formatted_aprs += ':'
                    formatted_aprs += frame['text']
                    click.echo(click.style(port_name + ' << ', fg='magenta') + formatted_aprs)

                    for plugin_module in plugin_modules:
                        something_read = True
                        plugin_module.handle_packet(frame, port, port_name)
        except Exception as ex:
            # We want to keep this thread alive so long as the application runs.
            traceback.print_exc(file=sys.stdout)
            click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
                       click.style('Caught exception while reading packet: %s' % str(ex), bold=True))

        if something_read is False:
            time.sleep(1)


1			"""
2			Module that contains the command line app.
3
4			Why does this file exist, and why not put this in __main__?
5
6			You might be tempted to import things from __main__ later, but that will cause
7			problems: the code will get executed twice:
8
9			- When you run `python -mapex` python will execute
10			``__main__.py`` as a script. That means there won't be any
11			``apex.__main__`` in ``sys.modules``.
12			- When you import __main__ it will get executed again (as a module) because
13			there's no ``apex.__main__`` in ``sys.modules``.
14
15			Also see (1) from http://click.pocoo.org/5/setuptools/#setuptools-integration
16			"""
17
18			# These imports are for python3 compatibility inside python2
19			from __future__ import absolute_import
20			from __future__ import division
21			from __future__ import print_function
22
23			import os
24			import signal
25			import sys
26			import threading
27			import time
28			import traceback
29			import cachetools
30			import click
31
32			import apex.aprs
33			from apex.kiss import constants as kissConstants
34			from apex.plugin_loader import get_plugins
35			from apex.plugin_loader import load_plugin
36
37			configparser = None
38			if sys.version_info < (3, 0):
39			import ConfigParser # noqa: F401
40			if configparser is None:
41			configparser = ConfigParser
42			elif sys.version_info >= (3, 0):
43			import configparser
44
45			__author__ = 'Jeffrey Phillips Freeman (WI2ARD)'
46			__maintainer__ = 'Jeffrey Phillips Freeman (WI2ARD)'
47			__email__ = '[email protected]'
48			__license__ = 'Apache License, Version 2.0'
49			__copyright__ = 'Copyright 2016, Syncleus, Inc. and contributors'
50			__credits__ = []
51
52			port_map = {}
53			running = True
54			plugin_modules = []
55			plugin_threads = []
56
57
58			def find_config(config_paths, verbose):
59			config_file = 'apex.conf'
60			rc_file = '.apexrc'
61			cur_path = os.path.join(os.curdir, config_file)
62			home_path = os.path.join(os.path.expanduser('~'), rc_file)
63			etc_path = os.pathsep + os.path.join('etc', config_file)
64			if config_paths is None:
65			if os.name == 'posix':
66			config_paths = [cur_path, home_path, etc_path]
67			else:
68			config_paths = [cur_path, home_path]
69			elif isinstance(config_paths, str):
70			config_paths = [config_paths]
71			elif not isinstance(config_paths, list):
72			raise TypeError('config_paths argument was neither a string nor a list')
73
74			if verbose:
75			click.echo('Searching for configuration file in the following locations: %s' % repr(config_paths))
76
77			config = configparser.ConfigParser()
78			for config_path in config_paths:
79			try:
80			if len(config.read(config_path)) > 0:
81			return config
82			except IOError:
83			pass
84
85			return None
86
87
88			@click.command(context_settings=dict(auto_envvar_prefix='APEX'))
89			@click.option('-c',
90			'--configfile',
91			type=click.Path(exists=True, file_okay=True, dir_okay=False, readable=True, resolve_path=True),
92			help='Configuration file for APEX.')
93			@click.option('-v', '--verbose', is_flag=True, help='Enables verbose mode.')
94			def main(verbose, configfile):
95
96			config = find_config(configfile, verbose)
97			if config is None:
98			click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
99			click.style('No apex configuration found, can not continue.', bold=True))
100			return
101			for section in config.sections():
102			if section.startswith("TNC "):
103			tnc_name = section.split(" ")[1]
104			if config.has_option(section, 'com_port') and config.has_option(section, 'baud'):
105			com_port = config.get(section, 'com_port')
106			baud = config.get(section, 'baud')
107			kiss_tnc = apex.kiss.KissSerial(com_port=com_port, baud=baud)
108			elif config.has_option(section, 'tcp_host') and config.has_option(section, 'tcp_port'):
109			tcp_host = config.get(section, 'tcp_host')
110			tcp_port = config.get(section, 'tcp_port')
111			kiss_tnc = apex.kiss.KissTcp(host=tcp_host, tcp_port=tcp_port)
112			else:
113			click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
114			click.style("""Invalid configuration, must have both com_port and baud set or tcp_host and
115			tcp_port set in TNC sections of configuration file""", bold=True))
116			return
117
118			if not config.has_option(section, 'kiss_init'):
119			click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
120			click.style("""Invalid configuration, must have kiss_init set in TNC sections of
121			configuration file""", bold=True))
122			return
123			kiss_init_string = config.get(section, 'kiss_init')
124			if kiss_init_string == 'MODE_INIT_W8DED':
125			kiss_tnc.connect(kissConstants.MODE_INIT_W8DED)
126			elif kiss_init_string == 'MODE_INIT_KENWOOD_D710':
127			kiss_tnc.connect(kissConstants.MODE_INIT_KENWOOD_D710)
128			elif kiss_init_string == 'NONE':
129			kiss_tnc.connect()
130			else:
131			click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
132			click.style('Invalid configuration, value assigned to kiss_init was not recognized: %s'
133			% kiss_init_string, bold=True))
134			return
135
136			aprs_tnc = apex.aprs.Aprs(data_stream=kiss_tnc)
137
138			for port in range(1, 1 + int(config.get(section, 'port_count'))):
139			port_name = tnc_name + '-' + str(port)
140			port_section = 'PORT ' + port_name
141			port_identifier = config.get(port_section, 'identifier')
142			port_net = config.get(port_section, 'net')
143			tnc_port = int(config.get(port_section, 'tnc_port'))
144			port_map[port_name] = {'identifier': port_identifier, 'net': port_net, 'tnc': aprs_tnc,
145			'tnc_port': tnc_port}
146
147			aprsis = None
148			if config.has_section('APRS-IS'):
149			aprsis_callsign = config.get('APRS-IS', 'callsign')
150			if config.has_option('APRS-IS', 'password'):
151			aprsis_password = config.get('APRS-IS', 'password')
152			else:
153			aprsis_password = -1
154			aprsis_server = config.get('APRS-IS', 'server')
155			aprsis_server_port = config.get('APRS-IS', 'server_port')
156			aprsis = apex.aprs.ReconnectingPacketBuffer(apex.aprs.IGate(aprsis_callsign, aprsis_password))
157			aprsis.connect(aprsis_server, int(aprsis_server_port))
158
159			click.echo("Press ctrl + c at any time to exit")
160
161			packet_cache = cachetools.TTLCache(10000, 5)
162			# start the plugins
163			try:
164			plugin_loaders = get_plugins()
165			if not len(plugin_loaders):
166			click.echo(click.style('Warning: ', fg='yellow') +
167			click.style('No plugins were able to be discovered, will only display incoming messages.'))
168			for plugin_loader in plugin_loaders:
169			if verbose:
170			click.echo('Plugin found at the following location: %s' % repr(plugin_loader))
171			loaded_plugin = load_plugin(plugin_loader)
172			plugin_modules.append(loaded_plugin)
173			new_thread = threading.Thread(target=loaded_plugin.start, args=(config, port_map, packet_cache, aprsis))
174			new_thread.start()
175			plugin_threads.append(new_thread)
176			except IOError:
177			click.echo(click.style('Warning: ', fg='yellow') +
178			click.style('plugin directory not found, will only display incoming messages.'))
179
180			def sigint_handler(signal, frame):
181			global running
182
183			running = False
184
185			click.echo()
186			click.echo('SIGINT caught, shutting down..')
187
188			for plugin_module in plugin_modules:
189			plugin_module.stop()
190			if aprsis:
191			aprsis.close()
192			# Lets wait until all the plugins successfully end
193			for plugin_thread in plugin_threads:
194			plugin_thread.join()
195			for port in port_map.values():
196			port['tnc'].data_stream.close()
197
198			click.echo('APEX successfully shutdown.')
199			sys.exit(0)
200
201			signal.signal(signal.SIGINT, sigint_handler)
202
203			if verbose:
204			click.echo('Starting packet processing...')
205			while running:
206			something_read = False
207			try:
208			for port_name in port_map.keys():
209			port = port_map[port_name]
210			frame = port['tnc'].read()
211			if frame:
212			formatted_aprs = '>'.join([click.style(frame['source'], fg='green'), click.style(frame['destination'], fg='blue')])
213			paths = []
214			for path in frame['path']:
215			paths.append(click.style(path, fg='cyan'))
216			paths = ','.join(paths)
217			if frame['path']:
218			formatted_aprs = ','.join([formatted_aprs, paths])
219			formatted_aprs += ':'
220			formatted_aprs += frame['text']
221			click.echo(click.style(port_name + ' << ', fg='magenta') + formatted_aprs)
222
223			for plugin_module in plugin_modules:
224			something_read = True
225			plugin_module.handle_packet(frame, port, port_name)
226			except Exception as ex:
227			# We want to keep this thread alive so long as the application runs.
228			traceback.print_exc(file=sys.stdout)
229			click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
230			click.style('Caught exception while reading packet: %s' % str(ex), bold=True))
231
232			if something_read is False:
233			time.sleep(1)
234

Syncleus / apex

Push — master ( 244a24...befce1 )

find_config() F

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like