How to use the esper.kube.make_cluster function in esper

byts = open(path, 'rb').read()
                if len(byts) / (4 * 128) != len(all_ids[i]):
                    print(i)
                f.write(byts)
                f.flush()

    print('done')
    exit()

if __name__ == "__main__":

    cfg = cluster_config(
        num_workers=0, worker=worker_config('n1-standard-64'),
        pipelines=[face_embedding.FaceEmbeddingPipeline])

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

        db = db_wrapper.db

        embs = embed_faces(
            db,
            videos=[v.for_scannertools() for v in 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': 8

ScannerJobConfig(io_packet_size=1000, work_packet_size=20, pipelines_per_worker=4),
            ScannerJobConfig(io_packet_size=1000, work_packet_size=20, pipelines_per_worker=8),
            ScannerJobConfig(io_packet_size=1000, work_packet_size=20, pipelines_per_worker=16)
        ])]
        bench('gender', {'videos': videos, 'frames': [frames_for_video(v) for v in videos]},
              run_pipeline, configs, no_delete=True, force=True)


    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')

#         worker=worker_config('n1-standard-16'))
#     with make_cluster(cfg, no_delete=True) as db_wrapper:

# videos = videos
#videos = list(Video.objects.filter(id__gte=91250, id__lte=91350))
# videos = [Video.objects.get(id=63970)]
videos = videos

with Timer('Shot boundaries'):
    cfg = cluster_config(
        num_workers=60,
        worker=worker_config('n1-highmem-16'),
        workers_per_node=2,
        num_load_workers=1,
        num_save_workers=2)
    with make_cluster(cfg, no_delete=True) as db_wrapper:

    # from esper.scannerutil import ScannerWrapper
    # if True:
    #     db_wrapper = ScannerWrapper.create()

        db = db_wrapper.db

        job_config = ScannerJobConfig(io_packet_size=10000, work_packet_size=400, batch=400)
        hists = run_pipeline(db, videos, batch=job_config.batch, run_opts={
            'io_packet_size': job_config.io_packet_size,
            'work_packet_size': job_config.work_packet_size,
        })
        print('hists', len(hists))

        hists, videos = unzip([(h, v) for (h, v) in zip(hists, videos) if v.num_frames < 800000])
        boundaries = compute_shot_boundaries(

price_per_hour = cluster_config.price(no_master=True)
            price_per_video = (time / 3600.0) * price_per_hour / float(sample_size)
            return price_per_video, metrics
        else:
            return None

    results = []

    for (cluster_config, job_configs) in configs:

        # Only bring up the cluster if there exists a job config that hasn't been computed
        if not force and all([pcache.has(run_name(cluster_config, job_config)) for job_config in job_configs]):
            results.append([pcache.get(run_name(cluster_config, job_config)) for job_config in job_configs])

        else:
            with make_cluster(cluster_config, no_delete=no_delete) as db_wrapper:
                log.info('Cluster config: {}'.format(cluster_config))

                def try_config(job_config):
                    log.info('Job config: {}'.format(job_config))
                    try:
                        return test_config(
                            args, db_wrapper, cluster_config, job_config)
                    except TestFailure as e:
                        print(e)
                        return (str(e), None)
                    except Exception as e:
                        traceback.print_exc()
                        return (traceback.format_exc(), None)

                def try_config_cached(job_config):
                    return pcache.get(run_name(cluster_config, job_config), force=force, fn=lambda: try_config(job_config))

with Timer('run'):

    print('Getting frames')
    def load_frames():
        return [[f['number'] for f in Frame.objects.filter(video=v, shot_boundary=False).values('number').order_by('number')]
                for v in tqdm(videos)]
    frames = pcache.get('face_frames', load_frames)

    cfg = cluster_config(
        num_workers=100,
        worker=worker_config('n1-standard-64'),
        num_load_workers=2,
        num_save_workers=2)
    with make_cluster(cfg, sql_pool=4, no_delete=True) as db_wrapper:

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

        db = db_wrapper.db

        print('Starting detection')
        detect_faces(
            db,
            videos=[v.for_scannertools() for v in videos],
            db_videos=videos,
            frames=frames,
            frame_ids=[ScannerSQLTable(Frame, v, num_elements=len(f),
                                       filter='query_frame.shot_boundary = false')
                       for v, f in zip(videos, frames)],
            run_opts={

self._db.ops.PrepareClothingBbox(
                frame=self._sources['frame_sampled'].op, bboxes=bboxes)
        }


detect_clothing_bboxes = ClothingBboxesPipeline.make_runner()
detect_clothing = ClothingDetectionPipeline.make_runner()

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(