more tests and many bugfixes for sqlcache

Author: kmodexc

pytrackunit/sqlcache.py

@@ -68,6 +68,7 @@
     unit text, 
     start int, 
     end int,
+    CONSTRAINT time CHECK (start < end),
     primary key(unit, start)
 )
 '''
@@ -85,6 +86,39 @@
 )
 '''
 
+def error_item_to_sql_item(_x,meta):
+    """
+    returns the error as a tuple and converts the time to unix timestamp (milliseconds)
+    """
+    _id = meta['id']
+    #print(x['time'])
+    _time = int(get_datetime(_x['time']).timestamp()*1000)
+    #print(_time,x['time'],get_datetime(x['time']))
+    _spn = _x['spn']
+    _fmi = _x['fmi']
+    _oc = _x['occurrenceCount']
+    if 'name' in _x:
+        _name = _x['name']
+    else:
+        _name = 'none'
+    if 'description' in _x:
+        _desc = _x['description']
+    else:
+        _desc = 'none'
+    return (_id,_time,_spn,_fmi,_oc,_name,_desc)
+def sql_item_to_error_item(obj):
+    """
+    the operation error_item_to_sql_item reversed
+    """
+    _x = {}
+    _x['id'] = obj[0]
+    _x['time'] = datetime.fromtimestamp(obj[1]/1000.0).isoformat()
+    _x['spn'] = obj[2]
+    _x['fmi'] = obj[3]
+    _x['occurrenceCount'] = obj[4]
+    _x['name'] = obj[5]
+    _x['description'] = obj[6]
+    return _x
 class SqlInsertIter:
     """iterator for tucache data"""
     def __init__(self, sqliter, meta=None, _db=None):
@@ -103,29 +137,12 @@ def __aiter__(self):
         return self
 
     async def __anext__(self):
-        def map_f(_x,meta):
-            _id = meta['id']
-            #print(x['time'])
-            _time = int(get_datetime(_x['time']).timestamp()*1000)
-            #print(_time,x['time'],get_datetime(x['time']))
-            _spn = _x['spn']
-            _fmi = _x['fmi']
-            _oc = _x['occurrenceCount']
-            if 'name' in _x:
-                _name = _x['name']
-            else:
-                _name = 'none'
-            if 'description' in _x:
-                _desc = _x['description']
-            else:
-                _desc = 'none'
-            return (_id,_time,_spn,_fmi,_oc,_name,_desc)
-
         try:
             data, meta = await self.sqliter.__anext__()
             if self._db is not None:
                 #print(data)
-                sqldata = map(lambda x: map_f(x,meta),data)
+                data = list(data)
+                sqldata = map(lambda x: error_item_to_sql_item(x,meta),data)
                 self.cur.executemany("INSERT INTO error VALUES (?,?,?,?,?,?,?)",sqldata)
             return data, meta
         except StopAsyncIteration as exc:
@@ -136,7 +153,17 @@ def map_f(_x,meta):
             print("Committed")
             self._db.commit()
             raise StopAsyncIteration from exc
-
+        except sqlite3.IntegrityError as exc1:
+            print("Integrety error with",meta)
+            print("Try to find double entry (insert side)")
+            sqldata = map(lambda x: error_item_to_sql_item(x,meta),data)
+            for in_item in sqldata:
+                try:
+                    self.cur.execute("INSERT INTO error VALUES (?,?,?,?,?,?,?)",in_item)
+                except sqlite3.IntegrityError as exc2:
+                    print("Item throwing Integrity error",in_item)
+            self._db.rollback()
+            raise sqlite3.IntegrityError from exc1
 class SqlCache:
     """Sql cache can cache trackunit data in Sqlite DB"""
     def __init__(self,auth=None,_dir=None,db_file="webdb.db",upstream_cache=None):
@@ -170,16 +197,16 @@ def get_history(self, unit, start):
             print(meta)
         except StopIteration:
             print("no element found")
-
     def get_candata(self,unit,start):
         """returns history data of a vehicle with start and end date"""
         #tbd
-    def get_errors_upstream(self, veh_id, start_ts, end_ts, previter=None):
+    def get_faults_upstream(self, veh_id, start_ts, end_ts, previter=None):
         """gets errors from upstream cache"""
         cur = self._db.cursor()
 
         cur.execute("INSERT INTO errormeta VALUES (?,?,?)",(veh_id,start_ts,end_ts))
-        self._db.commit()
+        # wait until data is in database and commit then
+        #self._db.commit()
 
         start = datetime.fromtimestamp(start_ts/1000.0)
         end = datetime.fromtimestamp(end_ts/1000.0)
@@ -200,48 +227,48 @@ def get_errors_upstream(self, veh_id, start_ts, end_ts, previter=None):
 
         return previter, _len
 
-    def get_errors(self, veh_id, start, end, previter=None):
-        """returns error in between the given datetime objects"""
-        start_ts = int(start.timestamp()*1000)
-        end_ts = int(end.timestamp()*1000)
-        return self.get_errors_unixts(veh_id, start_ts, end_ts, previter)
-
-    def get_errors_unixts(self, veh_id, start_ts, end_ts, previter=None):
+    def get_faults_unixts(self, veh_id, start_ts, end_ts, previter=None):
         """returns error in between the given datetime objects"""
 
         cur = self._db.cursor()
 
         # Query meta database to check whether there the data is in database
         try:
