Merge pull request #1 from einsteinmaster/asyncio

Changed from mutlithreading to asyncio

Author: kmodexc

pyproject.toml

@@ -4,7 +4,8 @@ requires = [
 	"setuptools",
 	"wheel",
 	"matplotlib",
-	"requests",
+	"aiohttp",
+	"aiofiles",
 	"pytest"
 ]
 build-backend = "setuptools.build_meta"

pytrackunit/sqlcache.py

@@ -0,0 +1,44 @@
+"""Module for caching data in sql db"""
+
+import sqlite3
+import os.path
+
+CREATE_HISTORY_META = '''
+create table histmeta(
+    unit text, start text, end text
+)
+'''
+
+CREATE_HISTORY_TABLE = '''
+CREATE TABLE history
+(unit text not null, time text not null, event int, keyId text, location text, address text, heading int, speed real, 
+km real, run1 real,run2 real,run3 real,run4 real,runOdo real,temperature1 real,temperature2 real,
+input1 int,input2 int,input3 int,input4 int,input5 int,input6 int,input7 int,input8 int,input9 int,input10 int,
+output1 int,output2 int,output3 int,output4 int,output5 int,
+analogInput1 real,analogInput2 real,analogInput4 real,
+Input1ChangeCounter INT,Input2ChangeCounter INT,Input3ChangeCounter INT,Input4ChangeCounter INT,
+batteryLevel real,externalPower real,
+primary key (unit,time))
+'''
+
+class SqlCache:
+    """Sql cache can cache trackunit data in Sqlite DB"""
+    def __init__(self,db_file="webdb.db"):
+        create_tables = not os.path.isfile(db_file)
+        self.web_db_path = db_file
+        self._db = sqlite3.connect(self.web_db_path)
+        if create_tables:
+            cur = self._db.cursor()
+            cur.execute(CREATE_HISTORY_TABLE)
+            cur.execute(CREATE_HISTORY_META)
+            self._db.commit()
+    def get_history(self, unit, start):
+        """returns history data of a vehicle with start and end date"""
+        cur = self._db.cursor()
+        meta = next(cur.execute(
+            f"select * from histmeta where unit='{unit}' and start<='{start}' order by start"))
+        print(meta)
+    def get_candata(self,unit,start):
+        """returns history data of a vehicle with start and end date"""
+        #tbd
+        

pytrackunit/trackunit.py

@@ -2,9 +2,7 @@
 
 import json
 import os.path
-from math import ceil
-from multiprocessing import Pool
-from datetime import timedelta,datetime
+import asyncio
 from .tucache import TuCache
 
 class TrackUnit:
@@ -23,8 +21,7 @@ def __init__(self,config_filename=None,api_key=None,verbose=False):
             with open(config["apikey-location"],encoding="utf8") as file_apikey:
                 api_key = file_apikey.readline()
         self.cache = TuCache(('API',api_key),_dir=config["webcache-location"],verbose=verbose)
-        self.req_period = 30
-        self.tdelta_end = None
+
     @property
     def verbose(self):
         """returns verbose mode value. in verbose mode, diagnostic output is printed to console."""
@@ -33,62 +30,36 @@ def verbose(self):
     def verbose(self, value):
         """sets the verbose mode. in verbose mode, diagnostic output is printed to console."""
         self.cache.cache.verbose = value
-    def get(self,req):
-        """get method"""
-        url = r'https://api.trackunit.com/public/'+req
-        data = self.cache.get(url)
-        if data is None:
-            raise Exception("no data: "+str(data))
-        return data
+
     def get_unitlist(self,_type=None,sort_by_hours=True):
         """unitList method"""
-        data = self.get('Unit')
+        data = asyncio.run(self.cache.get_url('Unit'))
         if _type is not None:
             data = list(filter(lambda x: " " in x['name'] and _type in x['name'],data))
         if sort_by_hours:
             data.sort(key=lambda x: (x['run1'] if 'run1' in x else 0),reverse=True)
         return data
