How to use the esper.prelude.par_for function in esper

exit()

videos = videos
cfg = cluster_config(
    num_workers=100, worker=worker_config('n1-standard-64'),
    pipelines=[gender_detection.GenderDetectionPipeline])

with make_cluster(cfg, sql_pool=2, no_delete=True) as db_wrapper:
    db = db_wrapper.db

# if True:
#     db_wrapper = ScannerWrapper.create()

    frames = pcache.get('gender_frames', lambda: par_for(frames_for_video, videos, workers=8))
    videos, frames = unzip([(v, f) for (v, f) in zip(videos, frames) if len(f) > 0])
    videos = list(videos)
    frames = list(frames)
    detect_genders(
        db,
        videos=[v.for_scannertools() for v in videos],
        db_videos=videos,
        frames=frames,
        faces=[ScannerSQLTable(Face, v, num_elements=len(f),
                               filter='query_frame.shot_boundary = false')
               for v, f in zip(videos, frames)],
        run_opts={
            'io_packet_size': 500,
            'work_packet_size': 20,
            'pipeline_instances_per_node': 16
        })

videos = list(Video.objects.filter(threeyears_dataset=True).order_by('id'))
def load_frames():
    return par_for(frames_for_video, videos, workers=8)
frames = pcache.get('emb_frames', load_frames, force=True)
videos, frames = unzip([(v, f) for (v, f) in zip(videos, frames)
                        if len(f) > 0])
videos = list(videos)
frames = list(frames)

# Export packed embeddings and IDs into single files
if False:
    def get_ids(video):
        return [f['id'] for f in Face.objects.filter(frame__video=video).order_by('frame__number', 'id').values('id')]

    all_ids = pcache.get('emb_ids', (lambda: par_for(get_ids, videos, workers=4)))

    import struct
    with open('/app/data/embs/sevenyears_ids.bin', 'wb') as f:
        for i, ids in tqdm(enumerate(all_ids)):
            path = '/app/data/embs/{:07d}.bin'.format(i)
            if os.path.isfile(path):
                f.write(b''.join([struct.pack('=Q', i) for i in ids]))

    with open('/app/data/embs/sevenyears_embs.bin', 'wb') as f:
        for i in tqdm(list(range(len(videos)))):
            path = '/app/data/embs/{:07d}.bin'.format(i)
            if os.path.isfile(path):
                byts = open(path, 'rb').read()
                if len(byts) / (4 * 128) != len(all_ids[i]):
                    print(i)
                f.write(byts)

videos = list(Video.objects.all().order_by('id'))

cfg = cluster_config(
    num_workers=100,
    worker=worker_config('n1-standard-16', gpu=1),
    pipelines=[clothing_detection.ClothingDetectionPipeline])

with make_cluster(cfg, sql_pool=2, no_delete=True) as db_wrapper:
    # if True:
    #     db_wrapper = ScannerWrapper.create()

    db = db_wrapper.db

    print('Fetching frames')
    frames = pcache.get('clothing_frames', lambda: par_for(frames_for_video, videos, workers=8))
    videos, frames = unzip([(v, f) for (v, f) in zip(videos, frames) if len(f) > 0])
    videos = list(videos)
    frames = list(frames)

    videos = videos
    frames = frames

    bbox_tables = [
        ScannerSQLTable(
            Face,
            v,
            num_elements=len(f),
            filter='MOD(query_frame.number, CAST(FLOOR(query_video.fps * 3) AS INTEGER)) = 0'
            if v.threeyears_dataset else
            'MOD(query_frame.number, CAST(CEIL(query_video.fps * 3) AS INTEGER)) = 0')
        for v, f in zip(videos, frames)

for (frame_num, frame_cloth) in zip(vid_frames, list(outp.load())):
            faces = list(
                Face.objects.filter(frame__video=video,
                                    frame__number=frame_num).order_by('id').values('id'))
            for (cloth, face) in zip(frame_cloth, face_ids[frame_num]):
                hcs.append(
                    HairColor(
                        labeler=labeler,
                        face_id=face['id'],
                        color_id=hc_names[cloth.to_dict()['Hair color 3']]))

        pcache.set(video.item_name() + '-haircolor', hcs)

    #pcache.set('haircolors', flatten(
    par_for(run, list(zip(videos, frames, clothing)), workers=16)
    #))

def load_frames():
    return par_for(frames_for_video, videos, workers=8)
frames = pcache.get('emb_frames', load_frames, force=True)

def load_embs(i):
            path = '/app/data/embs/{:07d}.bin'.format(i)
            if os.path.isfile(path):
                return

            print(i)
            flat_emb = [emb.tobytes() for frame_embs in embs[i].load() for emb in frame_embs]
            with open(path, 'wb') as f:
                f.write(b''.join(flat_emb))

        print('embs', len(embs))
        # for i in tqdm(range(len(embs))):
        #     load_embs(i)

        for l in tqdm(list(batch(list(range(len(embs))), 100))):
            par_for(load_embs, l, workers=8, progress=False)