def get_event_sensors(rows, row_columns, start_time, stop_time, max_samples=None):
    """
    Returns:
        (event_sensors, sensor_names)
        event_sensors: Dict of type with value as a list of sensors
        sensor_names: Dict of type with value as name
    """
    event_sensors = {}  # [type] = values
    sensor_types = {}  # [name] = type
    for row in rows:
        print(row_columns[row].keys())
        try:
            sensor_types.update(msgpack.loads(row_columns[row]['meta:sensor_types']))
        except KeyError:
            continue
        for sensor_name, samples in msgpack.loads(row_columns[row]['meta:sensor_samples']).items():
            sensor_type = sensor_types[sensor_name]
            for s in samples:
                if not (start_time <= s[1] <= stop_time):
                    continue
                event_sensors.setdefault(sensor_type, []).append(s)
    sensor_names = {v: k for k, v in sensor_types.items()}  # [type] = name
    for t in event_sensors:
        event_sensors[t].sort(key=lambda x: x[1])
    if max_samples is not None:
        for t in event_sensors:
            if len(event_sensors[t]) < max_samples:
                continue
            skip = len(event_sensors[t]) / (max_samples - 1)
            event_sensors[t] = event_sensors[t][:-1:skip] + [event_sensors[t][-1]]
    return event_sensors, sensor_names

    def get(self, _type, key, key_path=None):
        k = self._key(_type, key)
        with self.index_lock:
            if not k in self.index:
                return None

        with self.locks[k]:
            if not key_path:
                # TODO: Should we do an extra copy?
                return msgpack.loads(self.caches[k])
            else:
                path = key_path.split('/')
                # Hard limit to avoid nasty lengthy requests
                if len(path) > 7 or len(path) < 1:
                    return None
                else:
                    cached = msgpack.loads(self.caches[k])
                    data = dict()
                    data['type'] = cached['type'] + '/' + key_path
                    data['key'] = cached['key']
                    data['timestamp'] = cached['data'].get('timestamp', None)

                    d = cached['data']
                    try:
                        d = reduce(lambda di, key: di.get(key, None), path, d)
                    except AttributeError:
                        d = None

                    if d:
                        data['data'] = d
                        return copy(data)
                    else:
                        return None

def my_loads(header, frames):
    obj = MyObject(**msgpack.loads(frames[0], raw=False))

    # to provide something to test against, lets just attach the context to
    # the object itself
    obj.context = msgpack.loads(frames[1], raw=False)
    return obj

    def load(data):
        r = Request()
        try:
            r.__dict__ = msgpack.loads(data, encoding="utf-8")
        except (UnicodeDecodeError):
            r.__dict__ = msgpack.loads(data, encoding="ISO-8859-1")

        return r

def assert_packed_msg_equal(b1, b2):
    """Assert that two packed msgpack messages are equal."""
    msg1 = msgpack.loads(b1, encoding='utf8')
    msg2 = msgpack.loads(b2, encoding='utf8')
    assert sorted(msg1.keys()) == sorted(msg2.keys())
    for (k1, v1), (k2, v2) in zip(sorted(msg1.items()), sorted(msg2.items())):
        assert k1 == k2
        assert v1 == v2

def yield_api():
    try:
        chunk = yield s.enqueue('chunkMe', num)
        msgpack.loads(chunk)
        while True:
            ch = yield
            msgpack.loads(ch)
    except ChokeEvent:
        IOLoop.current().stop()

 def get_notify_data(self, notify_id):
     try:
         ret = self.redis.hget(self.__hash_name__, notify_id)
         if ret is not None:
             _tmp = loads(ret) 
             log.debug('ct=%d|difftm=%d', _tmp['ct'], int(time.time()*1000)-_tmp['ct'])
             return loads(ret)['data']
     except Exception, e:
         log.warn(traceback.format_exc())

def _unpack_msgpack_snappy(str):
    if str.startswith(b'S'):
        tmp = snappy.uncompress(str[1:])
        # print "SNAPPY: ", len(str), len(tmp)
        obj = msgpack.loads(tmp, encoding='utf-8')
    elif str.startswith(b'\0'):
        obj = msgpack.loads(str[1:], encoding='utf-8')
    else:
        return None
    
    return obj

 def loads(self, msg):
     '''
     Run the correct loads serialization format
     '''
     if self.serial == 'msgpack':
         return msgpack.loads(msg, use_list=True)
     elif self.serial == 'pickle':
         try:
             return pickle.loads(msg)
         except Exception:
             return msgpack.loads(msg, use_list=True)