-    def general_daydiff_get(self,furl,tdelta,threads=1):
-        """returns data for timedependant requests for a given daydelta"""
-        if self.tdelta_end is None:
-            end = datetime.now()
-        else:
-            end = self.tdelta_end
-        end = end.replace(hour=0,minute=0,second=0,microsecond=0)
-        if isinstance(tdelta,datetime):
-            start = end+tdelta
-        else:
-            irange = int(tdelta)
-            if irange <= 0:
-                return []
-            start = end-timedelta(days=irange)
-        return self.general_time_range_get(furl,start,end,threads)
-    def general_time_range_get(self,furl,start=None,end=None,threads=1):
-        """returns data for timedependant requests for a start and enddate"""
-        total_data = []
-        days = (end-start).days
-        requests = []
-        for week in range(ceil(days/self.req_period)):
-            wstart = start+timedelta(days=week*self.req_period)
-            wend = wstart+timedelta(days=min(self.req_period,(end-wstart).days))
-            requests.append(furl(\
-                wstart.strftime("%Y-%m-%dT%H:%M:%S"),\
-                wend.strftime("%Y-%m-%dT%H:%M:%S")))
-        if threads > 1:
-            with Pool(threads) as _p:
-                map_result = _p.map(self.get,requests)
-        else:
-            map_result = map(self.get,requests)
-        for _r in map_result:
-            total_data += _r
-        return total_data
-    def get_history(self,veh_id,tdelta,threads=1):
+
+    async def _a_get_history(self,veh_id,tdelta):
+        """async getHistory method"""
+        data = []
+        _it, _ = self.cache.get_history(veh_id,tdelta)
+        async for _d in _it:
+            data += _d
+        return data
+
+    async def _a_get_candata(self,veh_id,tdelta=None):
+        """async getCanData method"""
+        data = []
+        _it, _ = self.cache.get_candata(veh_id,tdelta)
+        async for _d in _it:
+            data += _d
+        return data
+
+    def get_history(self,veh_id,tdelta):
         """getHistory method"""
-        return self.general_daydiff_get(lambda t1,t2: \
-            'Report/UnitHistory?unitId='+veh_id+'&from='+t1+'.0000001Z&to='+t2+'.0000000Z',\
-                tdelta,threads)
-    def get_candata(self,veh_id,tdelta=None,threads=1):
+        return asyncio.run(self._a_get_history(veh_id,tdelta))
+
+    def get_candata(self,veh_id,tdelta=None):
         """getCanData method"""
-        return self.general_daydiff_get(lambda t1,t2: \
-            'Report/UnitExtendedInfo?Id='+veh_id+'&from='+t1+'.0000001Z&to='+t2+'.0000000Z',\
-                tdelta,threads)
+        return asyncio.run(self._a_get_candata(veh_id,tdelta))

pytrackunit/tucache.py

@@ -1,46 +1,69 @@
 """tucache module"""
 
-import sqlite3
-import os.path
+from datetime import datetime, timedelta
+from math import ceil
 from .webcache import WebCache
+from .tuiter import ReqIter, TuIter
+
+URL_BASE = r'https://api.trackunit.com/public/'
 
 class TuCache:
     """tucache class"""
-    def __init__(self,auth=None,_dir=None,use_sqlite=False,verbose=False):
+    def __init__(self,auth=None,_dir=None,verbose=False):
         self.cache = WebCache(auth=auth,_dir=_dir,verbose=verbose)
-        self.use_sqlite = use_sqlite
-        if self.use_sqlite:
-            self.web_db_path = os.path.join(_dir,"webdb.db")
-            self._db = sqlite3.connect(self.web_db_path)
-            cur = self._db.cursor()
-            _res = list(cur.execute(
-                '''
-                SELECT name FROM sqlite_master WHERE type="table" AND name="history"
-                '''))
-            if len(_res) == 0:
-                cur.execute('''CREATE TABLE history
-                (unit text not null, time text not null, event int, keyId text, location text, address text, heading int, speed real, 
-                km real, run1 real,run2 real,run3 real,run4 real,runOdo real,temperature1 real,temperature2 real,
-                input1 int,input2 int,input3 int,input4 int,input5 int,input6 int,input7 int,input8 int,input9 int,input10 int,
-                output1 int,output2 int,output3 int,output4 int,output5 int,
-                analogInput1 real,analogInput2 real,analogInput4 real,
-                Input1ChangeCounter INT,Input2ChangeCounter INT,Input3ChangeCounter INT,Input4ChangeCounter INT,
-                batteryLevel real,externalPower real,
-                primary key (unit,time))''')
-                self._db.commit()
+        self.req_period = 30
+        self.tdelta_end = None
     def clean(self):
         """deletes all cached data"""
         self.cache.clean()
-    def get(self,url):
+    async def get_url(self,url):
         """takes the data from cache if possible. otherwise data is loaded from web"""
