Pileus Git - ~andy/csm213a-hw/blob - vis/logger.py

   1 import sys
   2
   3 from datetime import datetime
   4
   5 from xively import XivelyAPIClient
   6 from xively import Datastream
   7 from xively import Datapoint
   8
   9 if sys.version_info<(3,0,0):
  10         from Queue   import Queue
  11         from thread  import get_ident, start_new_thread
  12 else:
  13         from queue   import Queue
  14         from _thread import get_ident, start_new_thread
  15
  16 class Logger:
  17         # Constructor
  18         def __init__(self, config):
  19                 self.config = config
  20                 self.api    = None
  21                 self.feed   = None
  22                 self.last   = 0
  23                 self.limit  = 500
  24                 self.states = []
  25                 self.queue  = Queue()
  26                 self.thread = None
  27                 self.error  = ''
  28
  29         # Methods
  30         def connect(self):
  31                 if not self.running():
  32                         self.last = datetime.utcnow()
  33                         start_new_thread(self.task, ())
  34
  35         def disconnect(self):
  36                 if self.running():
  37                         self.thread = None
  38                         self.queue.put([])
  39
  40         def running(self):
  41                 return bool(self.thread)
  42
  43         def append(self, state):
  44                 if not self.running():
  45                         return
  46                 self.states.append(state)
  47                 delta = datetime.utcnow() - self.last
  48                 if delta.total_seconds() > self.config.maxrate:
  49                         self.queue.put(self.states)
  50                         self.last   = datetime.utcnow()
  51                         self.states = []
  52
  53         # Private methods
  54         def split(self, items, limit):
  55                 split   = {}
  56                 count   = 0
  57                 sensors = {'acc': ['x','y','z'],
  58                            'mag': ['x','y','z'],
  59                            'tch': ['p','d'],
  60                            'lgt': [''],
  61                            'a2d': ['0','1','2','3','4','5']}
  62                 for state in items:
  63                         for sns in sensors:
  64                                 data = getattr(state, sns)
  65                                 for i in range(len(data)):
  66                                         if data[i] == None:
  67                                                 continue
  68                                         if count >= limit:
  69                                                 print('upload: cropping upload')
  70                                                 return split
  71                                         chan  = sns + sensors[sns][i]
  72                                         if not split.has_key(chan):
  73                                                 split[chan] = []
  74                                         point = Datapoint(state.time, data[i])
  75                                         split[chan].append(point)
  76                                         count += 1
  77                 return split
  78
  79         def task(self):
  80                 print('thread - start')
  81                 self.thread = get_ident()
  82
  83                 # Setup connection
  84                 api  = XivelyAPIClient(self.config.apikey)
  85                 feed = api.feeds.get(self.config.feedid)
  86
  87                 # Process requests from main
  88                 while self.running():
  89                         # Pull data from queue (blocking)
  90                         states = self.queue.get()
  91                         print('thread - got %d states' % len(states))
  92
  93                         # Make sure it's valid
  94                         if len(states) == 0:
  95                                 continue
  96
  97                         # Split data and limit to 500 points per upload
  98                         upload = self.split(states, self.limit)
  99
 100                         # Create datastreams
 101                         feed.datastreams = \
 102                                 [ Datastream(id=chan, datapoints=upload[chan])
 103                                   for chan in upload.keys() ]
 104
 105                         # Upload it
 106                         try:
 107                                 print('thread - upload start')
 108                                 feed.update()
 109                                 print('thread - upload done')
 110                         except Exception as ex:
 111                                 self.error = str(ex)
 112                                 break
 113
 114                 # Notify main that we're done
 115                 print('thread - exit')
 116                 self.thread = None