def load_market_data():
    try:
        fp_bm = get_datafile('benchmark.msgpack', "rb")
    except IOError:
        print """
data msgpacks aren't distribute with source.
Fetching data from Yahoo Finance.
""".strip()
        dump_benchmarks()
        fp_bm = get_datafile('benchmark.msgpack', "rb")

    bm_list = msgpack.loads(fp_bm.read())
    bm_returns = []
    for packed_date, returns in bm_list:
        event_dt = tuple_to_date(packed_date)
        #event_dt = event_dt.replace(
        #    hour=0,
        #    minute=0,
        #    second=0,
        #    tzinfo=pytz.utc
        #)

        daily_return = risk.DailyReturn(date=event_dt, returns=returns)
        bm_returns.append(daily_return)

    fp_bm.close()

    bm_returns = sorted(bm_returns, key=attrgetter('date'))

    try:
        fp_tr = get_datafile('treasury_curves.msgpack', "rb")
    except IOError:
        print """
data msgpacks aren't distribute with source.
Fetching data from data.treasury.gov
""".strip()
        dump_treasury_curves()
        fp_tr = get_datafile('treasury_curves.msgpack', "rb")

    tr_list = msgpack.loads(fp_tr.read())
    tr_curves = {}
    for packed_date, curve in tr_list:
        tr_dt = tuple_to_date(packed_date)
        #tr_dt = tr_dt.replace(hour=0, minute=0, second=0, tzinfo=pytz.utc)
        tr_curves[tr_dt] = curve

    fp_tr.close()

    tr_curves = OrderedDict(sorted(
                            ((dt, c) for dt, c in tr_curves.iteritems()),
                            key=lambda t: t[0]))

    return bm_returns, tr_curves

 def from_token(cls, token):
     if PY3:
         # The argument raw=False is only available on new versions of
         # msgpack, and only really needed on Python 3. Gate it behind
         # a PY3 check to avoid causing issues on Debian-packaged versions.
         decoded = msgpack.loads(decode_base64(token), raw=False)
     else:
         decoded = msgpack.loads(decode_base64(token))
     return RoomListNextBatch(**{
         cls.REVERSE_KEY_DICT[key]: val
         for key, val in decoded.items()
     })

    def loads(cls, data_str, msg_pack=False):
        if msg_pack:
            if isinstance(data_str, bytes):
                data = msgpack.loads(data_str, encoding='utf-8')
            else:
                data = msgpack.loads(data_str)
        else:
            data = json.loads(data_str)

        if isinstance(data, (list, tuple,)):
            return [cls.restore(d) for d in data]
        return cls.restore(data)

 def test_compare(self):
     for x in glob.glob("picarus_takeout_models/test_models/picarus-*.msgpack.gz"):
         print(x)
         m0 = PicarusModel(x)
         m1 = PicarusCommandModel(x)
         for y in glob.glob("picarus_takeout_models/test_images/*"):
             outm0 = m0.process_binary(y)
             outm1 = m1.process_binary(y)
             if outm0 != outm1:
                 print(msgpack.loads(outm0))
                 print(msgpack.loads(outm1))
             self.assertEqual(outm0, outm1)

