





"""simple docstring"""




import multiprocessing
import os
from typing import BinaryIO, Optional, Union

import fsspec

from .. import Dataset, Features, NamedSplit, config
from ..formatting import query_table
from ..packaged_modules.json.json import Json
from ..utils import logging
from ..utils.typing import NestedDataStructureLike, PathLike
from .abc import AbstractDatasetReader
class     UpperCamelCase__(			__A ):





						def __init__(       self		,__UpperCAmelCase		,__UpperCAmelCase = None		,__UpperCAmelCase = None		,__UpperCAmelCase = None		,__UpperCAmelCase = False		,__UpperCAmelCase = False		,__UpperCAmelCase = None		,__UpperCAmelCase = None		,**__UpperCAmelCase		,)					->				Tuple:
													super().__init__(
													    __UpperCAmelCase		,split=__UpperCAmelCase		,features=__UpperCAmelCase		,cache_dir=__UpperCAmelCase		,keep_in_memory=__UpperCAmelCase		,streaming=__UpperCAmelCase		,num_proc=__UpperCAmelCase		,**__UpperCAmelCase		,)
													A__	     =							field
													A__	     =							path_or_paths if isinstance(__UpperCAmelCase		,__UpperCAmelCase     ) else {self.split: path_or_paths}
													A__	     =							Json(
													    cache_dir=__UpperCAmelCase		,data_files=__UpperCAmelCase		,features=__UpperCAmelCase		,field=__UpperCAmelCase		,**__UpperCAmelCase		,)





						def 				snake_case__	(       self     )					->				Optional[int]:
													# Build iterable dataset
													if self.streaming:
																				A__	     =							self.builder.as_streaming_dataset(split=self.split     )
													# Build regular (map-style) dataset
													else:
																				A__	     =							None
																				A__	     =							None
																				A__	     =							None
																				A__	     =							None

																				self.builder.download_and_prepare(
																				    download_config=__UpperCAmelCase		,download_mode=__UpperCAmelCase		,verification_mode=__UpperCAmelCase		,base_path=__UpperCAmelCase		,num_proc=self.num_proc		,)
																				A__	     =							self.builder.as_dataset(
																				    split=self.split		,verification_mode=__UpperCAmelCase		,in_memory=self.keep_in_memory     )
													return dataset






