Merge pull request #38 from Meisterschueler/update_receivers

Add celery task update-receivers

README.md

@@ -139,7 +139,7 @@ available commands:
 
   [show.receiver]
     hardware_stats         Show some statistics of receiver hardware.
-    list_all               Show a list of all receivers (NOT IMPLEMENTED).
+    list_all               Show a list of all receivers.
     software_stats         Show some statistics of receiver software.
 ```
 
@@ -151,7 +151,7 @@ Only the command `logbook.compute` requires a running task server (celery) at th
   - `import_ddb` - Import registered devices from the ddb
   - `import_file` - Import registered devices from a local file
 - ogn.collect.receiver
-  - `populate` - Generate Receiver table (NOT IMPLEMENTED)
+  - `update_receivers` - Populate/update receiver table
 - ogn.collect.logbook
   - `compute_takeoff_and_landing` - Generate TakeoffLanding table
 - ogn.collect.heatmap

config/default.py

@@ -11,6 +11,10 @@ CELERYBEAT_SCHEDULE = {
         'task': 'ogn.collect.heatmap.update_beacon_receiver_distance_all',
         'schedule': timedelta(minutes=5),
     },
+    'update-receiver-table': {
+        'task': 'ogn.collect.receiver.update_receivers',
+        'schedule': timedelta(minutes=5),
+    },
 }
 
 CELERY_TIMEZONE = 'UTC'

ogn/collect/celery.py

@@ -26,7 +26,9 @@ def close_db(signal, sender):
 
 app = Celery('ogn.collect',
              include=["ogn.collect.database",
+                      "ogn.collect.heatmap",
                       "ogn.collect.logbook",
-                      "ogn.collect.heatmap"])
+                      "ogn.collect.receiver"
+                      ])
 
 app.config_from_envvar("OGN_CONFIG_MODULE")

ogn/collect/database.py

@@ -1,9 +1,11 @@
-from ogn.utils import get_ddb
-from ogn.model import Device, AddressOrigin
-
 from celery.utils.log import get_task_logger
+
+from ogn.model import Device, AddressOrigin
+from ogn.utils import get_ddb
+
 from ogn.collect.celery import app
 
+
 logger = get_task_logger(__name__)
 
 
@@ -24,7 +26,7 @@ def import_ddb():
 
     logger.info("Import registered devices fom the DDB...")
     counter = update_devices(app.session, AddressOrigin.ogn_ddb, get_ddb())
-    logger.info("Imported %i devices." % counter)
+    logger.info("Imported {} devices.".format(counter))
 
 
 @app.task
@@ -33,4 +35,4 @@ def import_file(path='tests/custom_ddb.txt'):
 
     logger.info("Import registered devices from '{}'...".format(path))
     counter = update_devices(app.session, AddressOrigin.user_defined, get_ddb(path))
-    logger.info("Imported %i devices." % counter)
+    logger.info("Imported {} devices.".format(counter))

ogn/collect/logbook.py

@@ -86,6 +86,6 @@ def compute_takeoff_and_landing():
     result = app.session.execute(ins)
     counter = result.rowcount
     app.session.commit()
-    logger.debug("New/recalculated takeoffs and landings: %s" % counter)
+    logger.debug("New/recalculated takeoffs and landings: {}".format(counter))
 
     return counter

ogn/collect/receiver.py

@@ -0,0 +1,89 @@
+from sqlalchemy.sql import func, null
+from sqlalchemy.sql.functions import coalesce
+from sqlalchemy import and_, or_
+
+from celery.utils.log import get_task_logger
+
+from ogn.model import Receiver, ReceiverBeacon
+from ogn.utils import get_country_code
+from ogn.collect.celery import app
+
+logger = get_task_logger(__name__)
+
+
+@app.task
+def update_receivers():
+    """Update the receiver table."""
+    # get current receiver data
+    last_entry_sq = app.session.query(coalesce(func.max(Receiver.lastseen), '2015-01-01 00:00:00').label('last_entry')) \
+                               .subquery()
+
+    last_receiver_beacon_sq = app.session.query(ReceiverBeacon.name, func.min(ReceiverBeacon.timestamp).label('firstseen'), func.max(ReceiverBeacon.timestamp).label('lastseen')) \
+                                 .filter(ReceiverBeacon.timestamp >= last_entry_sq.c.last_entry) \
+                                 .group_by(ReceiverBeacon.name) \
+                                 .subquery()
+
+    # update existing receivers
+    sq = app.session.query(ReceiverBeacon.name, ReceiverBeacon.latitude, ReceiverBeacon.longitude, ReceiverBeacon.altitude, last_receiver_beacon_sq.c.firstseen, last_receiver_beacon_sq.c.lastseen, ReceiverBeacon.version, ReceiverBeacon.platform) \
+            .filter(and_(ReceiverBeacon.name == last_receiver_beacon_sq.c.name, ReceiverBeacon.timestamp == last_receiver_beacon_sq.c.lastseen)) \
+            .subquery()
+
+    # -set country code to None if lat or lon changed
+    upd = app.session.query(Receiver) \
+                     .filter(and_(Receiver.name == sq.c.name,
+                                  or_(Receiver.latitude != sq.c.latitude,
+                                      Receiver.longitude != sq.c.longitude)
+                                  )
+                             ) \
+                     .update({"latitude": sq.c.latitude,
+                              "longitude": sq.c.longitude,
+                              "country_code": None})
+
+    logger.info("Count of receivers who changed lat or lon: {}".format(upd))
+    app.session.commit()
+
+    # -update lastseen of known receivers
+    upd = app.session.query(Receiver) \
+                     .filter(Receiver.name == sq.c.name) \
+                     .update({"altitude": sq.c.altitude,
+                              "lastseen": sq.c.lastseen,
+                              "version": sq.c.version,
+                              "platform": sq.c.platform})
+
+    logger.info("Count of receivers who where updated: {}".format(upd))
+
+    # add new receivers
+    empty_sq = app.session.query(ReceiverBeacon.name, ReceiverBeacon.latitude, ReceiverBeacon.longitude, ReceiverBeacon.altitude, last_receiver_beacon_sq.c.firstseen, last_receiver_beacon_sq.c.lastseen, ReceiverBeacon.version, ReceiverBeacon.platform) \
+                          .filter(and_(ReceiverBeacon.name == last_receiver_beacon_sq.c.name,
+                                       ReceiverBeacon.timestamp == last_receiver_beacon_sq.c.lastseen)) \
+                          .outerjoin(Receiver, Receiver.name == ReceiverBeacon.name) \
+                          .filter(Receiver.name == null()) \
+                          .order_by(ReceiverBeacon.name)
+
+    for receiver_beacon in empty_sq.all():
+        receiver = Receiver()
+        receiver.name = receiver_beacon.name
+        receiver.latitude = receiver_beacon.latitude
+        receiver.longitude = receiver_beacon.longitude
+        receiver.altitude = receiver_beacon.altitude
+        receiver.firstseen = receiver_beacon.firstseen
+        receiver.lastseen = receiver_beacon.lastseen
+        receiver.version = receiver_beacon.version
+        receiver.platform = receiver_beacon.platform
+
+        app.session.add(receiver)
+        logger.info("{} added".format(receiver.name))
+
+    app.session.commit()
+
+    # update country code if None
+    unknown_country_query = app.session.query(Receiver) \
+                                       .filter(Receiver.country_code == null()) \
+                                       .order_by(Receiver.name)
+
+    for receiver in unknown_country_query.all():
+        receiver.country_code = get_country_code(receiver.latitude, receiver.longitude)
+        if receiver.country_code is not None:
+            logger.info("Updated country_code for {} to {}".format(receiver.name, receiver.country_code))
+
+    app.session.commit()

ogn/commands/logbook.py

@@ -21,7 +21,7 @@ def compute():
     print("Compute takeoffs and landings...")
     result = compute_takeoff_and_landing.delay()
     counter = result.get()
-    print("New/recalculated takeoffs/landings: %s" % counter)
+    print("New/recalculated takeoffs/landings: {}".format(counter))
 
 
 @manager.command
@@ -140,7 +140,7 @@ def show(airport_name, latitude, longitude, altitude):
         .outerjoin(Device, union_query.c.address == Device.address) \
         .order_by(union_query.c.reftime)
 
-    print('--- Logbook (%s) ---' % airport_name)
+    print('--- Logbook ({}) ---'.format(airport_name))
 
     def none_datetime_replacer(datetime_object):
         return '--:--:--' if datetime_object is None else datetime_object.time()

ogn/commands/showreceiver.py

@@ -13,8 +13,7 @@ receiver_beacons_per_day = 24 * 60 / 5
 
 @manager.command
 def list_all():
-    """Show a list of all receivers (NOT IMPLEMENTED)."""
-
+    """Show a list of all receivers."""
     timestamp_24h_ago = datetime.utcnow() - timedelta(days=1)
 
     sq = session.query(distinct(ReceiverBeacon.name).label('name'),

ogn/utils.py

@@ -5,6 +5,7 @@ from io import StringIO
 from .model import Device, AddressOrigin
 
 from geopy.geocoders import Nominatim
+from geopy.exc import GeopyError
 
 DDB_URL = "http://ddb.glidernet.org/download"
 
@@ -48,17 +49,19 @@ def get_trackable(ddb):
     l = []
     for i in ddb:
         if i.tracked and i.address_type in address_prefixes:
-            l.append('{}{}'.format(address_prefixes[i.address_type], i.address))
+            l.append("{}{}".format(address_prefixes[i.address_type], i.address))
     return l
 
 
 def get_country_code(latitude, longitude):
     geolocator = Nominatim()
-    location = geolocator.reverse("%f, %f" % (latitude, longitude))
     try:
-        country_code = location.raw["address"]["country_code"]
+        location = geolocator.reverse("{}, {}".format(latitude, longitude))
+        country_code = location.raw['address']['country_code']
     except KeyError:
         country_code = None
+    except GeopyError:
+        country_code = None
     return country_code
 
 
@@ -71,7 +74,6 @@ def haversine_distance(location0, location1):
     lon1 = radians(location1[1])
 
     distance = 6366000 * 2 * asin(sqrt((sin((lat0 - lat1) / 2))**2 + cos(lat0) * cos(lat1) * (sin((lon0 - lon1) / 2))**2))
-    print(distance)
     phi = degrees(atan2(sin(lon0 - lon1) * cos(lat1), cos(lat0) * sin(lat1) - sin(lat0) * cos(lat1) * cos(lon0 - lon1)))
 
     return distance, phi

tests/test_utils.py

@@ -1,4 +1,5 @@
 import unittest
+import unittest.mock as mock
 
 from ogn.utils import get_ddb, get_trackable, get_country_code, haversine_distance
 from ogn.model import AddressOrigin
@@ -44,6 +45,15 @@ class TestStringMethods(unittest.TestCase):
         country_code = get_country_code(latitude, longitude)
         self.assertEqual(country_code, None)
 
+    @mock.patch('ogn.utils.Nominatim')
+    def test_gec_country_code_exception(self, nominatim_mock):
+        from geopy.exc import GeocoderTimedOut
+        instance = nominatim_mock.return_value
+
+        instance.reverse.side_effect = GeocoderTimedOut('Too busy')
+        country_code = get_country_code(0, 0)
+        self.assertIsNone(country_code)
+
     def test_haversine_distance(self):
         # delta: one latitude degree
         location0 = (0, 0)