def _unpack(str) :

    if str[0] == 'S':
        tmp = snappy.uncompress(str[1:])
        obj = msgpack.loads(tmp)
    elif str[0] == '\0':
        obj = msgpack.loads(str[1:])
    else:
        return None

    #print "UNPACK", obj
    return obj

 def fetchAll():
     chunk = yield service.enqueue('chunkMe', str(sys.argv[1]))
     chunk = msgpack.loads(chunk)
     size = len(chunk)
     counter = 0
     while True:
         ch = yield
         chunk = msgpack.loads(ch)
         size += len(chunk)
         counter += 1
         print(counter, len(chunk), size)
         if chunk == 'Done':
             break

 def _run(self, picarus_model_class):
     results = {}
     model_path = "picarus_takeout_models/test_models/"
     image_path = "picarus_takeout_models/test_images/"
     for x in glob.glob(model_path + "picarus-*.msgpack.gz"):
         model_results = {}
         model = picarus_model_class(x)
         for y in glob.glob(image_path + "*"):
             model_results[os.path.basename(y)] = base64.b64encode(model.process_binary(y))
         results[os.path.basename(x)] = model_results
     json.dump(results, gzip.GzipFile("test_model_outputs-%s.js.gz" % (picarus_model_class.__name__,), "w"))
     prev_results = json.load(gzip.GzipFile("picarus_takeout_models/test_models/test_model_outputs.js.gz"))
     num_checked = 0
     failed_models = []
     failed_images = {}
     for x in set(results).intersection(set(prev_results)):
         for y in set(results[x]).intersection(set(prev_results[x])):
             num_checked += 1
             if results[x][y] == prev_results[x][y]:
                 continue
             try:
                 a = msgpack.loads(base64.b64decode(results[x][y]))
             except:
                 raise ValueError("Cant test for partial equality because not msgpack encoded")
             b = msgpack.loads(base64.b64decode(prev_results[x][y]))
             if not self.almostEqualAny(a, b):
                 if 1:
                     print("Current(b64msgpack)--------")
                     print(results[x][y])
                     print("Previous(b64msgpack)-------")
                     print(prev_results[x][y])
                     print("Current--------")
                     print(msgpack.loads(base64.b64decode(results[x][y])))
                     print("Previous-------")
                     print(msgpack.loads(base64.b64decode(prev_results[x][y])))
                 try:
                     failed_images[y] += 1
                 except KeyError:
                     failed_images[y] = 1
                 failed_models.append(model_path + x)
                 print(
                     "Process Failed[%s][%s][%s][%s]"
                     % (x, y, hashlib.sha1(results[x][y]).hexdigest(), hashlib.sha1(prev_results[x][y]).hexdigest())
                 )
     print("Number of models * images checked[%d][%r]" % (num_checked, picarus_model_class))
     blame_components(failed_models)
     print(failed_images)
     self.assertEqual(len(failed_models), 0)

    def from_frame(cls, frame):
        """
        Construct a ``Message`` from a raw binary frame.

        :param frame: The binary frame.

        :returns: A constructed ``Message`` instance.
        """

        # Load the data
        data = msgpack.loads(frame)

        # A PDU must be a msgpack-encoded dict
        if not isinstance(data, dict):
            raise ValueError('invalid PDU')

        # It must always have a __version__ and a type
        try:
            version = data['__version__']
            msg_type = data['msg_type']
        except KeyError as e:
            raise ValueError("missing required PDU field %s" % e)

        # Construct a message; we pass the frame in to prime the frame
        # cache
        return cls(msg_type, __version__=version, __frame__=frame, **dict(
            (k, v) for k, v in data.items()
            if k not in ('__version__', 'msg_type')))

 def test_set(self):
     "success type (+OK)"
     self.query("DEL/hello")
     f = self.query("SET/hello/world.msg")
     self.assertTrue(f.headers.getheader("Content-Type") == "application/x-msgpack")
     obj = msgpack.loads(f.read())
     self.assertTrue(obj == {"SET": (True, "OK")})

    def parse(self):
        '''
        Parses body. Assumes already unpacked.
        Results in updated .data
        '''
        bk = self.packet.data['bk']

        if bk not in raeting.BODY_KIND_NAMES:
            self.packet.data['bk']= raeting.bodyKinds.unknown
            emsg = "Unrecognizable packet body."
            raise raeting.PacketError(emsg)

        self.data = odict()

        if bk == raeting.bodyKinds.json:
            if self.packed:
                kit = json.loads(self.packed, object_pairs_hook=odict)
                if not isinstance(kit, Mapping):
                    emsg = "Packet body not a mapping."
                    raise raeting.PacketError(emsg)
                self.data = kit
        elif bk == raeting.bodyKinds.msgpack:
            if self.packed:
                if not msgpack:
                    emsg = "Msgpack not installed."
                    raise raeting.PacketError(emsg)
                kit = msgpack.loads(self.packed, object_pairs_hook=odict)
                if not isinstance(kit, Mapping):
                    emsg = "Packet body not a mapping."
                    raise raeting.PacketError(emsg)
                self.data = kit
        elif bk == raeting.bodyKinds.raw:
            self.data = self.packed # return as string
        elif bk == raeting.bodyKinds.nada:
            pass

    def downloadBatch(self):
        self._send()        
        jids=self._downloadbatch.keys()
        self.blobstor._cmdchannel.send_multipart([msgpack.dumps([[0,"getresults",{},jids]]),"S",str(60),self.blobstor.sessionkey])
        res= self.blobstor._cmdchannel.recv_multipart()
       
        for item in res:
            if item=="":
                continue
            else:                
                jid,rcode,result=msgpack.loads(item)
                if rcode==0:
                    jid,key,dest,link,repoid,chmod,chownuid,chowngid=self._downloadbatch[jid]
                    key2=result[0]
                    if key2<>key:
                        raise RuntimeError("Keys need to be the same")
                    blob=result[2]
                    serialization=result[1]
                    
                    self._downloadFilePhase2(blob,dest,key,chmod,chownuid,chowngid,link,serialization)
                else:
                    ##TODO
                    pass

        self._downloadbatchSize=0
        self._downloadbatch={}