find_config() - Code Metrics - Syncleus/apex - Measure and Improve Code Quality continuously with Scrutinizer

find_config() F
last analyzed 2016-12-13 21:00 UTC

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	3
Bugs	0	Features	1

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

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
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 click
import six

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

from .buffers import NonrepeatingBuffer
from .util import echo_colorized_error
from .util import echo_colorized_warning

configparser = None
if six.PY2:
    import ConfigParser  # noqa: F401
    if configparser is None:
        configparser = ConfigParser
elif six.PY3:
    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__ = []

config = None
aprsis = None
port_map = {}
running = True
plugin_modules = []
plugin_threads = []


def sigint_handler(signal=None, frame=None):
    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'].close()

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


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.sep + 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


def configure(configfile, verbose=False):
    global config
    config = find_config(configfile, verbose)
    if config is None:
        echo_colorized_error('No apex configuration found, can not continue.')
        return False
    for section in config.sections():
        if section.startswith("TNC "):
            tnc_name = section.strip().split(" ")[1].strip()
            if tnc_name is 'IGATE':
                echo_colorized_error('IGATE was used as the name for a TNC in the configuration, this name is reserved')
                return False
            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.buffers.ReconnectingPacketBuffer(apex.aprs.AprsKiss(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.buffers.ReconnectingPacketBuffer(apex.aprs.AprsKiss(apex.kiss.KissTcp(host=tcp_host, tcp_port=tcp_port)))
            else:
                echo_colorized_error("""Invalid configuration, must have both com_port and baud set or tcp_host and
                           tcp_port set in TNC sections of configuration file""")
                return False

            if not config.has_option(section, 'kiss_init'):
                echo_colorized_error('Invalid configuration, must have kiss_init set in TNC sections of configuration file')
                return False
            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:
                echo_colorized_error('Invalid configuration, value assigned to kiss_init was not recognized: %s'
                                     % kiss_init_string)
                return False

            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': NonrepeatingBuffer(kiss_tnc, port_name, tnc_port),
                                       'tnc_port': tnc_port}

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

    return True


@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):
    # load the configuration, if it fails, exit
    if not configure(configfile, verbose):
        return

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

    # start the plugins
    try:
        plugin_loaders = get_plugins()
        if not len(plugin_loaders):
            echo_colorized_warning('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, aprsis))
            new_thread.start()
            plugin_threads.append(new_thread)
    except IOError:
        echo_colorized_warning('plugin directory not found, will only display incoming messages.')

    signal.signal(signal.SIGINT, sigint_handler)

    # process all incoming frames by sending them to each of the plugins.
    if verbose:
        click.echo('Starting packet processing...')
    while running:
        something_read = False
        for port_name in port_map.keys():
            port = port_map[port_name]
            frame = port['tnc'].read()
            if frame:
                for plugin_module in plugin_modules:
                    something_read = True
                    plugin_module.handle_packet(frame, port, port_name)

        if something_read is False:
            time.sleep(1)


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

Syncleus / apex

find_config() F last analyzed 2016-12-13 21:00 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

find_config() F
last analyzed 2016-12-13 21:00 UTC