-        data = self.cache.get(url)
+        data = await self.cache.get(URL_BASE+url)
         if self.cache.return_only_cache_files:
             return [data]
         if self.cache.dont_return_data:
             return []
         if self.cache.dont_read_files and len(data) == 0:
             return []
-        # if self.use_sqlite and "api.trackunit.com/public/Report/UnitHistory" in url:
-        #     cur = self.db.cursor()
-        #     _data = list(map(lambda x: (x["time"])))
         return data.get('list')
+    def general_daydiff_get(self,furl,tdelta,previter=None):
+        """returns data for timedependant requests for a given daydelta"""
+        if self.tdelta_end is None:
+            end = datetime.now()
+        else:
+            end = self.tdelta_end
+        end = end.replace(hour=0,minute=0,second=0,microsecond=0)
+        if isinstance(tdelta,datetime):
+            start = end+tdelta
+        else:
+            irange = int(tdelta)
+            if irange <= 0:
+                return []
+            start = end-timedelta(days=irange)
+        return self.general_time_range_get(furl,start,end,previter)
+    def general_time_range_get(self,furl,start=None,end=None,previter=None):
+        """returns data for timedependant requests for a start and enddate"""
+        days = (end-start).days
+        requests = []
+        for week in range(ceil(days/self.req_period)):
+            wstart = start+timedelta(days=week*self.req_period)
+            wend = wstart+timedelta(days=min(self.req_period,(end-wstart).days))
+            requests.append(furl(\
+                wstart.strftime("%Y-%m-%dT%H:%M:%S"),\
+                wend.strftime("%Y-%m-%dT%H:%M:%S")))
+        internal_iter = ReqIter(self,iter(requests))
+        if previter is None:
+            previter = TuIter()
+        previter.add(internal_iter)
+        return previter,len(requests)
+
+    def get_history(self,veh_id,tdelta,previter=None):
+        """getHistory method"""
+        return self.general_daydiff_get(lambda t1,t2: \
+            'Report/UnitHistory?unitId='+veh_id+'&from='+t1+'.0000001Z&to='+t2+'.0000000Z',\
+                tdelta,previter)
+    def get_candata(self,veh_id,tdelta=None,previter=None):
+        """getCanData method"""
+        return self.general_daydiff_get(lambda t1,t2: \
+            'Report/UnitExtendedInfo?Id='+veh_id+'&from='+t1+'.0000001Z&to='+t2+'.0000000Z',\
+                tdelta,previter)

pytrackunit/tuiter.py

@@ -0,0 +1,64 @@
+"""iterator for tucache data"""
+
+class ReqIter:
+    """iterator for web (url) requests"""
+    def __init__(self,cache,requests):
+        self.cache = cache
+        self.requests = requests
+        self.iter_started = False
+
+    def __aiter__(self):
+        if self.iter_started:
+            raise Exception("cant start tuiter more than once")
+        self.iter_started = True
+        return self
+
+    async def __anext__(self):
+        try:
+            return await self.cache.get_url(next(self.requests))
+        except StopIteration as exc:
+            raise StopAsyncIteration from exc
+
+class SqlIter:
+    """iterator for tucache data"""
+    def __init__(self, sqliter):
+        self.sqliter = sqliter
+        self.iter_started = False
+
+    def __aiter__(self):
+        if self.iter_started:
+            raise Exception("cant start tuiter more than once")
+        self.iter_started = True
+        return self
+
+    async def __anext__(self):
+        try:
+            return next(self.sqliter)
+        except StopIteration as exc:
+            raise StopAsyncIteration from exc
+
+class TuIter:
+    """iterator holding all internal iterators"""
+    def __init__(self) -> None:
+        self.iterators = []
+        self.iterator_pos = 0
+        self.iter_started = False
+
+    def add(self,_iter):
+        """addes an internal iterator to this iterators list"""
+        self.iterators.append(_iter)
+
+    def __aiter__(self):
+        if self.iter_started:
+            raise Exception("cant start tuiter more than once")
+        self.iter_started = True
+        return self
+
+    async def __anext__(self):
+        try:
+            return await self.iterators[self.iterator_pos].__anext__()
+        except StopAsyncIteration as exc:
+            self.iterator_pos += 1
+            if self.iterator_pos >= len(self.iterators):
+                raise StopAsyncIteration from exc
+            return await self.__anext__()