Ti"NddlmZmZddlZddlZddlZddlZddlmZmZm Z m Z ddl Z ddl m ZddlmcmcmZddl ZddlZddlmcmZddlmcmZddlmZddlm Z!e ddZ"Gdd ee"Z#Gd d ee"Z$Gd d ee"Z%Gdde#e"Z&Gdde$e"Z'Gdde#Z(dej)de*dej+de#fdZ,dddej)de*de*de-de#f dZ.ddde#dej)d e-de#fd!Z/ddd"de$dej)d e-d#e-de$f d$Z0ddddd%d&d'ej1d(e j2j3dzde-d)e*dzd e-d*ed+de%e4ej5ej6ffd,Z7ddddddd%d-dej)dej+de*de*d(e j2j3dzd e-de-d)e*dzd.e*d/e*d*e8de%e4ej5ej6ffd0Z9ddddd%d1dej)de*de*d(e j2j3dzd e-de-d)e*dzd*e8de%e4ej5ej6ffd2Z:Gd3d4Z;d5ZGd:d;e%Z?dS)<)IteratorSequenceN)LiteralProtocol SupportsIndexTypeVar)DroidRldsDatasetT_coT) covariantc.eZdZdZdedefdZdefdZdS)Datasetz+Interface for a dataset with random access.indexreturnc td)Nz3Subclasses of Dataset should implement __getitem__.NotImplementedErrorselfrs Y5Z::::5545555"""""""rr1cPeZdZdddedeejdefdZdZ de fd Z d S) IterableTransformedDatasetF is_batchedr2r3rDcT||_tj||_||_dSr5)r6r7r8r9 _is_batched)rr2r3rDs rr:z#IterableTransformedDataset.__init__Bs+  %-j99%rc#<KjD]jrotdD}fdt |D}fd|D}t jjdg|RVxVdS)Nc30K|]}|jdVdS)rN)shape).0vs r z6IterableTransformedDataset.__iter__..Rs(!F!F!'!*!F!F!F!F!F!FrcXg|]%tjfd&S)c|Sr5r#)xis rz@IterableTransformedDataset.__iter__...Us QqTrjaxtreemap)rJrPsamples @r z7IterableTransformedDataset.__iter__..Us0%f%f%fqchll>>>>6&J&J%f%f%frc:g|]}|Sr#)r9)rJsrs rrWz7IterableTransformedDataset.__iter__..Xs%NNNatq11NNNrc.tj|dS)Nraxisnpstack)rOs rrQz5IterableTransformedDataset.__iter__..[sbhqq.A.A.Ar) r6rFnextvaluesrangerSrTrUr9)r batch_sizeindividual_samples transformedrVs` @rr'z#IterableTransformedDataset.__iter__Msm . .F ."!F!Ffmmoo!F!F!FFF &g%f%f%fTYZdTeTe%f%f%f"ONNN;MNNN hl#A#APKPPPPPPPoof------ . .rrc*t|jSr5r=rs rrz"IterableTransformedDataset.__len___r?rN) rrr r%rr7r@boolr:r'r"rr#rrrBrBAs ! & & &  &[89 &  & & & &...$"""""""rrBcDeZdZdejdefdZdedefdZ defdZ dS) FakeDataset model_config num_samplescV||_|\|_|_dSr5) _num_samples inputs_spec_observation_spec _action_spec)rrjrks rr:zFakeDataset.__init__ds,'4@4L4L4N4N1 1 1 1rrrcRtj|dtjffd }tj||j}tj||j}i| d|iS)Nspecc|tj\}|jdd}|jt jkr#tj||ddS|jt jkr#tj ||ddSt j ||jS)Ngg?)rIminvalmaxvalri)rIdtype) rSrandomsplitrIrwjnpfloat32uniformint32randintzeros)rrdata_rngrIrngs rmake_from_specz/FakeDataset.__getitem__..make_from_specksJ,,S11MCJqrrNEzS[((z))(%UX)YYYzSY&&z))(%RV)WWW95 ;;; ;ractions) rSrxkey __index__ShapeDtypeStructrTrUrorpto_dict)rrr observationactionrs @rrzFakeDataset.__getitem__hsjnnU__..// <!5 < < < < < <hll>43IJJ nd.?@@ !!## v   rc|jSr5)rmrs rrzFakeDataset.__len__~   rN) rrr _modelBaseModelConfigr"r:rdictrrr#rrrirics|OV%;O#OOOO  4    ,!!!!!!!rrir,action_horizonrjrcJ|j}|td|dkrt|dStj|tj|jfd|jDd}|jr(t|tj j g}|S) zCreate a dataset for training.Nz*Repo ID is not set. Cannot create dataset.fakei)rkcHi|]}|fdtDS)c$g|] }|jz  Sr#)fps)rJt dataset_metas rrWz3create_torch_dataset...s!FFF1!l&&FFFr)rb)rJrrrs r z(create_torch_dataset..sE   KNCFFFFn0E0EFFF   rg.A)delta_timestamps tolerance_s) repo_id ValueErrorrilerobot_datasetLeRobotDatasetMetadataLeRobotDatasetaction_sequence_keysprompt_from_taskr1r7PromptFromLeRobotTasktasks)r,rrjrr2rs ` @rcreate_torch_datasetrs!GEFFF&`. use_quantiles) r norm_statsrr1repack_transformsinputsdata_transformsr7 Normalizeuse_quantile_normmodel_transforms)r2r,rrs rtransform_datasetrsJf$$_$  ! )` !+   * 1  ( /   !*K#; MMGoVVVG GI   + + - - *k&2EE".==??&//11 FG *U->-M-M-O-OO  )  %):)<)<< L8&688999!)"i//XD,W    K +{ 3 33r)rrrrrc|dkrtdt||||}t|||d}t|||} t || S)aCreate an RLDS data loader for training. Note: This data loader requires some extra dependencies -- see examples/droid/README_train.md Args: data_config: The data configuration. action_horizon: The action horizon. batch_size: The batch size. sharding: The sharding to use for the data loader. If None, the data loader will use a single device sharding. skip_norm_stats: Whether to skip data normalization. shuffle: Whether to shuffle the data. num_batches: Determines the number of batches to return. If the number exceeds the number of batches in the dataset, the data loader will loop over the dataset. If not provided, will iterate over the dataset indefinitely. rz-PyTorch RLDS data loader is not supported yetrTrrr)rrrRLDSDataLoaderr) r,rrcrrrrrr2rs rrrUs{6I!"QRRR!+~zSZ[[[G(+kopppG K +{ 3 33rceZdZdZdddddddddedejjdzd ed e j j j dzd edzd ed ede fdZede j j jfdZdZdS)rz!Torch data loader implementation.NFrrS)rrrrrrrrrrrrrrrc .tjdkrtdt||kr#t d|dt|d||_|w| dkrqtjtjtj dtj d |_||_ d} |d krtj d } tj} | |tjjt)jtjjj|||duo|||| |d kt.t0d | |_dS)aCreate a PyTorch data loader. Args: dataset: The dataset to load. local_batch_size: The local batch size for each process. sharding: The sharding to use for the data loader. shuffle: Whether to shuffle the data. num_batches: If provided, determines the number of returned batches. If the number is larger than the number of batches in the dataset, the data loader will loop over the dataset. If not provided, will iterate over the dataset indefinitely. num_workers: The number of worker processes to use. If zero, the data loader will execute in the main process. seed: The seed to use for shuffling the data. rt6Data loading with multiple processes is not supported.zLocal batch size (z#) is larger than the dataset size (z).NrSBrrspawnT) rcrrrmultiprocessing_contextpersistent_workers collate_fnworker_init_fnr generator)rSrrr>r _shardingr NamedShardingMeshdevices PartitionSpec _num_batchesmultiprocessing get_contextr Generator manual_seedrrr*typingcastr _collate_fn_worker_init_fn _data_loader) rr2rrrrrrrr mp_contextrs rr:zTorchDataLoader.__init__sy8     " "%&^__ _ w<<* * *w2Bwwgjkrgsgswwwxx x"   U 2 2 \77 !!#+--88 **3//DN( ??(4W==JO%% d###!K,77 K (0' : :'_0#$.*Q"*8   rrc|jSr5)rrs r torch_loaderzTorchDataLoader.torch_loaders   rc#bKd} tj} j |jkrdS t|}n#t$rYncwxYw|dz }j&t jfd|Vn,t jtj |V)NrTrtc8tjj|Sr5rS"make_array_from_process_local_datarrOrs rrQz*TorchDataLoader.__iter__..s1WX\Xfhi1j1jr) iterrrr` StopIterationrrSrTrUr as_tensorr num_items data_iterbatchs` rr'zTorchDataLoader.__iter__s  ?T.//I ?$0Y$BS5S5SF OOEE$EQ >-(,,'j'j'j'jlqrrrrrr(,,u>>>>> ? ?A A A)rrr r!r"rSrShardingrgrrrSamplerstrr:propertyr*rr'r#rrrr~s++2637"&> > > > ,'$. >  > !)D0> 4Z> > > > > > > @!ek.9!!!X!?????rrc2tjjdg|RS)z5Collate the batch elements into batched numpy arrays.cBtjd|DdS)Nc6g|]}tj|Sr#)r^asarray)rJrOs rrWz1_collate_fn....s -H-H-Hbjmm-H-H-Hrrr[r])xss rrQz_collate_fn..s$BH-H-HR-H-H-Hq$Q$Q$QrrR)itemss rrrs# 8<QQ ZTY Z Z ZZr worker_idcBdtjd<dtjd<dS)zETell JAX inside the worker process not to preallocate the GPU memory.falseXLA_PYTHON_CLIENT_PREALLOCATEplatformXLA_PYTHON_CLIENT_ALLOCATORN)osenviron)rs rrrs$3:BJ./0:BJ,---rcTeZdZdZddddedejjdzdedzfdZ dZ dS) rzShallow wrapper around the DROID data loader to make it compatible with openpi. All batching already happens in the DROID dataset, so we don't need to do anything here. Nrr2rrcf||_||_tjdkrt d|ltjtjtjdtj d}||_ ||_dS)Nrtrrr) r6rrSrrrrrrrr)rr2rrs rr:zRLDSDataLoader.__init__s  '     " "%&^__ _  |11 !!#+--88 **3//H "'rc#Kd} tj} j |jkrdS t|}n#t$rYn/wxYw|dz }t jfd|V`v)NrTrtc8tjj|Sr5rrs rrQz)RLDSDataLoader.__iter__..sS-STXTbde-f-fr)rr6rr`rrSrTrUr s` rr'zRLDSDataLoader.__iter__s  oT]++I o$0Y$BS5S5SF OOEE$EQ hll#f#f#f#fhmnnnnn o or ) rrr r!r rSrrr"r:r'r#rrrrs26"& (((!(,'$. ( 4Z ((((. o o o o orrcJeZdZdejdeezfdZdejfdZdZ dS)rr,rc"||_||_dSr5) _data_configr)rr,rs rr:zDataLoaderImpl.__init__s''rrc|jSr5)r&rs rr,zDataLoaderImpl.data_configrrc#pK|jD]+}tj||dfV,dS)Nr)rr Observation from_dict)rr s rr'zDataLoaderImpl.__iter__sP& H HE$..u55uY7GG G G G G H HrN) rrr r.r/rrr:r,r'r#rrrrsq(G$6(_WeEe((((!W/!!!!HHHHHrr)@collections.abcrrrrrrrrrrrS jax.numpynumpyrz'lerobot.common.datasets.lerobot_datasetcommonrrr^ropenpi.models.modelmodelsrropenpi.training.configtrainingrr."openpi.training.droid_rlds_datasetr openpi.transformsr3r7r r r%r*r1rBrir/r"rrrgrrr TrainConfigrrtupler)Actionsrrrrrrrrrr#rrr9s........ <<<<<<<<<<<< AAAAAAAAAAAA $$$$$$$$$(((((((((??????''''''wv&&&UUUUUhtnUUUUUUUUhtnUUUYYYYY$YYY " " " " " " " """"""!6"""D!!!!!'!!!>#58HNH^ < #    $ejwW5G]anu4"  #     @.2"!+0---  -l#d*- - t -  -'(-f(&.89:----l.2!"B4B4B4#B4(B4B4 B4 l#d* B4B4B4tB4B4 B4B4f(&.89:B4B4B4B4T.2!"&4&4&4#&4&4&4 l#d* &4  &4&4t&4&4f(&.89:&4&4&4&4RW?W?W?W?W?W?W?W?t[[[;s;t;;;;)o)o)o)o)o)o)o)oX H H H H HZ H H H H Hr