Merge pull request #273 from kb1lqc/telemetrySQL

Fix Faraday-Telemetry For Use on Raspberry Pi

faraday/aprs.py

@@ -102,37 +102,37 @@ def aprs_worker(config):
     rate = config.getint("APRSIS", "RATE")
 
     # Local variable initialization
-    telemSequence = 0
     conn = False
 
     # Start infinite loop to send station data to APRS-IS
     while True:
-        # Query telemetry database for station data
-        stations = getStations()
-        stationData = getStationData(stations)
-
-        # Indicate number of stations tracking
-        str = "Tracking {0} Faraday stations..."
-        logger.info(str.format(len(stations)))
-
-        # Iterate through all stations sending telemetry and position data
-        if not conn:
-            sock = connectAPRSIS()
-        try:
-            conn = sendPositions(telemSequence, stationData, sock)
+        # Count through 8280 telemetry sequences (BASE91 limit)
+        for sequence in range(8280):
+            # Query telemetry database for station data
+            stations = getStations()
+            stationData = getStationData(stations)
+
+            # Indicate number of stations tracking
+            str = "Tracking {0} Faraday stations..."
+            logger.info(str.format(len(stations)))
+
+            # Iterate through all stations sending telemetry and position data
+            if not conn:
+                sock = connectAPRSIS()
+            try:
+                conn = sendPositions(sequence, stationData, sock)
 
-            # Just send labels, Parameters, and Equations every 10th loop
-            if telemSequence % 10 == 0:
-                sendTelemLabels(stationData, sock)
-                sendParameters(stationData, sock)
-                sendEquations(stationData, sock)
-            telemSequence += 1
+                # Just send labels, Parameters, and Equations every 10th loop
+                if sequence % 10 == 0:
+                    sendTelemLabels(stationData, sock)
+                    sendParameters(stationData, sock)
+                    sendEquations(stationData, sock)
 
-        except StandardError as e:
-            logger.error(e)
+            except StandardError as e:
+                logger.error(e)
 
-        # Sleep for intended update rate (seconds)
-        sleep(rate)
+            # Sleep for intended update rate (seconds)
+            sleep(rate)
 
 
 # Now act upon the command line arguments
@@ -166,7 +166,14 @@ def getStations():
     logger.debug(url)
 
     r = requests.get(url)
-    results = r.json()
+
+    try:
+        results = r.json()
+
+    except ValueError as e:
+        # JSON ValueError error, just return a blank value
+        logger.error(e)
+        results = ''
 
     # Return extracted JSON data
     return results

faraday/telemetry.py

@@ -65,6 +65,36 @@
 args = parser.parse_args()
 
 
