Added ognutils

ogn/db_utils.py

@@ -1,51 +1,18 @@
-from urllib.request import urlopen
+from ogn.db import session
-
+from ogn.model import Flarm
-from .db import session
+from ogn.ognutils import get_devices_from_ddb
-from .model import AddressOrigin, Flarm
 
 
-def get_devices_from_ddb():
+def put_into_db(beacon):
-    session.query(Flarm.address_origin == AddressOrigin.ogn_ddb).delete()
+    session.add(beacon)
-
-    response = urlopen("http://ddb.glidernet.org/download")
-    lines = response.readlines()
-    for line in lines:
-        if (line.decode()[0] == "#"):
-            continue
-
-        flarm = Flarm()
-        flarm.parse_ogn(line.decode())
-        session.add(flarm)
-
     session.commit()
 
 
-def get_devices_from_flarmnet():
+def fill_flarm_db():
-    session.query(Flarm.address_origin == AddressOrigin.flarmnet).delete()
+    session.query(Flarm).delete()
-
+    flarms = get_devices_from_ddb()
-    response = urlopen("http://flarmnet.org/files/data.fln")
+    session.bulk_save_objects(flarms)
-    lines = response.readlines()
-    for line in lines:
-        if (len(line) != Flarm.FLARMNET_LINE_LENGTH):
-            continue
-
-        flarm = Flarm()
-        flarm.parse_flarmnet(line.decode())
-        session.add(flarm)
-
     session.commit()
 
-
-def put_position_into_db(position):
-    session.add(position)
-    session.commit()
-
-
-def put_receiver_into_db(receiver):
-    session.add(receiver)
-    session.commit()
-
-
 if __name__ == '__main__':
-    get_devices_from_ddb()
+    fill_flarm_db()
-    get_devices_from_flarmnet()

ogn/ognproceed.py

@@ -4,7 +4,6 @@ from time import time
 from ogn import db_utils
 from ogn import settings
 from ogn.aprs_parser import parse_aprs
-from ogn.model import Position, Receiver
 
 
 def proceed():
@@ -53,10 +52,8 @@ def proceed_line(line):
         print('Reason: %s' % e)
         return
 
-    if isinstance(result, Position):
+    if result is not None:
-        db_utils.put_position_into_db(result)
+        db_utils.put_into_db(result)
-    elif isinstance(result, Receiver):
-        db_utils.put_receiver_into_db(result)
 
 if __name__ == '__main__':
     while True:

ogn/ognutils.py

@@ -0,0 +1,28 @@
+from urllib.request import urlopen
+
+from ogn.model import Flarm
+
+from geopy.geocoders import Nominatim
+
+
+def get_devices_from_ddb():
+    devices = list()
+
+    response = urlopen("http://ddb.glidernet.org/download")
+    lines = response.readlines()
+    for line in lines:
+        if (line.decode()[0] == "#"):
+            continue
+
+        flarm = Flarm()
+        flarm.parse_ogn(line.decode())
+        devices.append(flarm)
+
+    return devices
+
+
+def get_country_code(latitude, longitude):
+    geolocator = Nominatim()
+    location = geolocator.reverse("%f, %f" % (latitude, longitude))
+    country_code = location.raw["address"]["country_code"]
+    return country_code

tests/test_ognutils.py

@@ -0,0 +1,14 @@
+import unittest
+from ogn.ognutils import get_devices_from_ddb, get_country_code
+
+
+class TestStringMethods(unittest.TestCase):
+    def test_get_devices_from_ddb(self):
+        devices = get_devices_from_ddb()
+        self.assertGreater(len(devices), 1000)
+
+    def test_get_country_code(self):
+        latitude = 48.0
+        longitude = 11.0
+        country_code = get_country_code(latitude, longitude)
+        self.assertEquals(country_code, 'de')