kopia lustrzana https://github.com/glidernet/ogn-python
Merge pull request #14 from kerel-fs/logging
Improve logging in manage.py and fix ogn.gatewaypull/17/head
Fabian P. Schmidt
commit
49104499d9
|
|
@ -1,69 +1,88 @@
|
|||
import socket
|
||||
import logging
|
||||
from time import time
|
||||
|
||||
from ogn.gateway import settings
|
||||
from ogn.commands.dbutils import session
|
||||
from ogn.aprs_parser import parse_aprs
|
||||
from ogn.aprs_utils import create_aprs_login
|
||||
from ogn.exceptions import AprsParseError, OgnParseError
|
||||
from ogn.logger import logger
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from ogn.model import Base
|
||||
|
||||
|
||||
class ognGateway:
|
||||
def __init__(self):
|
||||
pass
|
||||
def __init__(self, aprs_user, aprs_filter=''):
|
||||
self.logger = logging.getLogger(__name__)
|
||||
self.logger.info("Connect to OGN as {} with filter '{}'".format(aprs_user, (aprs_filter if aprs_filter else 'full-feed')))
|
||||
self.aprs_user = aprs_user
|
||||
self.aprs_filter = aprs_filter
|
||||
|
||||
def connect_db(self):
|
||||
self.session = session
|
||||
|
||||
def connect(self, aprs_user):
|
||||
def connect(self):
|
||||
# create socket, connect to server, login and make a file object associated with the socket
|
||||
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
|
||||
self.sock.connect((settings.APRS_SERVER_HOST, settings.APRS_SERVER_PORT))
|
||||
|
||||
login = create_aprs_login(aprs_user, -1, settings.APRS_APP_NAME, settings.APRS_APP_VER, settings.APRS_FILTER)
|
||||
if self.aprs_filter:
|
||||
port = settings.APRS_SERVER_PORT_CLIENT_DEFINED_FILTERS
|
||||
else:
|
||||
port = settings.APRS_SERVER_PORT_FULL_FEED
|
||||
|
||||
self.sock.connect((settings.APRS_SERVER_HOST, port))
|
||||
self.logger.debug('Server port {}'.format(port))
|
||||
|
||||
login = create_aprs_login(self.aprs_user, -1, settings.APRS_APP_NAME, settings.APRS_APP_VER, self.aprs_filter)
|
||||
self.sock.send(login.encode())
|
||||
self.sock_file = self.sock.makefile('rw')
|
||||
|
||||
def disconnect(self):
|
||||
# close everything
|
||||
self.sock.shutdown(0)
|
||||
self.sock.close()
|
||||
self.logger.info('Disconnect')
|
||||
try:
|
||||
# close everything
|
||||
self.sock.shutdown(0)
|
||||
self.sock.close()
|
||||
except OSError as e:
|
||||
self.logger.error('Socket close error', exc_info=True)
|
||||
|
||||
def run(self, callback, autoreconnect=False):
|
||||
self.process_beacon = callback
|
||||
|
||||
def run(self):
|
||||
keepalive_time = time()
|
||||
while True:
|
||||
if time() - keepalive_time > settings.APRS_KEEPALIVE_TIME:
|
||||
logger.debug('Sending keepalive')
|
||||
self.sock.send("#keepalive".encode())
|
||||
try:
|
||||
keepalive_time = time()
|
||||
while True:
|
||||
if time() - keepalive_time > settings.APRS_KEEPALIVE_TIME:
|
||||
self.logger.info('Send keepalive')
|
||||
self.sock.send('#keepalive'.encode())
|
||||
keepalive_time = time()
|
||||
|
||||
# Read packet string from socket
|
||||
packet_str = self.sock_file.readline().strip()
|
||||
# Read packet string from socket
|
||||
packet_str = self.sock_file.readline().strip()
|
||||
|
||||
# A zero length line should not be return if keepalives are being sent
|
||||
# A zero length line will only be returned after ~30m if keepalives are not sent
|
||||
if len(packet_str) == 0:
|
||||
logger.warning('Read returns zero length string. Failure. Orderly closeout')
|
||||
break
|
||||
# A zero length line should not be return if keepalives are being sent
|
||||
# A zero length line will only be returned after ~30m if keepalives are not sent
|
||||
if len(packet_str) == 0:
|
||||
self.logger.warning('Read returns zero length string. Failure. Orderly closeout')
|
||||
break
|
||||
|
||||
self.proceed_line(packet_str)
|
||||
self.proceed_line(packet_str)
|
||||
except BrokenPipeError:
|
||||
self.logger.error('BrokenPipeError', exc_info=True)
|
||||
except socket.error:
|
||||
self.logger.error('socket.error', exc_info=True)
|
||||
|
||||
if autoreconnect:
|
||||
self.connect()
|
||||
else:
|
||||
return
|
||||
|
||||
def proceed_line(self, line):
|
||||
try:
|
||||
beacon = parse_aprs(line)
|
||||
self.logger.debug('Received beacon: {}'.format(beacon))
|
||||
except AprsParseError:
|
||||
logger.error('AprsParseError while parsing line: %s' % line, exc_info=True)
|
||||
self.logger.error('AprsParseError while parsing line: {}'.format(line), exc_info=True)
|
||||
return
|
||||
except OgnParseError:
|
||||
logger.error('OgnParseError while parsing line: ' % line, exc_info=True)
|
||||
self.logger.error('OgnParseError while parsing line: {}'.format(line), exc_info=True)
|
||||
return
|
||||
|
||||
if beacon is not None:
|
||||
self.session.add(beacon)
|
||||
self.session.commit()
|
||||
self.process_beacon(beacon)
|
||||
|
|
|
|||
|
|
@ -1,49 +1,45 @@
|
|||
import socket
|
||||
import logging
|
||||
|
||||
from ogn.gateway.client import ognGateway
|
||||
from ogn.logger import logger
|
||||
from ogn.commands.dbutils import session
|
||||
|
||||
from manager import Manager
|
||||
manager = Manager()
|
||||
|
||||
DB_URI = 'sqlite:///beacons.db'
|
||||
logging_formatstr = '%(asctime)s - %(levelname).4s - %(name)s - %(message)s'
|
||||
log_levels = ['CRITICAL', 'ERROR', 'WARNING', 'INFO', 'DEBUG']
|
||||
|
||||
|
||||
@manager.command
|
||||
def run(aprs_user="anon-dev"):
|
||||
def run(aprs_user='anon-dev', logfile='main.log', loglevel='INFO'):
|
||||
"""Run the aprs client."""
|
||||
|
||||
# User input validation
|
||||
if len(aprs_user) < 3 or len(aprs_user) > 9:
|
||||
print("aprs_user must be a string of 3-9 characters")
|
||||
print('aprs_user must be a string of 3-9 characters.')
|
||||
return
|
||||
if loglevel not in log_levels:
|
||||
print('loglevel must be an element of {}.'.format(log_levels))
|
||||
return
|
||||
|
||||
user_interrupted = False
|
||||
gateway = ognGateway()
|
||||
# Enable logging
|
||||
log_handlers = [logging.StreamHandler()]
|
||||
if logfile:
|
||||
log_handlers.append(logging.FileHandler(logfile))
|
||||
logging.basicConfig(format=logging_formatstr, level=loglevel, handlers=log_handlers)
|
||||
|
||||
print("Connect to DB")
|
||||
logger.info('Connect to DB')
|
||||
gateway.connect_db()
|
||||
print('Start ogn gateway')
|
||||
gateway = ognGateway(aprs_user)
|
||||
gateway.connect()
|
||||
|
||||
while user_interrupted is False:
|
||||
logger.info("Connect OGN gateway as {}".format(aprs_user))
|
||||
gateway.connect(aprs_user)
|
||||
def process_beacon(beacon):
|
||||
session.add(beacon)
|
||||
session.commit()
|
||||
|
||||
try:
|
||||
logger.info('Run gateway')
|
||||
gateway.run()
|
||||
except KeyboardInterrupt:
|
||||
logger.error('User interrupted', exc_info=True)
|
||||
user_interrupted = True
|
||||
except BrokenPipeError:
|
||||
logger.error('BrokenPipeError', exc_info=True)
|
||||
except socket.error:
|
||||
logger.error('Socket error', exc_info=True)
|
||||
try:
|
||||
gateway.run(callback=process_beacon, autoreconnect=True)
|
||||
except KeyboardInterrupt:
|
||||
print('\nStop ogn gateway')
|
||||
|
||||
try:
|
||||
logger.info('Close socket')
|
||||
gateway.disconnect()
|
||||
except OSError as e:
|
||||
print('Socket close error: {}'.format(e.strerror))
|
||||
logger.error('Socket close error', exc_info=True)
|
||||
|
||||
print("\nExit OGN gateway")
|
||||
gateway.disconnect()
|
||||
logging.shutdown()
|
||||
|
|
|
|||
|
|
@ -1,7 +1,8 @@
|
|||
APRS_SERVER_HOST = 'aprs.glidernet.org'
|
||||
APRS_SERVER_PORT = 10152
|
||||
APRS_SERVER_PORT_FULL_FEED = 10152
|
||||
APRS_SERVER_PORT_CLIENT_DEFINED_FILTERS = 14580
|
||||
|
||||
APRS_APP_NAME = 'ogn-gateway-python'
|
||||
APRS_APP_VER = '0.2'
|
||||
|
||||
APRS_FILTER = ''
|
||||
APRS_KEEPALIVE_TIME = 240
|
||||
|
|
|
|||
|
|
@ -1,17 +0,0 @@
|
|||
import logging
|
||||
|
||||
logger = logging.getLogger('main')
|
||||
logger.setLevel(logging.DEBUG)
|
||||
|
||||
filehandler = logging.FileHandler('main.log')
|
||||
filehandler.setLevel(logging.DEBUG)
|
||||
|
||||
consolehandler = logging.StreamHandler()
|
||||
consolehandler.setLevel(logging.INFO)
|
||||
|
||||
formatter = logging.Formatter("%(asctime)s - %(levelname).4s - %(name)s - %(message)s")
|
||||
filehandler.setFormatter(formatter)
|
||||
consolehandler.setFormatter(formatter)
|
||||
|
||||
logger.addHandler(filehandler)
|
||||
logger.addHandler(consolehandler)
|
||||
|
|
@ -1,39 +1,22 @@
|
|||
import unittest
|
||||
import unittest.mock as mock
|
||||
|
||||
import socket
|
||||
|
||||
from ogn.gateway.manage import run
|
||||
|
||||
|
||||
class GatewayTest(unittest.TestCase):
|
||||
|
||||
class GatewayManagerTest(unittest.TestCase):
|
||||
# try simple user interrupt
|
||||
@mock.patch('ogn.gateway.manage.ognGateway')
|
||||
def test_run_user_interruption(self, mock_gateway):
|
||||
instance = mock_gateway.return_value
|
||||
instance.run.side_effect = KeyboardInterrupt()
|
||||
|
||||
run("user_1")
|
||||
run(aprs_user="testuser")
|
||||
|
||||
instance.connect_db.assert_called_once_with()
|
||||
instance.connect.assert_called_once_with("user_1")
|
||||
instance.run.assert_called_once_with()
|
||||
instance.connect.assert_called_once_with()
|
||||
self.assertEqual(instance.run.call_count, 1)
|
||||
instance.disconnect.assert_called_once_with()
|
||||
|
||||
# make BrokenPipeErrors and a socket error (may happen) and then a user interrupt (important!)
|
||||
@mock.patch('ogn.gateway.manage.ognGateway')
|
||||
def test_run_multiple_errors(self, mock_gateway):
|
||||
instance = mock_gateway.return_value
|
||||
instance.run.side_effect = [BrokenPipeError(), socket.error(), KeyboardInterrupt()]
|
||||
instance.disconnect.side_effect = [True, True, OSError()]
|
||||
|
||||
run("user_2")
|
||||
|
||||
instance.connect_db.assert_called_once_with()
|
||||
self.assertEqual(instance.connect.call_count, 3)
|
||||
self.assertEqual(instance.run.call_count, 3)
|
||||
self.assertEqual(instance.disconnect.call_count, 3)
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,29 @@
|
|||
import unittest
|
||||
import unittest.mock as mock
|
||||
|
||||
from ogn.gateway.client import ognGateway
|
||||
from ogn.gateway.settings import APRS_APP_NAME, APRS_APP_VER
|
||||
|
||||
|
||||
class GatewayTest(unittest.TestCase):
|
||||
def test_initialisation(self):
|
||||
self.gw = ognGateway(aprs_user='testuser', aprs_filter='')
|
||||
self.assertEqual(self.gw.aprs_user, 'testuser')
|
||||
self.assertEqual(self.gw.aprs_filter, '')
|
||||
|
||||
@mock.patch('ogn.gateway.client.socket')
|
||||
def test_connect(self, mock_socket):
|
||||
self.gw = ognGateway(aprs_user='testuser', aprs_filter='')
|
||||
self.gw.connect()
|
||||
self.gw.sock.send.assert_called_once_with('user testuser pass -1 vers {} {}\n'.format(APRS_APP_NAME, APRS_APP_VER).encode('ascii'))
|
||||
self.gw.sock.makefile.asser_called_once_with('rw')
|
||||
|
||||
@mock.patch('ogn.gateway.client.socket')
|
||||
def test_disconnect(self, mock_socket):
|
||||
self.gw = ognGateway(aprs_user='testuser', aprs_filter='')
|
||||
self.gw.connect()
|
||||
self.gw.disconnect()
|
||||
self.gw.sock.shutdown.assert_called_once_with(0)
|
||||
self.gw.sock.close.assert_called_once_with()
|
||||
|
||||
|
||||
Ładowanie…
Reference in New Issue