+def openDB():
+    '''
+    Open telemetry database and return a SQLite3 connection
+
+    :return: SQLite connection
+    '''
+    # Read in name of database from configuration file
+    try:
+        dbFilename = telemetryConfig.get("DATABASE", "FILENAME")
+        dbPath = os.path.join(faradayHelper.userPath, 'lib', dbFilename)
+        dbFilename = os.path.join(dbPath)
+
+    except ConfigParser.Error as e:
+        logger.error("ConfigParse.Error: " + str(e))
+
+    # Connect to database, enter WAL mode, commit SQL
+    try:
+
+        conn = sqlite3.connect(dbFilename)
+        conn.execute("PRAGMA journal_mode=WAL;")
+        conn.commit()
+
+    except sqlite3.Error as e:
+        logger.error("Sqlite3.error: " + str(e))
+        conn.close()
+
+    # Return SQLite3 connection
+    return conn
+
+
 def initializeTelemetryConfig():
     '''
     Initialize telemetry configuration file from telemetry.sample.ini
@@ -299,7 +329,9 @@ def telemetry_worker(config):
                         logger.error("KeyError: " + str(e))
 
                     else:
-                        sqlInsert(parsedTelemetry)
+                        # Successful decode, open database and save telemetry
+                        workerDB = openDB()
+                        sqlInsert(workerDB, parsedTelemetry)
                         telemetryDicts[str(callsign) + str(nodeid)].append(parsedTelemetry)
         time.sleep(1)  # Slow down main while loop
 
@@ -629,11 +661,14 @@ def initDB():
         # after connecting. Close the database when complete.
         try:
             with open(dbSchema, 'rt') as f:
-                conn = sqlite3.connect(dbFilename)
-                cur = conn.cursor()
+                # Open connection to database, create table and WAL mode, close
+                initConn = openDB()
+                cur = initConn.cursor()
                 schema = f.read()
                 cur.executescript(schema)
-            conn.close()
+                cur.execute("PRAGMA journal_mode=WAL;")
+                initConn.commit()
+            initConn.close()
 
         except sqlite3.Error as e:
             logger.error("Sqlite3.Error: " + str(e))
@@ -703,25 +738,13 @@ def createTelemetryList(data):
     return temp
 
 
-def sqlInsert(data):
+def sqlInsert(dbConn, data):
     """
     Takes in a data tuple and inserts int into the telemetry SQLite table
 
     :param data: Telemetry dictionary
     :return: Status True or False on SQL insertion success
     """
-
-    # Read in name of database
-    try:
-        dbFilename = telemetryConfig.get("DATABASE", "FILENAME")
-        dbPath = os.path.join(faradayHelper.userPath, 'lib', dbFilename)
-        logger.debug("Telemetry Database: " + dbPath)
-        db = os.path.join(dbPath)
-
-    except ConfigParser.Error as e:
-        logger.error("ConfigParse.Error: " + str(e))
-        return False
-
     # Change dictionary into list with proper order
     telem = createTelemetryList(data)
 
@@ -733,29 +756,21 @@ def sqlInsert(data):
         paramSubs = ",".join(paramSubs)
         sql = "INSERT INTO TELEMETRY VALUES(" + paramSubs + ")"
 
-        # Connect to database, create SQL query, execute query, and close database
-        try:
-            conn = sqlite3.connect(db)
-
-        except sqlite3.Error as e:
-            logger.error("Sqlite3.Error: " + str(e))
-            return False
-
-        # Connect to database, create SQL query, execute query, and close database
+        # Connect to database, execute query, and close database
         try:
-            conn = sqlite3.connect(db)
-            # Use connection as context manager to rollback automatically if error
-            with conn:
-                conn.execute(sql, telem)
+            # Execute SQL
+            with dbConn:
+                dbConn.execute(sql, telem)
 
         except sqlite3.Error as e:
+            # An error occured, rollback and close connection, return False
             logger.error("Sqlite3.Error: " + str(e))
-            conn.rollback()
-            conn.close()
+            dbConn.rollback()
+            dbConn.close()
             return False
 
         # Completed, close database and return True
-        conn.close()
+        dbConn.close()
         return True
 
     else:
@@ -825,36 +840,25 @@ def queryDb(parameters):
     sql = sqlBeg + sqlWhereCall + sqlWhereID + sqlEpoch + sqlEnd
     logger.debug(sql)
 
-    # Read in name of database
-    try:
-        dbFilename = telemetryConfig.get("DATABASE", "FILENAME")
-        dbPath = os.path.join(faradayHelper.userPath, 'lib', dbFilename)
-        logger.debug("Telemetry Database: " + dbPath)
-        dbFilename = os.path.join(dbPath)
-
-    except ConfigParser.Error as e:
-        logger.error("ConfigParse.Error: " + str(e))
-        return False
-
-    # Connect to database, create SQL query, execute query, and close database
+    # Connect to database, execute query, and close database
     try:
-        conn = sqlite3.connect(dbFilename)
+        queryConn = openDB()
 
     except sqlite3.Error as e:
         logger.error("Sqlite3.Error: " + str(e))
         logger.error(paramTuple)
         return sqlData
 
-    conn.row_factory = sqlite3.Row  # Row_factory returns column/values
-    cur = conn.cursor()
+    queryConn.row_factory = sqlite3.Row  # Row_factory returns column/values
+    cur = queryConn.cursor()
 
     try:
         cur.execute(sql, paramTuple)
 
     except sqlite3.Error as e:
         logger.error("Sqlite3.Error: " + str(e))
         logger.error(paramTuple)
-        conn.close()
+        queryConn.close()
         return sqlData
 
     # Iterate through resulting data and create a list of dictionaries for JSON
@@ -868,11 +872,11 @@ def queryDb(parameters):
 
     except StandardError as e:
         logger.error("StandardError: " + str(e))
-        conn.close()
+        queryConn.close()
         return sqlData
 
     # Completed query, close database, return sqlData list of dictionaries
-    conn.close()
+    queryConn.close()
     return sqlData
 
 
@@ -943,25 +947,25 @@ def queryStationsDb(parameters):
         logger.error("ConfigParse.Error: " + str(e))
         return False
 
-    # Connect to database, create SQL query, execute query, and close database
+    # Connect to database, execute query, and close database
     try:
-        conn = sqlite3.connect(dbFilename)
+        queryStationsDB = openDB()
 
     except sqlite3.Error as e:
         logger.error("Sqlite3.error: " + str(e))
         logger.error(paramTuple)
         return sqlData
 
-    conn.row_factory = sqlite3.Row  # SQLite.Row returns columns,values
-    cur = conn.cursor()
+    queryStationsDB.row_factory = sqlite3.Row  # SQLite.Row returns columns,values
 
     try:
+        cur = queryStationsDB.cursor()
         cur.execute(sql, paramTuple)
 
     except sqlite3.Error as e:
         logger.error("Sqlite3.error: " + str(e))
         logger.error(paramTuple)
-        conn.close()
+        queryStationsDB.close()
         return sqlData
 
     # Parse through rows and create key:value dictionaries for each row.
@@ -976,11 +980,11 @@ def queryStationsDb(parameters):
 
     except StandardError as e:
         logger.error("StandardError: " + str(e))
-        conn.close()
+        queryStationsDB.close()
         return sqlData
 
     # Completed query, close database, return list of dictionary data for JSON
-    conn.close()
+    queryStationsDB.close()
     return sqlData
 
 

requirements.txt

@@ -20,3 +20,4 @@ pytest-flake8==0.8.1
 requests==2.12.5
 six==1.10.0
 Werkzeug==0.11.15
+aprslib>=0.6.46