find_config() - Code Metrics - Inspection of "Improved the output of the program when no configu..." - Syncleus/apex - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 1a4597...fcb1e6 )

by Jeffrey

created 2016-09-22 15:22 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 compatability inside python2
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals

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

if sys.version_info < (3, 0):
    import ConfigParser  # noqa: F401
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:
                raise Exception(
                    "Must have either both com_port and baud set or tcp_host and tcp_port set in configuration file")
            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:
                raise Exception("KISS init mode not specified")
            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()
        for plugin_loader in plugin_loaders:
            loaded_plugin = loadPlugin(plugin_loader)
            plugins.append(loaded_plugin)
            threading.Thread(target=loaded_plugin.start, args=(config, port_map, packet_cache, aprsis)).start()
    except FileNotFoundError:
        click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
                   click.style('plugin directory not found, this program has nothing to do.', bold=True))
        return

    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 = apex.aprs.util.format_aprs_frame(frame)
                    print(port_name + " << " + 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("caught exception while reading packet: " + str(ex))

        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 compatability inside python2
19			from __future__ import absolute_import
20			from __future__ import division
21			from __future__ import print_function
22			from __future__ import unicode_literals
23
24			import os
25			import signal
26			import sys
27			import threading
28			import time
29			import traceback
30			import cachetools
31			import click
32
33			import apex.aprs
34			from apex.kiss import constants as kissConstants
35			from apex.pluginloader import getPlugins
36			from apex.pluginloader import loadPlugin
37
38			if sys.version_info < (3, 0):
39			import ConfigParser # noqa: F401
40			elif sys.version_info >= (3, 0):
41			import configparser
42
43			__author__ = 'Jeffrey Phillips Freeman (WI2ARD)'
44			__maintainer__ = "Jeffrey Phillips Freeman (WI2ARD)"
45			__email__ = "[email protected]"
46			__license__ = 'Apache License, Version 2.0'
47			__copyright__ = 'Copyright 2016, Syncleus, Inc. and contributors'
48			__credits__ = []
49
50
51			def find_config(config_paths, verbose):
52			config_file = 'apex.conf'
53			rc_file = '.apexrc'
54			cur_path = os.path.join(os.curdir, config_file)
55			home_path = os.path.join(os.path.expanduser("~"), rc_file)
56			etc_path = os.path.join('etc', config_file)
57			if config_paths is None:
58			config_paths = [cur_path, home_path, etc_path]
59			elif isinstance(config_paths, str):
60			config_paths = [config_paths]
61			elif not isinstance(config_paths, list):
62			raise TypeError('config_paths argument was neither a string nor a list')
63
64			if verbose:
65			click.echo('Searching for configuration file in the following locations: %s' % repr(config_paths))
66
67			config = configparser.ConfigParser()
68			for config_path in config_paths:
69			try:
70			if len(config.read(config_path)) > 0:
71			return config
72			except IOError:
73			pass
74
75			return None
76
77
78			@click.command(context_settings=dict(auto_envvar_prefix='APEX'))
79			@click.option('-c',
80			'--configfile',
81			type=click.Path(exists=True, file_okay=True, dir_okay=False, readable=True, resolve_path=True),
82			help='Configuration file for APEX.')
83			@click.option('-v', '--verbose', is_flag=True, help='Enables verbose mode.')
84			def main(verbose, configfile):
85
86			port_map = {}
87			config = find_config(configfile, verbose)
88			if config is None:
89			click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
90			click.style('No apex configuration found, can not continue.', bold=True))
91			return
92			for section in config.sections():
93			if section.startswith("TNC "):
94			tnc_name = section.split(" ")[1]
95			if config.has_option(section, 'com_port') and config.has_option(section, 'baud'):
96			com_port = config.get(section, 'com_port')
97			baud = config.get(section, 'baud')
98			kiss_tnc = apex.aprs.AprsKiss(com_port=com_port, baud=baud)
99			elif config.has_option(section, 'tcp_host') and config.has_option(section, 'tcp_port'):
100			tcp_host = config.get(section, 'tcp_host')
101			tcp_port = config.get(section, 'tcp_port')
102			kiss_tnc = apex.aprs.AprsKiss(host=tcp_host, tcp_port=tcp_port)
103			else:
104			raise Exception(
105			"Must have either both com_port and baud set or tcp_host and tcp_port set in configuration file")
106			kiss_init_string = config.get(section, 'kiss_init')
107			if kiss_init_string == 'MODE_INIT_W8DED':
108			kiss_tnc.start(kissConstants.MODE_INIT_W8DED)
109			elif kiss_init_string == 'MODE_INIT_KENWOOD_D710':
110			kiss_tnc.start(kissConstants.MODE_INIT_KENWOOD_D710)
111			elif kiss_init_string == 'NONE':
112			kiss_tnc.start()
113			else:
114			raise Exception("KISS init mode not specified")
115			for port in range(1, 1 + int(config.get(section, 'port_count'))):
116			port_name = tnc_name + '-' + str(port)
117			port_section = 'PORT ' + port_name
118			port_identifier = config.get(port_section, 'identifier')
119			port_net = config.get(port_section, 'net')
120			tnc_port = int(config.get(port_section, 'tnc_port'))
121			port_map[port_name] = {'identifier': port_identifier, 'net': port_net, 'tnc': kiss_tnc,
122			'tnc_port': tnc_port}
123			if config.has_section('APRS-IS'):
124			aprsis_callsign = config.get('APRS-IS', 'callsign')
125			if config.has_option('APRS-IS', 'password'):
126			aprsis_password = config.get('APRS-IS', 'password')
127			else:
128			aprsis_password = -1
129			aprsis_server = config.get('APRS-IS', 'server')
130			aprsis_server_port = config.get('APRS-IS', 'server_port')
131			aprsis = apex.aprs.AprsInternetService(aprsis_callsign, aprsis_password)
132			aprsis.connect(aprsis_server, int(aprsis_server_port))
133
134			def sigint_handler(signal, frame):
135			for port in port_map.values():
136			port['tnc'].close()
137			sys.exit(0)
138
139			signal.signal(signal.SIGINT, sigint_handler)
140
141			click.echo("Press ctrl + c at any time to exit")
142
143			packet_cache = cachetools.TTLCache(10000, 5)
144			# start the plugins
145			plugins = []
146			try:
147			plugin_loaders = getPlugins()
148			for plugin_loader in plugin_loaders:
149			loaded_plugin = loadPlugin(plugin_loader)
150			plugins.append(loaded_plugin)
151			threading.Thread(target=loaded_plugin.start, args=(config, port_map, packet_cache, aprsis)).start()
152			except FileNotFoundError:
153			click.echo(click.style('Error: ', fg='red', bold=True, blink=True) +
154			click.style('plugin directory not found, this program has nothing to do.', bold=True))
155			return
156
157			while 1:
158			something_read = False
159			try:
160			for port_name in port_map.keys():
161			port = port_map[port_name]
162			frame = port['tnc'].read()
163			if frame:
164			formatted_aprs = apex.aprs.util.format_aprs_frame(frame)
165			print(port_name + " << " + formatted_aprs)
166			for plugin in plugins:
167			something_read = True
168			plugin.handle_packet(frame, port, port_name)
169			except Exception as ex:
170			# We want to keep this thread alive so long as the application runs.
171			traceback.print_exc(file=sys.stdout)
172			click.echo("caught exception while reading packet: " + str(ex))
173
174			if something_read is False:
175			time.sleep(1)
176

Syncleus / apex

Push — master ( 1a4597...fcb1e6 )

find_config() D

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like