def video_copy_detection(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_search.milvus_client(
        host=config.milvus_host,
        port=config.milvus_port,
        collection_name=config.collection,
        output_fields=['path'],
        metric_type='IP',
        limit = config.milvus_search_limit
    )
    if config.hbase_table:
        kv_op = ops.kvstorage.search_hbase(host=config.hbase_host, port=config.hbase_port, table=config.hbase_table, is_ndarray=True)
    if config.leveldb_path:
        kv_op = ops.kvstorage.from_leveldb(path=config.leveldb_path, is_ndarray=True)
    select_op = ops.video_copy_detection.select_video(top_k=config.top_k, reduce_function='sum', reverse=True)
    tn_op = ops.video_copy_detection.temporal_network(min_length=config.min_similar_length)
    norm_op = None if config.model == 'isc' else ops.towhee.np_normalize()
    filter_op = DuplicateFilter(config.threshold) if config.threshold else None


    return _video_copy_detection(
        decode_op, emb_op, milvus_op, kv_op, select_op, tn_op, norm_op, filter_op, allow_triton, config.device
    )
Define pipeline
towhee/pipelines/video_copy_detection.py