class     UpperCamelCase__:





						def __init__(       self		,__UpperCAmelCase		,__UpperCAmelCase		,__UpperCAmelCase = None		,__UpperCAmelCase = None		,**__UpperCAmelCase		,)					->				Dict:
													if num_proc is not None and num_proc <= 0:
																				raise ValueError(f'''num_proc {num_proc} must be an integer > 0.'''     )

													A__	     =							dataset
													A__	     =							path_or_buf
													A__	     =							batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
													A__	     =							num_proc
													A__	     =							'utf-8'
													A__	     =							to_json_kwargs





						def 				snake_case__	(       self     )					->				int:
													A__	     =							self.to_json_kwargs.pop('path_or_buf'		,__UpperCAmelCase     )
													A__	     =							self.to_json_kwargs.pop('orient'		,'records'     )
													A__	     =							self.to_json_kwargs.pop('lines'		,True if orient == 'records' else False     )
													A__	     =							self.to_json_kwargs.pop('index'		,False if orient in ['split', 'table'] else True     )
													A__	     =							self.to_json_kwargs.pop('compression'		,__UpperCAmelCase     )

													if compression not in [None, "infer", "gzip", "bz2", "xz"]:
																				raise NotImplementedError(f'''`datasets` currently does not support {compression} compression'''     )

													if isinstance(self.path_or_buf		,(str, bytes, os.PathLike)     ):
																				with fsspec.open(self.path_or_buf		,'wb'		,compression=__UpperCAmelCase     ) as buffer:
																											A__	     =							self._write(file_obj=__UpperCAmelCase		,orient=__UpperCAmelCase		,lines=__UpperCAmelCase		,index=__UpperCAmelCase		,**self.to_json_kwargs     )
													else:
																				if compression:
																											raise NotImplementedError(
																											    f'''The compression parameter is not supported when writing to a buffer, but compression={compression}'''
																											    ' was passed. Please provide a local path instead.'     )
																				A__	     =							self._write(
																				    file_obj=self.path_or_buf		,orient=__UpperCAmelCase		,lines=__UpperCAmelCase		,index=__UpperCAmelCase		,**self.to_json_kwargs     )
													return written





						def 				snake_case__	(       self		,__UpperCAmelCase     )					->				Dict:
													A__     ,  A__     ,  A__     ,  A__     ,  A__	     =							args

													A__	     =							query_table(
													    table=self.dataset.data		,key=slice(__UpperCAmelCase		,offset + self.batch_size     )		,indices=self.dataset._indices		,)
													A__	     =							batch.to_pandas().to_json(
													    path_or_buf=__UpperCAmelCase		,orient=__UpperCAmelCase		,lines=__UpperCAmelCase		,index=__UpperCAmelCase		,**__UpperCAmelCase     )
													if not json_str.endswith('\n'     ):
																				json_str += "\n"
													return json_str.encode(self.encoding     )





						def 				snake_case__	(       self		,__UpperCAmelCase		,__UpperCAmelCase		,__UpperCAmelCase		,__UpperCAmelCase		,**__UpperCAmelCase		,)					->				int:
													A__	     =							0

													if self.num_proc is None or self.num_proc == 1:
																				for offset in logging.tqdm(
																				    range(0		,len(self.dataset     )		,self.batch_size     )		,unit='ba'		,disable=not logging.is_progress_bar_enabled()		,desc='Creating json from Arrow format'		,):
																											A__	     =							self._batch_json((offset, orient, lines, index, to_json_kwargs)     )
																											written += file_obj.write(__UpperCAmelCase     )
													else:
																				A__     ,  A__	     =							len(self.dataset     ), self.batch_size
																				with multiprocessing.Pool(self.num_proc     ) as pool:
																											for json_str in logging.tqdm(
																											    pool.imap(
																											        self._batch_json		,[(offset, orient, lines, index, to_json_kwargs) for offset in range(0		,__UpperCAmelCase		,__UpperCAmelCase     )]		,)		,total=(num_rows // batch_size) + 1 if num_rows % batch_size else num_rows // batch_size		,unit='ba'		,disable=not logging.is_progress_bar_enabled()		,desc='Creating json from Arrow format'		,):
																																		written += file_obj.write(__UpperCAmelCase     )

													return written






"""simple docstring"""




from io import BytesIO
from typing import List, Union

import requests

from ..utils import add_end_docstrings, is_decord_available, is_torch_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_decord_available():
				import numpy as np
				from decord import VideoReader


if is_torch_available():
				from ..models.auto.modeling_auto import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING

__lowerCamelCase											=  logging.get_logger(__name__)
@add_end_docstrings(__A )
class     UpperCamelCase__(			__A ):





						def __init__(       self		,*__UpperCAmelCase		,**__UpperCAmelCase     )					->				List[Any]:
													super().__init__(*__UpperCAmelCase		,**__UpperCAmelCase     )
													requires_backends(self		,'decord'     )
													self.check_model_type(__UpperCAmelCase     )





						def 				snake_case__	(       self		,__UpperCAmelCase=None		,__UpperCAmelCase=None		,__UpperCAmelCase=None     )					->				Any:
													A__	     =							{}
													if frame_sampling_rate is not None:
																				A__	     =							frame_sampling_rate
													if num_frames is not None:
																				A__	     =							num_frames

													A__	     =							{}
													if top_k is not None:
																				A__	     =							top_k
													return preprocess_params, {}, postprocess_params





						def __call__(       self		,__UpperCAmelCase		,**__UpperCAmelCase     )					->				Optional[int]:
													return super().__call__(__UpperCAmelCase		,**__UpperCAmelCase     )





						def 				snake_case__	(       self		,__UpperCAmelCase		,__UpperCAmelCase=None		,__UpperCAmelCase=1     )					->				List[Any]:
													if num_frames is None:
																				A__	     =							self.model.config.num_frames

													if video.startswith('http://'     ) or video.startswith('https://'     ):
																				A__	     =							BytesIO(requests.get(__UpperCAmelCase     ).content     )

													A__	     =							VideoReader(__UpperCAmelCase     )
													videoreader.seek(0     )

													A__	     =							0
													A__	     =							num_frames * frame_sampling_rate - 1
													A__	     =							np.linspace(__UpperCAmelCase		,__UpperCAmelCase		,num=__UpperCAmelCase		,dtype=np.intaa     )

													A__	     =							videoreader.get_batch(__UpperCAmelCase     ).asnumpy()
													A__	     =							list(__UpperCAmelCase     )

													A__	     =							self.image_processor(__UpperCAmelCase		,return_tensors=self.framework     )
													return model_inputs





						def 				snake_case__	(       self		,__UpperCAmelCase     )					->				Optional[int]:
													A__	     =							self.model(**__UpperCAmelCase     )
													return model_outputs





						def 				snake_case__	(       self		,__UpperCAmelCase		,__UpperCAmelCase=5     )					->				int:
													if top_k > self.model.config.num_labels:
																				A__	     =							self.model.config.num_labels

													if self.framework == "pt":
																				A__	     =							model_outputs.logits.softmax(-1     )[0]
																				A__     ,  A__	     =							probs.topk(__UpperCAmelCase     )
													else:
																				raise ValueError(f'''Unsupported framework: {self.framework}'''     )

													A__	     =							scores.tolist()
													A__	     =							ids.tolist()
													return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(__UpperCAmelCase		,__UpperCAmelCase     )]





