more refactoring, tests passing again

nosetests.xml

@@ -1 +1 @@
-<?xml version="1.0" encoding="UTF-8"?><testsuite name="nosetests" tests="2" errors="0" failures="0" skip="0"><testcase classname="qsim.test.test_messaging.MessagingTestCase" name="test_client" time="0.520" /><testcase classname="qsim.test.test_messaging.MessagingTestCase" name="test_moving_average_to_client" time="50.254" /></testsuite>
+<?xml version="1.0" encoding="UTF-8"?><testsuite name="nosetests" tests="2" errors="0" failures="0" skip="0"><testcase classname="qsim.test.test_messaging.MessagingTestCase" name="test_client" time="0.233" /><testcase classname="qsim.test.test_messaging.MessagingTestCase" name="test_moving_average_to_client" time="46.324" /></testsuite>

qsim/messaging.py

@@ -0,0 +1,109 @@
+import json
+import uuid
+import zmq
+
+import util as qutil
+
+class ParallelBuffer(object):
+    """ holds several queues of events by key, allows retrieval in date order or by merging"""
+    def __init__(self, key_list):
+        self.out_socket = None
+        self.sent_count = 0
+        self.received_count = 0
+        self.draining = False
+        self.data_buffer = {}
+        for key in key_list:
+            self.data_buffer[key] = []
+    
+    def __len__(self):
+        return len(self.data_buffer)        
+    
+    def append(self, key, value):
+        self.data_buffer[key].append(value)
+        self.received_count += 1
+    
+    def next(self):
+        if(not(self.is_full() or self.draining)):
+            return
+            
+        cur = None
+        earliest = None
+        for source, events in self.data_buffer.iteritems():
+            if len(events) == 0:
+                continue
+            cur = events
+            if(earliest == None) or (cur[0]['dt'] <= earliest[0]['dt']):
+                earliest = cur
+        
+        if(earliest != None):
+            return earliest.pop(0)
+        
+    def is_full(self):
+        for source, events in self.data_buffer.iteritems():
+            if (len(events) == 0):
+                return False
+        return True
+    
+    def pending_messages(self):
+        total = 0
+        for source, events in self.data_buffer.iteritems():
+            total += len(events)
+        return total       
+        
+    def drain(self):
+        self.draining = True
+        while(self.pending_messages() > 0):
+            self.send_next()
+            
+    def send_next(self):
+        if(not(self.is_full() or self.draining)):
+            return
+            
+        event = self.next()
+        if(event != None):        
+            self.out_socket.send(json.dumps(event))
+            self.sent_count += 1   
+    
+    
+class MergedParallelBuffer(ParallelBuffer):
+    
+    def __init__(self, keys):
+        ParallelBuffer.__init__(self, keys)
+        self.feed = []
+        self.data_buffer["feed"] = self.feed
+    
+    def next(self):
+        if(not(self.is_full() or self.draining)):
+            return
+        
+        result = self.feed.pop(0)
+        for source, events in self.data_buffer.iteritems():
+            if(source == "feed"):
+                continue
+            if(len(events) > 0):
+                cur = events.pop(0)
+                result[source] = cur['value']
+        return result
+        
+        
+class FeedSync(object):
+    
+    def __init__(self, feed, name):
+        self.feed = feed
+        self.id = "{name}-{id}".format(name=name, id=uuid.uuid1())
+        self.feed.register_sync(self.id)
+        #qutil.logger.info("registered {id} with feed".format(id=self.id))
+        
+    def confirm(self):
+        context = zmq.Context()
+        #synchronize with feed
+        sync_socket = context.socket(zmq.REQ)
+        sync_socket.connect(self.feed.sync_address)
+        # send a synchronization request to the feed
+        sync_socket.send(self.id)
+        # wait for synchronization reply from the feed
+        sync_socket.recv()
+        sync_socket.close()
+        context.term()
+        qutil.logger.info("sync'd feed from {id}".format(id = self.id))
+   

qsim/simulator/config.py

@@ -1,3 +1,4 @@
+"""Tools for managing configuration data for sources and transforms."""
 import json
 
 class Config(object):
@@ -6,29 +7,36 @@ class Config(object):
     def __init__(self, props):
         self.store = props
         
-    def __setitem__(self,key,value):
+    def __setitem__(self, key, value):
         self.store[key] = value
         
     def __getitem__(self, key):
         if self.store.has_key(key):
             return self.store[key]
             
-    def __getattr__(self,attrname):
+    def __getattr__(self, attrname):
         if self.store.has_key(attrname):
             return self.store[attrname]
         else:
             raise AttributeError("No attribute named {name}".format(name=attrname))
         
     def get_integer(self, name, default=0):
