find_config() - Code Metrics - Inspection of "Fixed a bug introduced int he last commit." - Syncleus/apex - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 06942a...cb25ba )

by Jeffrey

created 2016-09-23 12:05 UTC

find_config() D

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	1

Metric	Value
cc	8
c	1
b	0
f	1
dl	0
loc	25
rs	4

"""
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.pluginloader import getPlugins
from apex.pluginloader import loadPlugin

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__ = []


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.path.join('etc', config_file)
    if config_paths is None:
        config_paths = [cur_path, home_path, etc_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):

    port_map = {}
    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.aprs.AprsKiss(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.aprs.AprsKiss(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.start(kissConstants.MODE_INIT_W8DED)
            elif kiss_init_string == 'MODE_INIT_KENWOOD_D710':
                kiss_tnc.start(kissConstants.MODE_INIT_KENWOOD_D710)
            elif kiss_init_string == 'NONE':
                kiss_tnc.start()
            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
            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': kiss_tnc,
                                       'tnc_port': tnc_port}
    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.AprsInternetService(aprsis_callsign, aprsis_password)
        aprsis.connect(aprsis_server, int(aprsis_server_port))

    def sigint_handler(signal, frame):
        for port in port_map.values():
            port['tnc'].close()
        sys.exit(0)

    signal.signal(signal.SIGINT, sigint_handler)

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

    packet_cache = cachetools.TTLCache(10000, 5)
    # start the plugins
    plugins = []
    try:
        plugin_loaders = getPlugins()
        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 = loadPlugin(plugin_loader)
            plugins.append(loaded_plugin)
            threading.Thread(target=loaded_plugin.start, args=(config, port_map, packet_cache, aprsis)).start()
    except IOError:
        click.echo(click.style('Warning: ', fg='yellow') +
                   click.style('plugin directory not found, will only display incoming messages.'))

    if verbose:
        click.echo('Starting packet processing...')
    while 1:
        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 += click.style(path, fg='magenta')
                    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 in plugins:
                        something_read = True
                        plugin.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.pluginloader import getPlugins
35			from apex.pluginloader import loadPlugin
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
53			def find_config(config_paths, verbose):
54			config_file = 'apex.conf'
55			rc_file = '.apexrc'
56			cur_path = os.path.join(os.curdir, config_file)
57			home_path = os.path.join(os.path.expanduser("~"), rc_file)
58			etc_path = os.path.join('etc', config_file)
59			if config_paths is None:
60			config_paths = [cur_path, home_path, etc_path]
61			elif isinstance(config_paths, str):
62			config_paths = [config_paths]
63			elif not isinstance(config_paths, list):
64			raise TypeError('config_paths argument was neither a string nor a list')
65
66			if verbose:
67			click.echo('Searching for configuration file in the following locations: %s' % repr(config_paths))
68
69			config = configparser.ConfigParser()
70			for config_path in config_paths:
71			try:
72			if len(config.read(config_path)) > 0:
73			return config
74			except IOError:
75			pass
76
77			return None
78
79
80			@click.command(context_settings=dict(auto_envvar_prefix='APEX'))
81			@click.option('-c',
82			'--configfile',
83			type=click.Path(exists=True, file_okay=True, dir_okay=False, readable=True, resolve_path=True),
84			help='Configuration file for APEX.')
85			@click.option('-v', '--verbose', is_flag=True, help='Enables verbose mode.')
86			def main(verbose, configfile):
87
88			port_map = {}
89			config = find_config(configfile, verbose)
90			if config is None:
91			click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
92			click.style('No apex configuration found, can not continue.', bold=True))
93			return
94			for section in config.sections():
95			if section.startswith("TNC "):
96			tnc_name = section.split(" ")[1]
97			if config.has_option(section, 'com_port') and config.has_option(section, 'baud'):
98			com_port = config.get(section, 'com_port')
99			baud = config.get(section, 'baud')
100			kiss_tnc = apex.aprs.AprsKiss(com_port=com_port, baud=baud)
101			elif config.has_option(section, 'tcp_host') and config.has_option(section, 'tcp_port'):
102			tcp_host = config.get(section, 'tcp_host')
103			tcp_port = config.get(section, 'tcp_port')
104			kiss_tnc = apex.aprs.AprsKiss(host=tcp_host, tcp_port=tcp_port)
105			else:
106			click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
107			click.style("""Invalid configuration, must have both com_port and baud set or tcp_host and
108			tcp_port set in TNC sections of configuration file""", bold=True))
109			return
110
111			if not config.has_option(section, 'kiss_init'):
112			click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
113			click.style("""Invalid configuration, must have kiss_init set in TNC sections of
114			configuration file""", bold=True))
115			return
116			kiss_init_string = config.get(section, 'kiss_init')
117			if kiss_init_string == 'MODE_INIT_W8DED':
118			kiss_tnc.start(kissConstants.MODE_INIT_W8DED)
119			elif kiss_init_string == 'MODE_INIT_KENWOOD_D710':
120			kiss_tnc.start(kissConstants.MODE_INIT_KENWOOD_D710)
121			elif kiss_init_string == 'NONE':
122			kiss_tnc.start()
123			else:
124			click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
125			click.style('Invalid configuration, value assigned to kiss_init was not recognized: %s'
126			% kiss_init_string, bold=True))
127			return
128			for port in range(1, 1 + int(config.get(section, 'port_count'))):
129			port_name = tnc_name + '-' + str(port)
130			port_section = 'PORT ' + port_name
131			port_identifier = config.get(port_section, 'identifier')
132			port_net = config.get(port_section, 'net')
133			tnc_port = int(config.get(port_section, 'tnc_port'))
134			port_map[port_name] = {'identifier': port_identifier, 'net': port_net, 'tnc': kiss_tnc,
135			'tnc_port': tnc_port}
136			if config.has_section('APRS-IS'):
137			aprsis_callsign = config.get('APRS-IS', 'callsign')
138			if config.has_option('APRS-IS', 'password'):
139			aprsis_password = config.get('APRS-IS', 'password')
140			else:
141			aprsis_password = -1
142			aprsis_server = config.get('APRS-IS', 'server')
143			aprsis_server_port = config.get('APRS-IS', 'server_port')
144			aprsis = apex.aprs.AprsInternetService(aprsis_callsign, aprsis_password)
145			aprsis.connect(aprsis_server, int(aprsis_server_port))
146
147			def sigint_handler(signal, frame):
148			for port in port_map.values():
149			port['tnc'].close()
150			sys.exit(0)
151
152			signal.signal(signal.SIGINT, sigint_handler)
153
154			click.echo("Press ctrl + c at any time to exit")
155
156			packet_cache = cachetools.TTLCache(10000, 5)
157			# start the plugins
158			plugins = []
159			try:
160			plugin_loaders = getPlugins()
161			if not len(plugin_loaders):
162			click.echo(click.style('Warning: ', fg='yellow') +
163			click.style("No plugins were able to be discovered, will only display incoming messages."))
164			for plugin_loader in plugin_loaders:
165			if verbose:
166			click.echo('Plugin found at the following location: %s' % repr(plugin_loader))
167			loaded_plugin = loadPlugin(plugin_loader)
168			plugins.append(loaded_plugin)
169			threading.Thread(target=loaded_plugin.start, args=(config, port_map, packet_cache, aprsis)).start()
170			except IOError:
171			click.echo(click.style('Warning: ', fg='yellow') +
172			click.style('plugin directory not found, will only display incoming messages.'))
173
174			if verbose:
175			click.echo('Starting packet processing...')
176			while 1:
177			something_read = False
178			try:
179			for port_name in port_map.keys():
180			port = port_map[port_name]
181			frame = port['tnc'].read()
182			if frame:
183			formatted_aprs = '>'.join([click.style(frame['source'], fg='green'), click.style(frame['destination'], fg='blue')])
184			paths = []
185			for path in frame['path']
186			paths += click.style(path, fg='magenta')
187			paths = ','.join(paths)
188			if frame['path']:
189			formatted_aprs = ','.join([formatted_aprs, paths])
190			formatted_aprs += ':'
191			formatted_aprs += frame['text']
192			click.echo(click.style(port_name + ' << ', fg='magenta') + formatted_aprs)
193
194			for plugin in plugins:
195			something_read = True
196			plugin.handle_packet(frame, port, port_name)
197			except Exception as ex:
198			# We want to keep this thread alive so long as the application runs.
199			traceback.print_exc(file=sys.stdout)
200			click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
201			click.style('Caught exception while reading packet: %s' % str(ex), bold=True))
202
203			if something_read is False:
204			time.sleep(1)
205

Syncleus / apex

Push — master ( 06942a...cb25ba )

find_config() D

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like