-            _, me_start, me_end = next(iter(cur.execute("""
-                select * from errormeta where unit = ? and 
-                (start <= ? and end >= ?) or 
-                (start <= ? and end >= ?) order by start
-                """,(veh_id,start_ts,start_ts,end_ts,end_ts))))
+            me_vehid, me_start, me_end = next(iter(cur.execute("""
+                select * from errormeta where unit = ? and (
+                (start <= ? and end > ?) or 
+                (start < ? and end >= ?) or
+                (start >= ? and end <= ?)
+                ) order by start
+                """,(veh_id,start_ts,start_ts,end_ts,end_ts,start_ts,end_ts))))
             me_start = int(float(me_start))
             me_end = int(float(me_end))
         except StopIteration:
-            print("Stop iteration. Get errors from upstream now")
-            return self.get_errors_upstream(veh_id,start_ts,end_ts,previter)
+            print("Stop iteration. Didnt find",\
+                veh_id,start_ts,end_ts,\
+                "Get errors from upstream now")
+            return self.get_faults_upstream(veh_id,start_ts,end_ts,previter)
+
+        print("Found block",me_vehid,me_start,me_end)
 
         # Depending on the start and end of the next block, apply divide and conquer
         # by recursive calls of this function.
         if me_start <= start_ts:
             if me_end < end_ts:
-                previter, cnt1 = self.get_errors_unixts(veh_id,start_ts,me_end,previter)
-                previter, cnt2 = self.get_errors_unixts(veh_id,me_end+1,start_ts,previter)
+                previter, cnt1 = self.get_faults_unixts(veh_id,start_ts,me_end,previter)
+                previter, cnt2 = self.get_faults_unixts(veh_id,me_end+1,end_ts,previter)
                 return previter,(cnt1+cnt2)
-            return self.get_errors_sql(veh_id,start_ts,end_ts,previter)
+            return self.get_faults_sql(veh_id,start_ts,end_ts,previter)
         if me_end >= end_ts:
-            previter, cnt1 = self.get_errors_unixts(veh_id,start_ts,me_start-1,previter)
-            previter, cnt2 = self.get_errors_unixts(veh_id,me_start,end_ts,previter)
+            previter, cnt1 = self.get_faults_unixts(veh_id,start_ts,me_start-1,previter)
+            previter, cnt2 = self.get_faults_unixts(veh_id,me_start,end_ts,previter)
             return previter,(cnt1+cnt2)
-        previter, cnt1 = self.get_errors_unixts(veh_id,start_ts,me_start-1,previter)
-        previter, cnt2 = self.get_errors_unixts(veh_id,me_start,me_end,previter)
-        previter, cnt3 = self.get_errors_unixts(veh_id,me_end+1,end_ts,previter)
+        previter, cnt1 = self.get_faults_unixts(veh_id,start_ts,me_start-1,previter)
+        previter, cnt2 = self.get_faults_unixts(veh_id,me_start,me_end,previter)
+        previter, cnt3 = self.get_faults_unixts(veh_id,me_end+1,end_ts,previter)
         return previter,(cnt1+cnt2+cnt3)
 
-    def get_errors_sql(self, veh_id, start_ts, end_ts, previter=None):
+    def get_faults_sql(self, veh_id, start_ts, end_ts, previter=None):
         """gets data of this period from db whether or not it was actually stored there"""
         cur = self._db.cursor()
 
@@ -252,6 +279,8 @@ def get_errors_sql(self, veh_id, start_ts, end_ts, previter=None):
             order by time
             """,(veh_id,start_ts,end_ts)))[0]
 
+        print("found",cnt,"in sql")
+
         meta = {}
         meta["id"] = veh_id
         meta["start"] = start_ts
@@ -261,9 +290,11 @@ def get_errors_sql(self, veh_id, start_ts, end_ts, previter=None):
             previter = TuIter()
 
         if cnt > 0:
-            previter.add(SqlIter(iter(cur.execute(\
+            previter.add(SqlIter(iter(map(lambda x: [x],map(sql_item_to_error_item,cur.execute(\
                 "select * from error where unit = ? and time >= ? and time <= ? order by time",\
-                    (veh_id,start_ts,end_ts))),meta))
+                    (veh_id,start_ts,end_ts))))),meta))
+        else:
+            print("could not find any item in block ", start_ts, end_ts,"for unit",veh_id)
 
         return previter, cnt
     def get_faults(self,veh_id,tdelta=None,previter=None):
@@ -273,14 +304,16 @@ def get_faults(self,veh_id,tdelta=None,previter=None):
         else:
             end = self.tdelta_end
         end = end.replace(hour=0,minute=0,second=0,microsecond=0)
-        if isinstance(tdelta,datetime):
-            start = end+tdelta
+        if isinstance(tdelta,timedelta):
+            start = end-tdelta
         else:
             irange = int(tdelta)
             if irange <= 0:
                 return []
             start = end-timedelta(days=irange)
-        return self.get_errors(veh_id,start,end,previter)
+        return self.get_faults_timedelta(veh_id,start,end,previter)
     def get_faults_timedelta(self,veh_id,start,end,previter=None):
-        """get_faults_timedelta"""
-        return self.get_errors(veh_id,start,end,previter)
+        """returns error in between the given datetime objects"""
+        start_ts = int(start.timestamp()*1000)
+        end_ts = int(end.timestamp()*1000)
+        return self.get_faults_unixts(veh_id, start_ts, end_ts, previter)