Merge pull request #6 from kerel-fs/feature/+tasks+managepy

Add common CLI manage.py; Add tasks

.gitignore

@@ -1,5 +1,6 @@
 # OGN stuff
 ogn.db
+beacons.db
 
 # Byte-compiled / optimized / DLL files
 __pycache__/
@@ -51,3 +52,10 @@ coverage.xml
 
 # Sphinx documentation
 docs/_build/
+
+# Editors
+*.swp
+*.swo
+
+# Python virtualenv
+env

CONTRIBUTORS

@@ -1 +1,2 @@
 Konstantin Gründger <konstantin.gruendger@web.de>
+Fabian P. Schmidt <kerel-fs@gmx.de>

README.md

@@ -4,3 +4,93 @@
 (https://travis-ci.org/Meisterschueler/ogn-python)
 [![Coverage Status](https://img.shields.io/coveralls/Meisterschueler/ogn-python.svg)]
 (https://coveralls.io/r/Meisterschueler/ogn-python)
+
+A python module for the [Open Glider Network](http://wiki.glidernet.org/).
+The submodule 'ogn.gateway' is an aprs client, saving all received beacons
+into a database with [SQLAlchemy](http://www.sqlalchemy.org/).
+Other submodules process this data.
+
+To schedule tasks like fetching ddb data,
+[Celery](http://www.celeryproject.org/) with [Redis](http://www.redis.io/) is used.
+
+
+## Installation and Setup
+1. Install python requirements
+```
+pip install -r requirements.txt
+```
+
+2. Install redis for asynchronous tasks ('ogn.collect.\*')
+```
+$ apt-get install redis-server
+```
+
+3. Create database
+```
+$ ./manage.py db.init
+```
+
+## Running the aprs client and task server
+This scripts run in the foreground and should be deamonized
+(eg. use [supervisord](http://supervisord.org/)).
+```
+# start aprs client
+$ ./manage.py gateway.run
+
+# start task server (make sure redis is up and running)
+$ celery -A ogn.collect worker -l info
+```
+
+## manage.py - CLI options
+```
+usage: manage.py [<namespace>.]<command> [<args>]
+
+positional arguments:
+  command     the command to run
+
+optional arguments:
+  -h, --help  show this help message and exit
+
+available commands:
+
+  [db]
+    init                   Initialize the database.
+    updateddb              Update the ddb data.
+
+  [gateway]
+    run                    Run the aprs client.
+
+  [logbook]
+    show                   Show a logbook for <airport_name> located at given position.
+
+  [show.receiver]
+    hardware_stats         Show some statistics of receiver hardware.
+    list                   Show a list of all receivers.
+    software_stats         Show some statistics of receiver sotware.
+```
+
+The task server must be running for `db.updateddb`.
+
+## TODO
+- [x] Write celery backend and add task 'fetchddb'
+- [x] Rewrite manage.py with <https://github.com/Birdback/manage.py> or flask-script
+- [x] Rename existing cli commands
+- [x] Document/Improve cli commands
+- [ ] Separate settings from module (currently at ogn/command/dbutils.py)
+- [ ] Enable granular data acquisition (eg. store receiver beacons only)
+- [ ] Database: Rename 'Flarm' to 'Device'?
+- [ ] Future Database-Migrations: Use Alembric?
+- [ ] Fix command/logbook.py (@Meisterschueler?)
+- [ ] Introduce scheduled tasks with 'celery beat' (eg. updateddb)
+
+### Scheduled tasks
+- ogn.collect.fetchddb (generate Flarm table)
+- ogn.collect.receiver (generate Receiver table)
+- ogn.collect.logbook  (generate TaoffLanding table)
+
+## How to use virtualenv
+```
+$ sudo apt-get install python-virtualenv
+
+$ virtualenv env
+$ source env/bin/activate

manage.py

@@ -0,0 +1,13 @@
+#!/usr/bin/env python
+
+from manager import Manager
+from ogn.commands import manager as command_manager
+from ogn.gateway.manage import manager as gateway_manager
+
+manager = Manager()
+manager.merge(command_manager)
+manager.merge(gateway_manager, namespace='gateway')
+
+
+if __name__ == '__main__':
+    manager.main()

ogn/collect/celery.py

@@ -0,0 +1,30 @@
+from __future__ import absolute_import
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from celery import Celery, Task
+from celery.signals import worker_init, worker_shutdown
+
+app = Celery('ogn.collect',
+             broker='redis://localhost:6379/0',
+             backend='redis://localhost:6379/0',
+             include=["ogn.collect.fetchddb"])
+
+DB_URI = 'sqlite:///beacons.db'
+
+@worker_init.connect
+def connect_db(signal, sender):
+    # Load settings like DB_URI...
+    engine = create_engine(DB_URI, echo=False)
+
+    Session = sessionmaker(bind=engine)
+    sender.app.session = Session()
+
+@worker_shutdown.connect
+def close_db(signal, sender):
+    sender.app.session.close()
+
+
+if __name__ == '__main__':
+    app.start()

ogn/collect/fetchddb.py

@@ -0,0 +1,29 @@
+from __future__ import absolute_import
+
+from celery.utils.log import get_task_logger
+from ogn.collect.celery import app
+
+from ogn.model import Flarm
+from ogn.utils import get_ddb
+
+logger = get_task_logger(__name__)
+
+
+@app.task
+def update_ddb_data():
+    logger.info("Update ddb data.")
+
+    app.session.query(Flarm).delete()
+
+    devices = get_ddb()
+    logger.info("Devices: %s"%str(devices))
+    app.session.bulk_save_objects(devices)
+
+    app.session.commit()
+    return len(devices)
+
+# TODO: Reimplement.
+def import_ddb_data(filename='custom.txt'):
+    flarms = get_ddb(filename)
+    db.session.bulk_save_objects(flarms)
+    session.commit()

ogn/collect/logbook.py

@@ -0,0 +1,72 @@
+from __future__ import absolute_import
+
+from datetime import datetime, timedelta
+
+from celery.utils.log import get_task_logger
+from ogn.collect.celery import app
+
+from sqlalchemy.sql import func, null
+from sqlalchemy import and_, or_, insert, between
+from sqlalchemy.sql.expression import case, true, false, label
+
+from ogn.model import Flarm, AircraftBeacon, TakeoffLanding
+
+logger = get_task_logger(__name__)
+
+
+@app.task
+def compute_takeoff_and_landing():
+    takeoff_speed = 30
+    landing_speed = 30
+
+    # get last takeoff_landing time as starting point for the following search
+    last_takeoff_landing_query = app.session.query(func.max(TakeoffLanding.timestamp))
+    last_takeoff_landing = last_takeoff_landing_query.one()[0]
+    if last_takeoff_landing is None:
+        last_takeoff_landing = datetime(2015, 1, 1, 0, 0, 0)
+
+    # make a query with current, previous and next position, so we can detect takeoffs and landings
+    sq = app.session.query(AircraftBeacon.address,
+                       func.lag(AircraftBeacon.address).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('address_prev'),
+                       func.lead(AircraftBeacon.address).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('address_next'),
+                       AircraftBeacon.timestamp,
+                       func.lag(AircraftBeacon.timestamp).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('timestamp_prev'),
+                       func.lead(AircraftBeacon.timestamp).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('timestamp_next'),
+                       AircraftBeacon.latitude,
+                       func.lag(AircraftBeacon.latitude).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('latitude_prev'),
+                       func.lead(AircraftBeacon.latitude).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('latitude_next'),
+                       AircraftBeacon.longitude,
+                       func.lag(AircraftBeacon.longitude).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('longitude_prev'),
+                       func.lead(AircraftBeacon.longitude).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('longitude_next'),
+                       AircraftBeacon.ground_speed,
+                       AircraftBeacon.track,
+                       func.lag(AircraftBeacon.track).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('track_prev'),
+                       func.lead(AircraftBeacon.track).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('track_next'),
+                       AircraftBeacon.ground_speed,
+                       func.lag(AircraftBeacon.ground_speed).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('ground_speed_prev'),
+                       func.lead(AircraftBeacon.ground_speed).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('ground_speed_next'),
+                       AircraftBeacon.altitude,
+                       func.lag(AircraftBeacon.altitude).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('altitude_prev'),
+                       func.lead(AircraftBeacon.altitude).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('altitude_next')) \
+        .filter(AircraftBeacon.timestamp > last_takeoff_landing) \
+        .order_by(func.date(AircraftBeacon.timestamp), AircraftBeacon.address, AircraftBeacon.timestamp) \
+        .subquery()
+
+    # find takeoffs and landings (look at the trigger_speed)
+    takeoff_landing_query = app.session.query(sq.c.address, sq.c.timestamp, sq.c.latitude, sq.c.longitude, sq.c.track, sq.c.ground_speed, sq.c.altitude, case([(sq.c.ground_speed>takeoff_speed, True), (sq.c.ground_speed<landing_speed, False)]).label('is_takeoff')) \
+        .filter(sq.c.address_prev == sq.c.address == sq.c.address_next) \
+        .filter(or_(and_(sq.c.ground_speed_prev < takeoff_speed,    # takeoff
+                         sq.c.ground_speed > takeoff_speed,
+                         sq.c.ground_speed_next > takeoff_speed),
+                    and_(sq.c.ground_speed_prev > landing_speed,    # landing
+                         sq.c.ground_speed < landing_speed,
+                         sq.c.ground_speed_next < landing_speed))) \
+        .order_by(func.date(sq.c.timestamp), sq.c.timestamp)
+
+    # ... and save them
+    ins = insert(TakeoffLanding).from_select((TakeoffLanding.address, TakeoffLanding.timestamp, TakeoffLanding.latitude, TakeoffLanding.longitude, TakeoffLanding.track, TakeoffLanding.ground_speed, TakeoffLanding.altitude, TakeoffLanding.is_takeoff), takeoff_landing_query)
+    app.session.execute(ins)
+    app.session.commit()
+
+
+

ogn/commands/__init__.py

@@ -0,0 +1,11 @@
+from .database import manager as database_manager
+from .showreceiver import manager as show_receiver_manager
+from .logbook import manager as logbook_manager
+
+from manager import Manager
+
+manager = Manager()
+
+manager.merge(database_manager, namespace='db')
+manager.merge(show_receiver_manager, namespace='show.receiver')
+manager.merge(logbook_manager, namespace='logbook')

ogn/commands/database.py

@@ -0,0 +1,22 @@
+from ogn.model import Base
+
+from manager import Manager
+manager = Manager()
+
+from ogn.collect.fetchddb import update_ddb_data
+
+@manager.command
+def init():
+    """Initialize the database."""
+    from ogn.commands.dbutils import engine
+    Base.metadata.create_all(engine)
+    print("Done.")
+
+
+@manager.command
+def updateddb():
+    """Update the ddb data."""
+    print("Updating ddb data...")
+    result = update_ddb_data.delay()
+    counter = result.get()
+    print("Imported %i devieces."%counter)

ogn/commands/dbutils.py

@@ -0,0 +1,6 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+engine = create_engine('sqlite:///beacons.db', echo=False)
+
+Session = sessionmaker(bind=engine)
+session = Session()

ogn/commands/logbook.py

@@ -1,68 +1,25 @@
+# -*- coding: utf-8 -*-
+
 from datetime import datetime, timedelta
 
 from sqlalchemy.sql import func, null
 from sqlalchemy import and_, or_, insert, between
 from sqlalchemy.sql.expression import case, true, false, label
 
-from ogn.db import session
 from ogn.model import Flarm, AircraftBeacon, TakeoffLanding
 
+from ogn.commands.dbutils import session
 
-def compute_takeoff_and_landing():
+from manager import Manager
-    takeoff_speed = 30
+manager = Manager()
-    landing_speed = 30
 
-    # get last takeoff_landing time as starting point for the following search
+@manager.command
-    last_takeoff_landing_query = session.query(func.max(TakeoffLanding.timestamp))
+def show(airport_name, latitude, longitude, altitude):
-    last_takeoff_landing = last_takeoff_landing_query.one()[0]
+    """Show a logbook for <airport_name> located at given position."""
-    if last_takeoff_landing is None:
+    latitude = float(latitude)
-        last_takeoff_landing = datetime(2015, 1, 1, 0, 0, 0)
+    longitude = float(longitude)
-
+    altitude = float(altitude)
-    # make a query with current, previous and next position, so we can detect takeoffs and landings
+    # get_logbook('Königsdorf', 47.83, 11.46, 601)
-    sq = session.query(AircraftBeacon.address,
-                       func.lag(AircraftBeacon.address).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('address_prev'),
-                       func.lead(AircraftBeacon.address).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('address_next'),
-                       AircraftBeacon.timestamp,
-                       func.lag(AircraftBeacon.timestamp).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('timestamp_prev'),
-                       func.lead(AircraftBeacon.timestamp).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('timestamp_next'),
-                       AircraftBeacon.latitude,
-                       func.lag(AircraftBeacon.latitude).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('latitude_prev'),
-                       func.lead(AircraftBeacon.latitude).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('latitude_next'),
-                       AircraftBeacon.longitude,
-                       func.lag(AircraftBeacon.longitude).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('longitude_prev'),
-                       func.lead(AircraftBeacon.longitude).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('longitude_next'),
-                       AircraftBeacon.ground_speed,
-                       AircraftBeacon.track,
-                       func.lag(AircraftBeacon.track).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('track_prev'),
-                       func.lead(AircraftBeacon.track).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('track_next'),
-                       AircraftBeacon.ground_speed,
-                       func.lag(AircraftBeacon.ground_speed).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('ground_speed_prev'),
-                       func.lead(AircraftBeacon.ground_speed).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('ground_speed_next'),
-                       AircraftBeacon.altitude,
-                       func.lag(AircraftBeacon.altitude).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('altitude_prev'),
-                       func.lead(AircraftBeacon.altitude).over(order_by=and_(AircraftBeacon.address, AircraftBeacon.timestamp)).label('altitude_next')) \
-        .filter(AircraftBeacon.timestamp > last_takeoff_landing) \
-        .order_by(func.date(AircraftBeacon.timestamp), AircraftBeacon.address, AircraftBeacon.timestamp) \
-        .subquery()
-
-    # find takeoffs and landings (look at the trigger_speed)
-    takeoff_landing_query = session.query(sq.c.address, sq.c.timestamp, sq.c.latitude, sq.c.longitude, sq.c.track, sq.c.ground_speed, sq.c.altitude, case([(sq.c.ground_speed>takeoff_speed, True), (sq.c.ground_speed<landing_speed, False)]).label('is_takeoff')) \
-        .filter(sq.c.address_prev == sq.c.address == sq.c.address_next) \
-        .filter(or_(and_(sq.c.ground_speed_prev < takeoff_speed,    # takeoff
-                         sq.c.ground_speed > takeoff_speed,
-                         sq.c.ground_speed_next > takeoff_speed),
-                    and_(sq.c.ground_speed_prev > landing_speed,    # landing
-                         sq.c.ground_speed < landing_speed,
-                         sq.c.ground_speed_next < landing_speed))) \
-        .order_by(func.date(sq.c.timestamp), sq.c.timestamp)
-
-    # ... and save them
-    ins = insert(TakeoffLanding).from_select((TakeoffLanding.address, TakeoffLanding.timestamp, TakeoffLanding.latitude, TakeoffLanding.longitude, TakeoffLanding.track, TakeoffLanding.ground_speed, TakeoffLanding.altitude, TakeoffLanding.is_takeoff), takeoff_landing_query)
-    session.execute(ins)
-    session.commit()
-
-
-def get_logbook(airport_name, latitude, longitude, altitude):
     latmin = latitude - 0.15
     latmax = latitude + 0.15
     lonmin = longitude - 0.15
@@ -132,8 +89,3 @@ def get_logbook(airport_name, latitude, longitude, altitude):
     none_aircraft_replacer = lambda aircraft_object: '(unknown)' if aircraft_object is None else aircraft_object
     for [reftime, address, takeoff, takeoff_track, landing, landing_track, duration, registration, aircraft] in logbook_query.all():
         print('%10s   %8s (%2s)   %8s (%2s)   %8s   %8s   %s' % (reftime.date(), none_datetime_replacer(takeoff), none_track_replacer(takeoff_track), none_datetime_replacer(landing), none_track_replacer(landing_track), none_timedelta_replacer(duration), none_registration_replacer(registration, address), none_aircraft_replacer(aircraft)))
-
-if __name__ == '__main__':
-    compute_takeoff_and_landing()
-    get_logbook('Königsdorf', 47.83, 11.46, 601)
-

ogn/commands/showreceiver.py

@@ -3,14 +3,19 @@ from datetime import datetime, timedelta
 from sqlalchemy.sql import func
 from sqlalchemy import distinct, and_
 
-from ogn.db import session
 from ogn.model import ReceiverBeacon
 
+from ogn.commands.dbutils import session
+
 back_24h = datetime.utcnow() - timedelta(days=1)
 receiver_messages_per_24h = 24*60 / 5
 
+from manager import Manager
+manager = Manager()
 
-def get_receiver_info():
+@manager.command
+def list():
+    """Show a list of all receivers."""
     sq = session.query(distinct(ReceiverBeacon.name).label('name'), func.max(ReceiverBeacon.timestamp).label('lastseen'), func.count(ReceiverBeacon.name).label('messages_count')).\
         filter(ReceiverBeacon.timestamp > back_24h).\
         group_by(ReceiverBeacon.name).\
@@ -25,7 +30,9 @@ def get_receiver_info():
         print('%9s: %3d%% avail, %s, %s ' % (receiver.name, 100.0*float(messages_count/receiver_messages_per_24h), receiver.version, receiver.platform))
 
 
-def get_software_stats():
+@manager.command
+def software_stats():
+    """Show some statistics of receiver sotware."""
     sq = session.query(ReceiverBeacon.name, func.max(ReceiverBeacon.timestamp).label('lastseen')).\
         filter(ReceiverBeacon.timestamp > back_24h).\
         group_by(ReceiverBeacon.name).\
@@ -41,7 +48,9 @@ def get_software_stats():
         print('%5s: %s' % (version, count))
 
 
-def get_hardware_stats():
+@manager.command
+def hardware_stats():
+    """Show some statistics of receiver hardware."""
     sq = session.query(ReceiverBeacon.name, func.max(ReceiverBeacon.timestamp).label('lastseen')).\
         filter(ReceiverBeacon.timestamp > back_24h).\
         group_by(ReceiverBeacon.name).\
@@ -55,9 +64,3 @@ def get_hardware_stats():
     print('\n--- Platforms ---')
     for [platform, count] in platforms.all():
         print('%7s: %s' % (platform, count))
-
-
-if __name__ == '__main__':
-    get_receiver_info()
-    get_software_stats()
-    get_hardware_stats()

ogn/db.py

@@ -1,15 +0,0 @@
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
-
-from .model import Base, AircraftBeacon, ReceiverBeacon, Flarm
-
-
-# prepare db
-#engine = create_engine('sqlite:///:memory:', echo=False)
-engine = create_engine('sqlite:///ogn.db', echo=False)
-#engine = create_engine('postgresql://postgres:secretpass@localhost:5432/ogn')
-
-Base.metadata.create_all(engine)
-
-Session = sessionmaker(bind=engine)
-session = Session()

ogn/db_utils.py

@@ -1,23 +0,0 @@
-from ogn.db import session
-from ogn.model import Flarm
-from ogn.ognutils import get_ddb
-
-
-def put_into_db(beacon):
-    session.add(beacon)
-    session.commit()
-
-
-def fill_flarm_db():
-    session.query(Flarm).delete()
-
-    flarms = get_ddb()
-    session.bulk_save_objects(flarms)
-
-    flarms = get_ddb('custom.txt')
-    session.bulk_save_objects(flarms)
-
-    session.commit()
-
-if __name__ == '__main__':
-    fill_flarm_db()

ogn/gateway/__init__.py

@@ -0,0 +1,70 @@
+import socket
+from time import time
+
+from ogn.gateway import settings
+from ogn.commands.dbutils import session
+from ogn.aprs_parser import parse_aprs
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from ogn.model import Base
+
+MODULE_VERSION = "0.1"
+
+class ognGateway:
+    def __init__(self):
+        pass
+
+    def connect_db(self):
+        self.session = session
+
+    def connect(self, aprs_user):
+        if len(aprs_user) < 3 or len(aprs_user) > 9:
+            print("aprs_user must be a string of 3-9 characters")
+        # 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 = 'user %s pass %s vers ogn-gateway-python %s %s\n'  % (aprs_user, settings.APRS_PASSCODE, MODULE_VERSION, settings.APRS_FILTER)
+        self.sock.send(login.encode())
+        self.sock_file = self.sock.makefile('rw')
+
+
+    def run(self):
+        keepalive_time = time()
+        while True:
+            if time()-keepalive_time > 60:
+                self.sock.send("#keepalive".encode())
+                keepalive_time = time()
+
+            # Read packet string from socket
+            try:
+                packet_str = self.sock_file.readline().strip()
+            except socket.error:
+                print('Socket error on readline')
+                continue
+
+            # 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:
+                print('Read returns zero length string. Failure.  Orderly closeout')
+                break
+
+            self.proceed_line(packet_str)
+        # close everything
+        print('Close socket')
+        self.sock.shutdown(0)
+        self.sock.close()
+
+    def proceed_line(self, line):
+        try:
+            beacon = parse_aprs(line)
+        except Exception as e:
+            print('Failed to parse line: %s' % line)
+            print('Reason: %s' % e)
+            return
+
+        if beacon is not None:
+            self.session.add(beacon)
+            self.session.commit()

ogn/gateway/manage.py

@@ -0,0 +1,19 @@
+from ogn.gateway import ognGateway
+
+DB_URI = 'sqlite:///beacons.db'
+
+from manager import Manager
+manager = Manager()
+
+@manager.command
+def run(aprs_user="anon-dev"):
+    """Run the aprs client."""
+    gateway = ognGateway()
+    print("Start OGN gateway")
+    gateway.connect_db()
+    gateway.connect(aprs_user)
+    try:
+        gateway.run()
+    except KeyboardInterrupt:
+        pass
+    print("\nOGN gateway Exit")

ogn/ognproceed.py

@@ -1,65 +0,0 @@
-import socket
-from time import time
-
-from ogn import db_utils
-from ogn import settings
-from ogn.aprs_parser import parse_aprs
-
-
-def proceed():
-    # create socket, connect to server, login and make a file object associated with the socket
-    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-    sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
-    sock.connect((settings.APRS_SERVER_HOST, settings.APRS_SERVER_PORT))
-    login = 'user %s pass %s vers PyGrabber 1.0 %s\n'  % (settings.APRS_USER, settings.APRS_PASSCODE, settings.APRS_FILTER)
-    sock.send(login.encode())
-    sock_file = sock.makefile('rw')
-
-    keepalive_time = time()
-
-    try:
-        while True:
-            if time()-keepalive_time > 60:
-                sock.send("#keepalive".encode())
-                keepalive_time = time()
-
-            # Read packet string from socket
-            try:
-                packet_str = sock_file.readline().strip()
-            except socket.error:
-                print('Socket error on readline')
-                continue
-
-            # 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:
-                print('Read returns zero length string. Failure.  Orderly closeout')
-                break
-
-            proceed_line(packet_str)
-    finally:
-        # close everything
-        print('Close socket')
-        sock.shutdown(0)
-        sock.close()
-
-
-def proceed_line(line):
-    try:
-        result = parse_aprs(line)
-    except Exception as e:
-        print('Failed to parse line: %s' % line)
-        print('Reason: %s' % e)
-        return
-
-    if result is not None:
-        db_utils.put_into_db(result)
-
-if __name__ == '__main__':
-    while True:
-        try:
-            print("Start Python_Test")
-            proceed()
-            print("Python Test Exit")
-        except OSError as e:
-            print("OSError %s" % e)

requirements.txt

@@ -2,3 +2,5 @@ SQLAlchemy==1.0.8
 nose==1.3.7
 coveralls==0.4.4
 geopy==1.11.0
+manage.py==0.2.10
+celery[redis]>=3.1,<3.2

tests/test_ognutils.py

@@ -1,5 +1,5 @@
 import unittest
-from ogn.ognutils import get_ddb, get_country_code
+from ogn.utils import get_ddb, get_country_code
 from ogn.model.address_origin import AddressOrigin