def scan_trace()

in src/rosbridge/timeseq.py [0:0]

• 42 lines of code
• 5 McCabe index (conditional complexity)

def scan_trace(dir, channels, sample_cb):
    manifest_fn = path.join(dir, 'manifest.json')
    with open(manifest_fn, 'rb') as f:
        manifest = ujson.load(f)
    if 0: print manifest

    begin_ts = manifest['beginTs']
    end_ts = manifest['endTs']
    print 'range', begin_ts, end_ts

    timeseq_infos = manifest['timeseqInfos']
    cur_samples = {}
    # There's some thought to saving memory here. Only load a chunk (of all timeseqs) at a time
    for chunk in get_chunks(begin_ts, end_ts):
        all_chunk_data = {}
        all_chunk_indexes = {}
        for tsi in timeseq_infos:
            ts_name = tsi['name'];
            if ('*' not in channels) and (ts_name not in channels): continue
            chunk_fn = path.join(dir, 'chunk_%s_%d.json.gz' % (ts_name, chunk))
            try:
                with gzip.open(chunk_fn, 'rb') as f:
                    print 'Reading', chunk_fn, '...'
                    all_chunk_data[ts_name] = ujson.load(f)
                    all_chunk_indexes[ts_name] = 0
            except:
                print chunk_fn, 'not found'
                if ts_name in all_chunk_data:
                    del all_chunk_data[ts_name]
                    del all_chunk_indexes[ts_name]
        while True:
            # Find the next item to change (smallest timestamp)
            cur_ts = end_ts + 1
            for ts_name in all_chunk_data:
                if all_chunk_indexes[ts_name] < len(all_chunk_data[ts_name]['times']):
                    ts1 = all_chunk_data[ts_name]['times'][all_chunk_indexes[ts_name]]
                    cur_ts = min(cur_ts, ts1)
            if cur_ts > end_ts: # Didn't find anything
                break
            # Now update cur_samples with all samples with matching timestamps
            for ts_name in all_chunk_data:
                if all_chunk_indexes[ts_name] < len(all_chunk_data[ts_name]['times']):
                    ts1 = all_chunk_data[ts_name]['times'][all_chunk_indexes[ts_name]]
                    if ts1 == cur_ts:
                        cur_samples[ts_name] = all_chunk_data[ts_name]['samples'][all_chunk_indexes[ts_name]]
                        all_chunk_indexes[ts_name] += 1
            # We copy so that customers can keep a copy that works after we mutate cur_samples again
            # scan_trace_deltat does this
            sample_cb(cur_ts, copy.copy(cur_samples))