changed device db import/handling

ogn/db_utils.py

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

ogn/model/flarm.py

@@ -1,8 +1,5 @@
-import re
-
 from sqlalchemy import Column, Integer, String, Unicode, Boolean, SmallInteger
 
-from .address_origin import AddressOrigin
 from .base import Base
 
 
@@ -23,46 +20,5 @@ class Flarm(Base):
 
     address_origin = Column(SmallInteger)
 
-    FLARMNET_LINE_LENGTH = 173
-
-    def parse_ogn(self, line):
-        PATTERN = "\'([FIO])\',\'(.{6})\',\'([^\']+)?\',\'([^\']+)?\',\'([^\']+)?\',\'([YN])\',\'([YN])\'"
-        ogn_re = re.compile(PATTERN)
-
-        result = ogn_re.match(line)
-        if result is None:
-            raise Exception("No valid string: %s" % line)
-
-        self.address_type = result.group(1)
-        self.address = result.group(2)
-        self.aircraft = result.group(3)
-        self.registration = result.group(4)
-        self.competition = result.group(5)
-        self.tracked = result.group(6) == "Y"
-        self.identified = result.group(7) == "Y"
-
-        self.address_origin = AddressOrigin.ogn_ddb
-
-    def parse_flarmnet(self, line):
-        rawString = self.hexToString(line)
-
-        self.address_type = None
-        self.address = rawString[0:6].strip()
-        self.name = rawString[6:27].strip()
-        self.airport = rawString[27:48].strip()
-        self.aircraft = rawString[48:69].strip()
-        self.registration = rawString[69:76].strip()
-        self.competition = rawString[76:79].strip()
-        self.frequency = rawString[79:89].strip()
-
-        self.address_origin = AddressOrigin.flarmnet
-
-    def hexToString(self, hexString):
-        result = ''
-        for i in range(0, len(hexString)-1, 2):
-            result += chr(int(hexString[i:i+2], 16))
-
-        return(result)
-
     def __repr__(self):
         return("<Flarm: %s,%s,%s,%s,%s,%s,%s,%s,%s,%s>" % (self.address_type, self.address, self.name, self.airport, self.aircraft, self.registration, self.competition, self.frequency, self.tracked, self.identified))

ogn/ognutils.py

@@ -1,21 +1,39 @@
-from urllib.request import urlopen
+import requests
+import csv
+from io import StringIO
 
-from ogn.model import Flarm
+from .model import Flarm, AddressOrigin
 
 from geopy.geocoders import Nominatim
 
+DDB_URL = "http://ddb.glidernet.org/download"
+
+
+def get_ddb(csvfile=None):
+    if csvfile is None:
+        r = requests.get(DDB_URL)
+        rows = '\n'.join(i for i in r.text.splitlines() if i[0] != '#')
+        address_origin = AddressOrigin.ogn_ddb
+    else:
+        r = open(csvfile, 'r')
+        rows = ''.join(i for i in r.readlines() if i[0] != '#')
+        address_origin = AddressOrigin.userdefined
+
+    data = csv.reader(StringIO(rows), quotechar="'", quoting=csv.QUOTE_ALL)
 
-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
-
+    for row in data:
         flarm = Flarm()
-        flarm.parse_ogn(line.decode())
+        flarm.address_type = row[0]
+        flarm.address = row[1]
+        flarm.aircraft = row[2]
+        flarm.registration = row[3]
+        flarm.competition = row[4]
+        flarm.tracked = row[5] == 'Y'
+        flarm.identified = row[6] == 'Y'
+
+        flarm.address_origin = address_origin
+
         devices.append(flarm)
 
     return devices

tests/model/test_flarm.py

@@ -1,38 +0,0 @@
-import unittest
-
-from ogn.model.address_origin import *
-from ogn.model.flarm import Flarm
-
-
-class TestStringMethods(unittest.TestCase):
-    def test_ddb(self):
-        flarm = Flarm()
-
-        flarm.parse_ogn("'F','DD9703','Twin Astir II','D-8203','7G','Y','N'\r\n")
-        self.assertEqual(flarm.address_type, 'F')
-        self.assertEqual(flarm.address, 'DD9703')
-        self.assertEqual(flarm.aircraft, 'Twin Astir II')
-        self.assertEqual(flarm.registration, 'D-8203')
-        self.assertEqual(flarm.competition, '7G')
-        self.assertTrue(flarm.tracked)
-        self.assertFalse(flarm.identified)
-
-        self.assertEqual(flarm.address_origin, AddressOrigin.ogn_ddb)
-
-    def test_flarmnet(self):
-        flarm = Flarm()
-
-        flarm.parse_flarmnet('444431323334486972616d205965616765722020202020202020204c535a46202020202020202020202020202020202056656e747573203263784d2020202020202020202052552d343731315836203132332e343536')
-        self.assertEqual(flarm.address, 'DD1234')
-        self.assertEqual(flarm.name, 'Hiram Yeager')
-        self.assertEqual(flarm.airport, 'LSZF')
-        self.assertEqual(flarm.aircraft, 'Ventus 2cxM')
-        self.assertEqual(flarm.registration, 'RU-4711')
-        self.assertEqual(flarm.competition, 'X6')
-        self.assertEqual(flarm.frequency, '123.456')
-
-        self.assertEqual(flarm.address_origin, AddressOrigin.flarmnet)
-
-
-if __name__ == '__main__':
-    unittest.main()

tests/test_ognutils.py

@@ -1,12 +1,27 @@
 import unittest
-from ogn.ognutils import get_devices_from_ddb, get_country_code
+from ogn.ognutils import get_ddb, get_country_code
+from ogn.model.address_origin import AddressOrigin
 
 
 class TestStringMethods(unittest.TestCase):
-    def test_get_devices_from_ddb(self):
-        devices = get_devices_from_ddb()
+    def test_get_devices(self):
+        devices = get_ddb()
         self.assertGreater(len(devices), 1000)
 
+    def test_get_ddb_from_file(self):
+        devices = get_ddb('tests/custom_ddb.txt')
+        self.assertEqual(len(devices), 3)
+        device = devices[0]
+
+        self.assertEqual(device.address, 'DD4711')
+        self.assertEqual(device.aircraft, 'HK36 TTC')
+        self.assertEqual(device.registration, 'D-EULE')
+        self.assertEqual(device.competition, '')
+        self.assertTrue(device.tracked)
+        self.assertTrue(device.identified)
+
+        self.assertEqual(device.address_origin, AddressOrigin.userdefined)
+
     def test_get_country_code(self):
         latitude = 48.0
         longitude = 11.0