def video_embedding(config):
    """
    Define pipeline
    """
    if config.start_time and config.end_time and config.start_time >= config.end_time:
        raise ValueError('Start time should be earlier than end time in pipeline config.')
    allow_triton, emb_op = _get_embedding_op(config)
    decode_op = ops.video_decode.ffmpeg(
        start_time=config.start_time, end_time=config.end_time, sample_type='time_step_sample', args={'time_step': 1}
    )
    milvus_op = ops.ann_insert.milvus_client(host=config.milvus_host, port=config.milvus_port, collection_name=config.collection)
    if config.hbase_table:
        kv_op = ops.kvstorage.insert_hbase(host=config.hbase_host, port=config.hbase_port, table=config.hbase_table)
    if config.leveldb_path:
        kv_op = ops.kvstorage.insert_leveldb(path=config.leveldb_path)
    norm_op = None if config.model == 'isc' else ops.towhee.np_normalize()

    return _video_embedding(decode_op, emb_op, milvus_op, kv_op, norm_op, allow_triton, config.device)
Define pipeline
towhee/pipelines/video_embedding.py