+        """get the named config property as an integer"""
         return self.get_value(name, default, type(1))
         
     def get_string(self, name, default=''):
+        """get the named config property as a string"""
         return self.get_value(name, default, type(''))
         
     def get_float(self, name, default=0.0):
+        """get the named config property as a float"""
         return self.get_value(name, default, type(1.0))
     
     def get_value(self, name, default, expected_type):
+        """
+            return the named config property as the expected_type.
+            if the property is missing, or is not of the right type, return default.
+        """
         if(self.store.has_key(name)):
                 val = self.store[name]
                 if isinstance(val, expected_type):
@@ -37,7 +45,9 @@ class Config(object):
             return default
                     
     def to_json(self):
+        """convert this config to a json string"""
         return json.dumps(self.store)
     
     def from_json(self, json_string):
+        """parse a json string into this config object's properties"""
         self.store = json.loads(json_string)

qsim/simulator/feed.py

@@ -1,6 +1,7 @@
 
 import qsim.simulator.sources as sources
 import qsim.util as qutil
+import qsim.messaging as qmsg
 import zmq
 import time
 import logging
@@ -9,7 +10,7 @@ import json
 class DataFeed(object):
     
     def __init__(self, config):
-        self.logger = qutil.logger
+        qutil.logger = qutil.logger
         
         self.data_address = "tcp://127.0.0.1:{port}".format(port=10101)
         self.sync_address = "tcp://127.0.0.1:{port}".format(port=10102)
@@ -27,14 +28,14 @@ class DataFeed(object):
                 ret = sources.RandomEquityTrades(info['sid'], self, name, info['count'])
                 self.data_workers[name] = ret
                 
-        self.data_buffer = qutil.ParallelBuffer(self.data_workers.keys())
+        self.data_buffer = qmsg.ParallelBuffer(self.data_workers.keys())
              
     def start_data_workers(self):
         """Start a sub-process for each datasource.""" 
         for source_id, source in self.data_workers.iteritems():
-            self.logger.info("starting {id}".format(id=source_id))
+            qutil.logger.info("starting {id}".format(id=source_id))
             source.start()
-        self.logger.info("ds processes launched")
+        qutil.logger.info("ds processes launched")
         
     def register_sync(self, sync_id):
         self.client_register[sync_id] = "UNCONFIRMED"
@@ -48,7 +49,7 @@ class DataFeed(object):
     
     def sync_clients(self):
         # Socket to receive signals
-        self.logger.info("waiting for all datasources and clients to be ready")
+        qutil.logger.info("waiting for all datasources and clients to be ready")
         self.syncservice = self.context.socket(zmq.REP)
         self.syncservice.bind(self.sync_address) 
         
@@ -56,12 +57,12 @@ class DataFeed(object):
             # wait for synchronization request
             msg = self.syncservice.recv()
             self.client_register[msg] = "CONFIRMED"
-            #self.logger.info("confirmed {id}".format(id=msg))
+            #qutil.logger.info("confirmed {id}".format(id=msg))
             # send synchronization reply
             self.syncservice.send('CONFIRMED')
         
         self.syncservice.close()
-        self.logger.info("sync'd all datasources and clients")
+        qutil.logger.info("sync'd all datasources and clients")
        
     def run(self):   
         # Prepare our context and sockets
@@ -86,7 +87,7 @@ class DataFeed(object):
         #wait for all feed subscribers
         self.sync_clients()
         
-        self.logger.info("entering feed loop on {addr}".format(addr=self.data_address))
+        qutil.logger.info("entering feed loop on {addr}".format(addr=self.data_address))
         
         while True:
             message = self.data_socket.recv()
@@ -105,7 +106,7 @@ class DataFeed(object):
         
         #send the DONE message
         self.feed_socket.send("DONE")
-        self.logger.info("received {n} messages, sent {m} messages".format(n=self.data_buffer.received_count, m=self.data_buffer.sent_count))
+        qutil.logger.info("received {n} messages, sent {m} messages".format(n=self.data_buffer.received_count, m=self.data_buffer.sent_count))
         self.data_socket.close()
         self.feed_socket.close()
         self.context.term()

qsim/simulator/sources.py

@@ -1,5 +1,5 @@
 """
-Provides data source handlers that can push messages to a qsim.simulator.DataFeed
+Provides data handlers that can push messages to a qsim.simulator.DataFeed
 """
 import datetime
 import zmq
@@ -8,6 +8,7 @@ import multiprocessing
 import random
 
 import qsim.util as qutil
