2Vh\.ddlZddlZddlZddlZddlZddlZddlZddlm Z ddl Z ddl m Z ddlmZddlmZe ddgGdd ZGd d eZiadaej,Zdad ad ZdZdZGddZGddeZddZy)N)closing) keras_export)data_adapter_utils) DataAdapterzkeras.utils.PyDatasetzkeras.utils.SequenceceZdZdZddZdZedZejdZedZ e jdZ edZ e jd Z d Z ed Z d Z d Zy) PyDataseta Base class for defining a parallel dataset using Python code. Every `PyDataset` must implement the `__getitem__()` and the `__len__()` methods. If you want to modify your dataset between epochs, you may additionally implement `on_epoch_end()`, or `on_epoch_begin` to be called at the start of each epoch. The `__getitem__()` method should return a complete batch (not a single sample), and the `__len__` method should return the number of batches in the dataset (rather than the number of samples). Args: workers: Number of workers to use in multithreading or multiprocessing. use_multiprocessing: Whether to use Python multiprocessing for parallelism. Setting this to `True` means that your dataset will be replicated in multiple forked processes. This is necessary to gain compute-level (rather than I/O level) benefits from parallelism. However it can only be set to `True` if your dataset can be safely pickled. max_queue_size: Maximum number of batches to keep in the queue when iterating over the dataset in a multithreaded or multiprocessed setting. Reduce this value to reduce the CPU memory consumption of your dataset. Defaults to 10. Notes: - `PyDataset` is a safer way to do multiprocessing. This structure guarantees that the model will only train once on each sample per epoch, which is not the case with Python generators. - The arguments `workers`, `use_multiprocessing`, and `max_queue_size` exist to configure how `fit()` uses parallelism to iterate over the dataset. They are not being used by the `PyDataset` class directly. When you are manually iterating over a `PyDataset`, no parallelism is applied. Example: ```python from skimage.io import imread from skimage.transform import resize import numpy as np import math # Here, `x_set` is list of path to the images # and `y_set` are the associated classes. class CIFAR10PyDataset(keras.utils.PyDataset): def __init__(self, x_set, y_set, batch_size, **kwargs): super().__init__(**kwargs) self.x, self.y = x_set, y_set self.batch_size = batch_size def __len__(self): # Return number of batches. return math.ceil(len(self.x) / self.batch_size) def __getitem__(self, idx): # Return x, y for batch idx. low = idx * self.batch_size # Cap upper bound at array length; the last batch may be smaller # if the total number of items is not a multiple of batch size. high = min(low + self.batch_size, len(self.x)) batch_x = self.x[low:high] batch_y = self.y[low:high] return np.array([ resize(imread(file_name), (200, 200)) for file_name in batch_x]), np.array(batch_y) ``` c.||_||_||_yN)_workers_use_multiprocessing_max_queue_size)selfworkersuse_multiprocessingmax_queue_sizes c/home/dcms/DCMS/lib/python3.12/site-packages/keras/src/trainers/data_adapters/py_dataset_adapter.py__init__zPyDataset.__init__]s $7!-cd}t|ds d|_d}t|ds d|_d}t|ds d|_d}|rt j dd yy) NFr Tr r zYour `PyDataset` class should call `super().__init__(**kwargs)` in its constructor. `**kwargs` can include `workers`, `use_multiprocessing`, `max_queue_size`. Do not pass these arguments to `fit()`, as they will be ignored.) stacklevel)hasattrr r r warningswarn)rrs r_warn_if_super_not_calledz#PyDataset._warn_if_super_not_calledbsltZ(DMDt34(-D %Dt./#%D D  MMG    rc:|j|jSr )rr rs rrzPyDataset.workersws &&(}}rc||_yr )r rvalues rrzPyDataset.workers|s  rc:|j|jSr )rr rs rrzPyDataset.use_multiprocessings &&((((rc||_yr )r r!s rrzPyDataset.use_multiprocessings $)!rc:|j|jSr )rr rs rrzPyDataset.max_queue_sizes &&(###rc||_yr )r r!s rrzPyDataset.max_queue_sizes $rct)zGets batch at position `index`. Args: index: position of the batch in the PyDataset. Returns: A batch NotImplementedError)rindexs r __getitem__zPyDataset.__getitem__ "!rcFt|dr t|Std)zNumber of batches in the PyDataset. Returns: The number of batches in the PyDataset or `None` to indicate that the dataset is infinite. __len__z`You need to implement the `num_batches` property: @property def num_batches(self): return ...)rlenr)rs r num_batcheszPyDataset.num_batchess* 4 #t9 ! >  rcy)z.Method called at the beginning of every epoch.Nrs ron_epoch_beginzPyDataset.on_epoch_begin rcy)z(Method called at the end of every epoch.Nr2rs r on_epoch_endzPyDataset.on_epoch_endr4rNrFr)__name__ __module__ __qualname____doc__rrpropertyrsetterrrr+r0r3r6r2rrrrsHT. * ^^))* *$$%% "      rrceZdZdZ ddZdZdZdZdZdZ d Z d Z d Z d Z d ZdZdZedZedZy)PyDatasetAdapterz.Adapter for `keras.utils.PyDataset` instances.NcF||_||_d|_||_d|_d|_|jj }|jj}|dkDs|dkDrA|r>t|j|||jj|j|_yyy)NFrr)rrrshuffle) py_dataset class_weightenqueuerrA_output_signature _within_epochrrOrderedEnqueuerr)rxrCrArrs rrzPyDatasetAdapter.__init__s (  !%"//))"ooAA Q;7Q;+>+$7#== DM,?;rct|tr|St|tjr|f}t|tr t |}t|t r t |dvr&tdt|dddt||jPt |dk(r tdt |dk(r)tj|d|j}||fz}|S) N>rrzPyDataset.__getitem__() must return a tuple or a dict. If a tuple, it must be ordered either (input,) or (inputs, targets) or (inputs, targets, sample_weights). Received: dz ... of type rJzGYou cannot specify `class_weight` and `sample_weight` at the same time.rr) isinstancedictnpndarraylisttupler/ ValueErrorstrtyperCrclass_weight_to_sample_weights)rbatchsws r_standardize_batchz#PyDatasetAdapter._standardize_batchs eT "L eRZZ (HE eT "%LE%'3u:Y+F!Z-.l4;- I     (5zQ <5zQ'FF!Hd//  rc#~KtjD]"}|j|j|$ywr ) itertoolscountrXrB)ris r_infinite_generatorz$PyDatasetAdapter._infinite_generators6" >A))$//!*<= = >s;=c#Kt|jj}|jr t |}t j||D]"}|j |j|$ywr )rangerBr0rArPrandomrX)rindicesr\s r_finite_generatorz"PyDatasetAdapter._finite_generators_334 <<7mG NN7 # >A))$//!*<= = >sA4A6c#K|jj|jjD]}|j|ywr )rDstartgetrX)rrVs r_infinite_enqueuer_generatorz-PyDatasetAdapter._infinite_enqueuer_generatorsC ]]&&( 1E))%0 0 1sAAc#0K|jj|jj}t |jj D]<\}}|j |||dz k\s"|jjyyw)Nr)rDrdrBr0 enumeratererXstop)rr0r\rVs r_finite_enqueuer_generatorz+PyDatasetAdapter._finite_enqueuer_generatorsz oo11 !$--"3"3"56 HAu))%0 0K!O# ""$  s A5B8Bc|j6|jj|jS|j S|jj|j S|j Sr )rDrBr0r]rbrfrjrs r _get_iteratorzPyDatasetAdapter._get_iteratorsh == **2//11--//**288::6688rcHtj|jSr )rget_numpy_iteratorrlrs rrnz#PyDatasetAdapter.get_numpy_iterators!44T5G5G5IJJrcHtj|jSr )rget_jax_iteratorrlrs rrpz!PyDatasetAdapter.get_jax_iterators!2243E3E3GHHrcddlm}|jj}|jt j }| t||}t|Dcgc] }|j|j|"}}t|dk(r tdt j||_|jjj|j |j}|j"O|jj%}d|j&_d|j*_|j/|}|S|j1|jj2}|Scc}w)Nr) tensorflowzThe PyDataset has length 0)output_signatureFr)keras.src.utils.module_utilsrrrBr0rErNUM_BATCHES_FOR_TENSOR_SPECminr_rXr/rRget_tensor_specdataDatasetfrom_generatorrlrDOptionsautotuneenabled threadingprivate_threadpool_size with_optionsprefetchAUTOTUNE)rtfr0 num_samplesr\batchesdsoptionss rget_tf_datasetzPyDatasetAdapter.get_tf_datasetsDAoo11  ! ! ),HHK&!+{; {+''(:;G7|q  !=>>%7%G%G%PD " WW__ + +   !33,  == $ggoo'G',G   $89G   5)B RWW--.B -s%E+cHtj|jSr )rget_torch_dataloaderrlrs rrz%PyDatasetAdapter.get_torch_dataloader:s!66t7I7I7KLLrc|jr tdd|_|jr|jj|jj y)NzL`on_epoch_begin` was called twice without `on_epoch_end` having been called.T)rFrRrDrdrBr3rs rr3zPyDatasetAdapter.on_epoch_begin=sN   & " == MM   ! &&(rc|jr|jj|jjd|_yNF)rDrirBr6rFrs rr6zPyDatasetAdapter.on_epoch_endHs1 == MM    $$&"rc.|jjSr )rBr0rs rr0zPyDatasetAdapter.num_batchesNs***rcyr r2rs r batch_sizezPyDatasetAdapter.batch_sizeRsrr)r8r9r:r;rrXr]rbrfrjrlrnrprrr3r6r<r0rr2rrr?r?sz8  08>>1  9KI@M )# ++rr?Fcf|rtrtjjStjSr )_FORCE_THREADPOOLmultiprocessingdummyPool)rs rget_pool_classres& "3$$)))   rcBttjatS)z,Lazily create the queue to track worker ids.)_WORKER_ID_QUEUErQueuer2rrget_worker_id_queuerls*002 rct||S)aGet the value from the PyDataset `uid` at index `i`. To allow multiple PyDatasets to be used at the same time, we use `uid` to get a specific one. A single PyDataset would cause the validation to overwrite the training PyDataset. This methods is called from worker threads. Args: uid: int, PyDataset identifier i: index Returns: The value at index `i`. )_SHARED_SEQUENCES)uidr\s r get_indexrts S !! $$rcPeZdZdZ d dZdZdZd dZdZdZ dZ d Z d Z y )PyDatasetEnqueueraBase class to enqueue inputs. The task of an Enqueuer is to use parallelism to speed up preprocessing. This is done with processes or threads. Example: ```python enqueuer = PyDatasetEnqueuer(...) enqueuer.start() datas = enqueuer.get() for data in datas: # Use the inputs; training, evaluating, predicting. # ... stop sometime. enqueuer.stop() ``` The `enqueuer.get()` should be an infinite stream of data. cj||_t tjddat tt rt|_tdz anKtj5tj|_txjdz c_ dddtj|_ tj||_ d|_tj |_d|_|r|j'|_yfd|_y#t$rdaYwxYw#1swYxYw)Nr\rrFc&tdSr)r)_rs rz,PyDatasetEnqueuer.__init__..s)>)>w)Gr)rB_SEQUENCE_COUNTERrValueOSErrorrLintrget_lockr"queuer ready_queue future_queuerunningr~Lockstart_stop_lock run_thread_get_executor_init executor_fn)rrBrrrs ` rrzPyDatasetEnqueuer.__init__s%  $ &$3$9$9#q$A! ' -(DH  " #++- -,22!''1,' -!;;=!KK7 (~~/ #66w?D  HD 1 &%&!  & - -sD#/D) D&%D&)D2c|jS)zWhether the enqueuer is running. This method is thread safe and called from many threads. Returns: boolean indicating whether this enqueuer is running. )rrs r is_runningzPyDatasetEnqueuer.is_runnings||rc\|j5|jr dddyd|_tj|j|_d|j |j _d|j _|j jdddy#1swYyxYw)zStarts the handler's workers. This method is thread safe but is called from the main thread. It is safe to call this method multiple times, extra calls are ignored. NT)targetWorker_) rrr~Thread_runrrnamedaemonrdrs rrdzPyDatasetEnqueuer.starts  ! ! $|| $ $ DL'..dii@DO%,TXXJ#7DOO %)DOO " OO ! ! # $ $ $sB"A5B""B+c|j5|js dddyd|_|ry |jjdd}t |t r||j}|jj ||jj|xd|_ dt|j<dddy#tj$rYnwxYw|jjW#1swYyxYw)aoStops running threads and wait for them to exit, if necessary. This method is thread safe and is called from various threads. Note that the `drain_queue_and_join` argument must be set correctly. It is safe to call this method multiple times, extra calls are ignored. Args: drain_queue_and_join: set to True to drain the queue of pending items and wait for the worker thread to complete. Set to False if invoked from a worker thread to avoid deadlocks. Note that setting this to False means this enqueuer won't be reused. NFT皙?blocktimeout)rrrrerL Exception task_donerputrEmptyrjoinrr)rdrain_queue_and_joinr"inputss rrizPyDatasetEnqueuer.stops ! ! /<< / /!DL#  $ 1 1 5 5D# 5 N%eY7"'K!&))335!- ,,008#DO*. dhh '- / /"!;;$$&' / /s4C= C=A6C %C= CC=CC==Dc>|jt|j<y)z&Sends current Iterable to all workers.N)rBrrrs r_send_py_datasetz"PyDatasetEnqueuer._send_py_datasets'+oo$((#rc(|jdy)NFr)rirs r__del__zPyDatasetEnqueuer.__del__s u -rct)z?Submits request to the executor and queue the `Future` objects.r(rs rrzPyDatasetEnqueuer._runs!!rct)Gets the Pool initializer for multiprocessing. Args: workers: Number of workers. Returns: Function, a Function to initialize the pool r()rrs rrz$PyDatasetEnqueuer._get_executor_init r,rct)6Creates a generator to extract data from the queue. Skip the data if it is `None`. This method is called from the main thread. Yields: The next element in the queue, i.e. a tuple `(inputs, targets)` or `(inputs, targets, sample_weights)`. r(rs rrezPyDatasetEnqueuer.gets "!rNr7)T) r8r9r:r;rrrdrirrrrrer2rrrrsB.! %HN $#/J6 ." " "rrc>eZdZdZ dfd ZdZdZdZxZS)rGaBuilds a Enqueuer from a PyDataset. Args: py_dataset: A `keras.utils.PyDataset` object. use_multiprocessing: use multiprocessing if True, otherwise threading shuffle: whether to shuffle the data at the beginning of each epoch ct|||||||_|jjt j |_yyr )superrrArBr0rZr[ra)rrBrrrrA __class__s rrzOrderedEnqueuer.__init__/sJ  !4n  ?? & & .%??,DL /rcfd}|S)rcxtdt|dtf}tj ||S)NT) initializerinitargs)rinit_pool_generatorr _DATA_POOLSadd)seqspoolrs rpool_fnz3OrderedEnqueuer._get_executor_init..pool_fnJs;'>$'/&9&;<D OOD !Krr2)rrrs ` rrz"OrderedEnqueuer._get_executor_init@s rc |jj[t|jj}|jr t |}t j|t ||_|jt|jt5}|jrd t|j}|jj|j!t"|j$|fd|jrddddy#t&$rYwxYw#1swYyxYw#t($r%}|jj|Yd}~yd}~wwxYw)zSubmits request to the executor and queue the `Future` objects. This method is the run method of worker threads. NTr)rBr0r_rArPr`iterrarrrrrnextrr apply_asyncrr StopIterationr)rraexecutorr\es rrzOrderedEnqueuer._runUs)  %**6  ; ;<<<"7mGNN7+#G}  ! ! #))*;<= oo' .))--$00TXXqMJ"&.oo'  )   %    ! !! $ $ %s[BD:!D.3ADD.D: D+(D.*D++D..D73D:7D:: E(E##E(c#K|jr" |jjd}|2td#tj$rYnwxYw |j jdd}|j j t|tr||j}||n9#tj$rYn$t$r}|jd|d}~wwxYw|jrЭw) rFrTrNrz@Iterator called after `on_epoch_end` or before `on_epoch_begin`.) rrrerrrrrLrrirR)rrr"rs rrezOrderedEnqueuer.getrsoo ))--E-: * N  );;   ))--D!-D!!++-eY/K% L;;   t 4 %oosXD A DADADAB:9D:C0 DC0C++C00D)rFrF) r8r9r:r;rrrre __classcell__)rs@rrGrG&s*! -"*%:' rrGc|atj}d|j|_|,tj j ||jz||j|jddyy)a_Initializer function for pool workers. Args: gens: State which should be made available to worker processes. random_seed: An optional value with which to seed child processes. id_queue: A multiprocessing Queue of worker ids. This is used to indicate that a worker process was created by Keras. Keras_worker_NTrr) rrcurrent_processrrNr`seedidentr)gens random_seedid_queue worker_procs rrrst!113K'{'7'7&89K {[%6%667 [&&dC @r)NN) rZmultiprocessing.dummyrrr`r~rweakref contextlibrnumpyrNkeras.src.api_exportr keras.src.trainers.data_adaptersr-keras.src.trainers.data_adapters.data_adapterrrr?rrWeakSetrrrrrrrrGrr2rrrs -?E&(>?@a a Aa H^{^D goo  %&\"\"~s 's lAr