def _threaded_download_and_preprocess_audio_and_video(
        allocated_content: list[str],
        preprocessor,
        preprocessing_config: Union[AudioPreprocessingConfig, VideoPreprocessingConfig],
        metric_obj: Optional[RequestMetrics] = None,
        return_individual_error: bool = True,
) -> list[PreprocessedContent]:
    """A thread calls this function to download audio and video for its allocated contents.

    Args:
        allocated_content: The content to be downloaded and preprocessed by this thread.
        preprocessor: The preprocessor to be used for preprocessing the content. E.g., LanguagebindModelPreprocessor
        preprocessing_config: The preprocessing configuration to be used for preprocessing the content.
        return_individual_error: If True, collect individual errors in the thread_results, otherwise, raise an error.
        metric_obj: The telemetry object to be used for measuring the time taken for each thread.

    Return:
        A list of tuples, where each tuple contains the chunk and the preprocessed content.

    Raise:
        InferenceError: If there is an error in downloading or preprocessing the content and
        return_individual_error is False.
        RuntimeError: If the number of results does not match the number of allocated content.
    """
    _id = f'media_download.{preprocessing_config.modality}.{threading.get_ident()}'
    thread_results: list[Union[InferenceErrorModel, list[Tuple[str, Tensor]]]] = []
    with metric_obj.time(f"{_id}.thread_time"):
        for url in allocated_content:
            try:
                media_downloader= StreamingMediaProcessor(
                    url = url,
                    preprocessors = preprocessor,
                    preprocessing_config = preprocessing_config,
                    enable_video_gpu_acceleration=_enable_video_gpu_acceleration()
                )
                results: list[Tuple[str, Tensor]] = media_downloader.process_media()
                thread_results.append(results)
            except InferenceError as e:
                metric_obj.increment_counter(f"{url}.InferenceError")
                if return_individual_error:
                    thread_results.append(InferenceErrorModel(error_message=str(e)))
                    continue
                else:
                    raise e
        if not len(thread_results) == len(allocated_content):
            raise RuntimeError(
                f"Thread {threading.get_ident()} had a problem when processing the content. "
                f"Expected {len(allocated_content)} results, but got {len(thread_results)} results"
            )
    return thread_results
A thread calls this function to download audio and video for its allocated contents.

    Args:
        allocated_content: The content to be downloaded and preprocessed by this thread.
        preprocessor: The preprocessor to be used for preprocessing the content. E.g., LanguagebindModelPreprocessor
        preprocessing_config: The preprocessing configuration to be used for preprocessing the content.
        return_individual_error: If True, collect individual errors in the thread_results, otherwise, raise an error.
        metric_obj: The telemetry object to be used for measuring the time taken for each thread.

    Return:
        A list of tuples, where each tuple contains the chunk and the preprocessed content.

    Raise:
        InferenceError: If there is an error in downloading or preprocessing the content and
        return_individual_error is False.
        RuntimeError: If the number of results does not match the number of allocated content.
    
src/marqo/inference/media_download_and_preprocess/media_download_and_preprocess.py