+import qsim.messaging as qmsg
 
 class DataSource(object):
     """
@@ -18,40 +19,41 @@ class DataSource(object):
     def __init__(self, feed, source_id):
         self.source_id              = source_id
         self.feed                   = feed
-        self.sync                   = qutil.FeedSync(self.feed, str(source_id))
-        self.data_address           = self.feed.data_address
-        qutil.logger.info("data address is {ds}".format(ds=self.feed.data_address))        
-        self.cur_event = None
+        self.cur_event              = None
+        self.context                = None
+        self.data_socket            = None
 
     def start(self):
         """Launch the datasource in a separate process."""
-        self.proc = multiprocessing.Process(target=self.run)
-        self.proc.start()
+        proc = multiprocessing.Process(target=self.run)
+        proc.start()
         
         
     def open(self):    
         """create zmq context and socket"""
-        qutil.logger.info("starting data source:{source_id} on {addr}"
-                            .format(source_id=self.source_id, addr=self.feed.data_address))
+        qutil.logger.info(
+            "starting data source:{source_id} on {addr}"
+                .format(source_id=self.source_id, addr=self.feed.data_address))
         
         self.context = zmq.Context()
         
         #create the data sink. Based on http://zguide.zeromq.org/py:tasksink2 
         self.data_socket = self.context.socket(zmq.PUSH)
-        self.data_socket.connect(self.data_address)
+        self.data_socket.connect(self.feed.data_address)
         
         #signal we are ready
-        self.sync.confirm()
+        sync = qmsg.FeedSync(self.feed, str(self.source_id))
+        sync.confirm()
     
     def run(self):  
         """Fully execute this datasource."""  
-        try:    
-            self.open()
-            self.send_all()
-            self.close()    
-        except Exception as err:
-            qutil.logger.exception("Unexpected failure running datasource - {name}."
-                                    .format(name=self.source_id))
+        self.open()
+        self.send_all()
+        self.close()    
+    
+    def send_all(self):
+        """Subclasses must implement this method."""
+        raise NotImplementedError()
             
     def send(self, event):
         """

qsim/test/client.py

@@ -5,40 +5,40 @@ import logging
 import json
 
 import qsim.util as qutil
+import qsim.messaging as qmsg
 
 class TestClient(object):
     
     def __init__(self,feed, address, bind=False):
-        self.logger         = qutil.logger
         self.feed           = feed
         self.address        = address
-        self.sync           = qutil.FeedSync(feed, "testclient")
+        self.sync           = qmsg.FeedSync(feed, "testclient")
         self.bind           = bind
         self.received_count = 0
         
     def run(self):
         
-        self.logger.info("running the client")
+        qutil.logger.info("running the client")
         self.context = zmq.Context()
         
         self.data_feed = self.context.socket(zmq.PULL)
 
         if(self.bind):
-            self.logger.info("binding to {address}".format(address=self.address))
+            qutil.logger.info("binding to {address}".format(address=self.address))
             self.data_feed.bind(self.address)
         else:
-            self.logger.info("connecting to {address}".format(address=self.address))
+            qutil.logger.info("connecting to {address}".format(address=self.address))
             self.data_feed.connect(self.address)
         
         self.sync.confirm()
         
-        self.logger.info("Starting the client loop")
+        qutil.logger.info("Starting the client loop")
 
         prev_dt = None
         while True:
             msg = self.data_feed.recv()
             if(msg == "DONE"):
-                self.logger.info("DONE!")
+                qutil.logger.info("DONE!")
                 break
             self.received_count += 1
             event = json.loads(msg)
@@ -48,8 +48,8 @@ class TestClient(object):
             
             prev_dt = event['dt']
             if(self.received_count % 100 == 0):
-                self.logger.info("received {n} messages".format(n=self.received_count))
+                qutil.logger.info("received {n} messages".format(n=self.received_count))
             
-        self.logger.info("received {n} messages".format(n=self.received_count))
+        qutil.logger.info("received {n} messages".format(n=self.received_count))
         self.data_feed.close()
         self.context.term()

qsim/transforms/core.py

@@ -19,6 +19,7 @@ all the transforms of that event into a single new message.
 """
 import zmq
 import json
+import qsim.messaging as qmsg
 import qsim.util as qutil
 import qsim.simulator.config as config
 
@@ -42,7 +43,7 @@ class BaseTransform(object):
         self.config             = config.Config(config_dict)
         self.state              = {}
         self.state['name']      = self.config.name
-        self.sync               = qutil.FeedSync(feed, self.state['name'])
+        self.sync               = qmsg.FeedSync(feed, self.state['name'])
         self.received_count     = 0
         self.sent_count         = 0 
         self.context            = None

qsim/transforms/merge.py

@@ -5,6 +5,7 @@ import zmq
 import technical as ta
 from core import BaseTransform
 import qsim.util as qutil
+import qsim.messaging as qmsg
         
 class MergedTransformsFeed(BaseTransform):
     """ Merge data feed and array of transform feeds into a single result vector.
