&`i- ddlZddlZddlZddlZddlmZddlmZmZm Z m Z ddl Z ddl ZddlmZddlmZddlmZmZddlmZmZmZddlmZmZdd lmZerdd lm Z d Z!ej"e#Z$d ed e%fdZ&dee%de%fdZ'e dddde(de ej)j*eej)j*ffdZ+eGddeZ,dS)N)Path) TYPE_CHECKINGListOptionalTuple) InputReader) IOContext)from_json_datapostprocess_actions)DEFAULT_POLICY_ID SampleBatchconcat_samples) PublicAPIoverride)SampleBatchType)AlgorithmConfigg?fpath extract_pathctjt|d5}||ddddS#1swxYwYdS)Nr)zipfileZipFilestr extractall)rrzip_refs t/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/rllib/offline/dataset_reader.py_unzip_this_pathrs US ) ))W<((())))))))))))))))))sAA  A pathsformatc&g}|D] }tjdt|rt|drt dd} t t||na#t $rT t ttj j |z |n #t $rt d|wxYwYnwxYwtt| t|j d|z }| |8t|dr| |qt| sctttj j |z }t| st d||}| | |S)zLIf a path in paths is a zip file, unzip it and use path of the unzipped filez\.zip$zs3://z?unzip_if_needed currently does not support remote paths from s3z./zFile not found: .)researchr startswith ValueErrorrFileNotFoundErrorr__file__parentabsolutestemappendexists)rr ret_pathspathr unzipped_path relative_paths r_unzip_if_neededr1s.I ' ' 9YD * * '4yy##G,,  U L G TL9999$ G G GG$T(^^%:%AD%H,WWWW(GGG+,Et,E,EFFFGXW  G \""++--4::?0M0MV0M0MMM   ] + + + +4yy##G,, '  &&&&Dzz((**)$'X(=(Dt(K$L$LM ..5577T/0R=0R0RSSS(D  &&&& s*A== C/B87C8CCCconfigr num_workersreturnc |jdksJd|j|j}|d}ddg}|||vrtd|d||d }|d }|r|s|rtd |r|s|std |st |t r|g}nJt |t r&t |d t s Jdntdt||}|d|pd}|dt}|r |} nf|dkr%tj ||d|i} n;|dkr%tj ||d|i} ntd||d kr| | gfS| |d|} | dg| zfS)aReturns a dataset and a list of shards. This function uses algorithm configs to create a dataset and a list of shards. The following config keys are used to create the dataset: input: The input type should be "dataset". input_config: A dict containing the following key and values: `format`: str, speciifies the format of the input data. This will be the format that ray dataset supports. See ray.data.Dataset for supported formats. Only "parquet" or "json" are supported for now. `paths`: str, a single string or a list of strings. Each string is a path to a file or a directory holding the dataset. It can be either a local path or a remote path (e.g. to an s3 bucket). `loader_fn`: Callable[None, ray.data.Dataset], Instead of specifying paths and format, you can specify a function to load the dataset. `parallelism`: int, The number of tasks to use for loading the dataset. If not specified, it will be set to the number of workers. `num_cpus_per_read_task`: float, The number of CPUs to use for each read task. If not specified, it will be set to 0.5. Args: config: The config dict for the algorithm. num_workers: The number of shards to create for remote workers. Returns: dataset: The dataset object. shards: A list of dataset shards. For num_workers > 0 the first returned shared would be a dummy None shard for local_worker. datasetzQMust specify config.input_ as 'dataset' if calling `get_dataset_and_shards`. Got rjsonparquetNzUnsupported format z. Supported formats are r loader_fnzBWhen using a `loader_fn`, you cannot specify a `format` or `path`.zMMust specify either a `loader_fn` or a `format` and `path` in `input_config`.rz%Paths must be a list of path strings.z6Paths must be a path string or a list of path strings. parallelismnum_cpus_per_read_tasknum_cpus)r:ray_remote_argsz!Un-supported Ray dataset format: F) num_blocksshuffle)input_ input_configgetr% isinstancerlistr1DEFAULT_NUM_CPUS_PER_TASKraydata read_json read_parquet repartitionsplit) r2r3rBrsupported_fmtsrr9r: cpus_per_taskr6 remote_shardss rget_dataset_and_shardsrPFsB =I % % % B28- B B & % % &L   h ' 'Fi(N fN:: R& R R. R R     W % %E  --I f   P    u i       eS ! ! WGEE t $ $ WeAh,, U U.U U U, UUVV V //""=+2BCCK $$ ";M F)++ 6  ($$ {Z.iteratorsI.5545@@B!||~~-------.rQ)r _ioctx_default_policy policy_map preprocessorr\countrGrH DataContext get_currentenable_progress_barsr2rC batch_sizemaxmathceilworker _policy_mapr preprocessorsprint worker_index_iter)r_rTrUr3r`rYs` @r__init__zDatasetReader.__init__s*y{{ 155t  !%ITTDM4G4G4I4I BG((**?+,001CQGGk(,,->BB {!%%fd33  O!$)DOk,I"J"JANNDO  {!-#';#5#@ '+'7';';z7DatasetReader._preprocess_if_needed.. s*LLLA*44Q77LLLrQ)rdr CUR_OBSNEXT_OBSnpstack)r_rzkeys` rrvz#DatasetReader._preprocess_if_neededsi   #+[-AB  %<rs  777777777777666666222222MMMMMMMMXXXXXXXXXX;;;;;;;;222222FEEEEEE  8 $ $)D))))) $DI$s$$$$N 23i/i/ i/,/i/ 38 T#("23 34i/i/i/ i/X nnnnnKnn nnnrQ