@@ -40,7 +41,7 @@ class MergedTransformsFeed(BaseTransform):
         
         keys = copy.copy(self.transforms.keys())
         keys.append("feed") #for the raw feed
-        self.data_buffer = qutil.MergedParallelBuffer(keys) 
+        self.data_buffer = qmsg.MergedParallelBuffer(keys) 
             
         self.buffers = {}
         for name, transform in self.transforms.iteritems():

qsim/util.py

@@ -1,13 +1,12 @@
 """
-Small classes to assist with db access, timezone calculations, and so on.
+Small classes to assist with timezone calculations, logger configuration,
+and other common operations.
 """
 
 import datetime
 import pytz
-import json
 import logging
-import uuid
-import zmq
+
 
 logger = logging.getLogger('QSimLogger')
             
@@ -26,112 +25,6 @@ def format_date(dt):
     dt_str = dt.strftime('%Y/%m/%d-%H:%M:%S') + "." + str(dt.microsecond / 1000)
     return dt_str
     
-    
-class ParallelBuffer(object):
-    """ holds several queues of events by key, allows retrieval in date order or by merging"""
-    def __init__(self, key_list):
-        self.out_socket = None
-        self.sent_count = 0
-        self.received_count = 0
-        self.draining = False
-        self.data_buffer = {}
-        for key in key_list:
-            self.data_buffer[key] = []
-    
-    def __len__(self):
-        return len(self.data_buffer)        
-    
-    def append(self, key, value):
-        self.data_buffer[key].append(value)
-        self.received_count += 1
-    
-    def next(self):
-        if(not(self.is_full() or self.draining)):
-            return
-            
-        cur = None
-        earliest = None
-        for source, events in self.data_buffer.iteritems():
-            if len(events) == 0:
-                continue
-            cur = events
-            if(earliest == None) or (cur[0]['dt'] <= earliest[0]['dt']):
-                earliest = cur
-        
-        if(earliest != None):
-            return earliest.pop(0)
-        
-    def is_full(self):
-        for source, events in self.data_buffer.iteritems():
-            if (len(events) == 0):
-                return False
-        return True
-    
-    def pending_messages(self):
-        total = 0
-        for source, events in self.data_buffer.iteritems():
-            total += len(events)
-        return total       
-        
-    def drain(self):
-        self.draining = True
-        while(self.pending_messages() > 0):
-            self.send_next()
-            
-    def send_next(self):
-        if(not(self.is_full() or self.draining)):
-            return
-            
-        event = self.next()
-        if(event != None):        
-            self.out_socket.send(json.dumps(event))
-            self.sent_count += 1   
-    
-    
-class MergedParallelBuffer(ParallelBuffer):
-    
-    def __init__(self, keys):
-        ParallelBuffer.__init__(self, keys)
-        self.feed = []
-        self.data_buffer["feed"] = self.feed
-    
-    def next(self):
-        if(not(self.is_full() or self.draining)):
-            return
-        
-        result = self.feed.pop(0)
-        for source, events in self.data_buffer.iteritems():
-            if(source == "feed"):
-                continue
-            if(len(events) > 0):
-                cur = events.pop(0)
-                result[source] = cur['value']
-        return result
-        
-        
-class FeedSync(object):
-    
-    def __init__(self, feed, name):
-        self.feed = feed
-        self.id = "{name}-{id}".format(name=name, id=uuid.uuid1())
-        self.feed.register_sync(self.id)
-        self.logger = logger
-        #self.logger.info("registered {id} with feed".format(id=self.id))
-        
-    def confirm(self):
-        context = zmq.Context()
-        #synchronize with feed
-        sync_socket = context.socket(zmq.REQ)
-        sync_socket.connect(self.feed.sync_address)
-        # send a synchronization request to the feed
-        sync_socket.send(self.id)
-        # wait for synchronization reply from the feed
-        sync_socket.recv()
-        sync_socket.close()
-        context.term()
-        self.logger.info("sync'd feed from {id}".format(id = self.id))
-        
-
 def configure_logging(loglevel=logging.DEBUG):  
     logger.setLevel(loglevel)
     handler = logging.handlers.RotatingFileHandler("/tmp/{lfn}.log".format(lfn="qsim-log"), maxBytes=10*1024*1024, backupCount=5)