o
    ίi?                     @   s   d dl Z d dlZd dlZd dlZd dlZd dlZd dlmZmZm	Z	m
Z
mZmZmZ d dlZd dlZd dlmZmZ eeZdd Zdd ZG dd	 d	ZG d
d dZG dd deZG dd deZG dd dZdS )    N)AnyCallableDictIteratorListOptionalcast)IterableDatasetget_worker_infoc                  C   s8   t jj } | d u rd}d}||fS | j}| j}||fS )N   r   )torchutilsdatar
   num_workersid)worker_infor   	worker_id r   H/home/ubuntu/.local/lib/python3.10/site-packages/core/data/data_mixer.pyr
      s   r
   c                 C   s&   t  \}}| | | }|| }||fS N)r
   )rank
world_sizer   r   dataloader_rankdataloader_world_sizer   r   r   get_global_rank_info   s   
r   c                   @   s   e Zd ZdedededefddZdd Zd	d
 Zdede	e
 fddZdee fddZdee fddZdd Zdd ZdS )JSONLIteratorfpathr   
world_rankinfinitec                 C   sd   d|  kr
|k sn J ||ft |ddd| _|| _|| _|| _d| _t| || _d| _d S )Nr   rzutf-8)encoding)	openfr   r   r   line_numitergeniter_id)selfr   r   r   r   r   r   r   __init__&   s   "
zJSONLIterator.__init__c                 C      | S r   r   r'   r   r   r   __iter__6      zJSONLIterator.__iter__c                 C   
   t | jS r   nextr$   r*   r   r   r   __next__9      
zJSONLIterator.__next__returnc                 c   s    	 | j dkrtd| j d| j d |  jd7  _	 | j | jd }| _|s,n| jd | j | j kr=t	
|V  q|sAn| d  q| j  d S )NTr   zStarting iteration z over z ...r   )r   loggerinfor&   r   r"   readliner#   r   jsonloadsset_positionclose)r'   r   liner   r   r   r%   <   s"   

zJSONLIterator.genpositionc              
   C   sp   t d| j d| j d| j d|  |d u r#| jd d| _d S t|t	s*J | j| | jd | _d S )NSetting JSONL position on  (/): r   r   )
r3   warningr   r   r   r"   seekr#   
isinstanceint)r'   r;   r   r   r   r8   L   s    

zJSONLIterator.set_positionc                 C   s>   | j  }|dkr| jdkrd S | jd | j | jksJ |S )Nr   r   )r"   tellr#   r   r   )r'   file_posr   r   r   get_position[   s
   
zJSONLIterator.get_positionc                 C      | j S )K
        Return the path to a sample file to infer the content key
        r   r*   r   r   r   get_example_fileb      zJSONLIterator.get_example_filec                 C   rG   zO
        Return an identifier for the dataset this iterator represents
        rI   r*   r   r   r   get_idh   rK   zJSONLIterator.get_idN)__name__
__module____qualname__strrC   boolr(   r+   r0   r   r   r%   r   r8   rF   rJ   rM   r   r   r   r   r   %   s"    
r   c                   @   s~   e Zd ZdZdedededefddZdd	 Zd
d Z	dede
e fddZdeeef fddZdd Zdd Zdd ZdS )JSONLDirectoryIteratora  
    The JSONLDirectoryIterator is a data wrapper around a dataset folder, which contains
    multiple JSONL files. Internally, it reuses the JSONLIterator class to iterate through
    each individual file, and then wraps onto the next file once the current one is exhausted.

    Once all files in the directory have been iterated over, we wrap back to the first file
    ( if infinite is true ).

    This enables us to iterate over a dataset one chunk at a time.

    Also, note that we open the next chunk file on an ondemand basis, which means that we can
    modify chunks mid training as well to add more data, fix issues, etc.
    dirpathr   r   r   c                    s   d|  kr
|k sn J ||f| _ | _| _dd t j D } fddt|D  _t jdksAJ d j  d|rPtt	t
 t j _ntt	t
 t j _t | _d  _d S )Nr   c                 S   s   g | ]
}t d |r|qS )z.*chunk\.\d+.*\.jsonl)re	fullmatch).0xr   r   r   
<listcomp>   s    
z3JSONLDirectoryIterator.__init__.<locals>.<listcomp>c                    s   g | ]
}t j j|qS r   )ospathjoinrT   )rW   fnamer*   r   r   rY      s    zSpecified dataset location z
 is empty.)rT   r   r   rZ   listdirsortedfpathslenr   r   rQ   	itertoolscyclefpaths_generatorr$   r%   jsonl_iterator)r'   rT   r   r   r   fnamesr   r*   r   r(   ~   s    "

zJSONLDirectoryIterator.__init__c                 C   r)   r   r   r*   r   r   r   r+      r,   zJSONLDirectoryIterator.__iter__c                 C   r-   r   r.   r*   r   r   r   r0      r1   zJSONLDirectoryIterator.__next__r2   c                 c   sJ    | j d ur| j E d H  | jD ]}t|| j| jdd| _ | j E d H  qd S )NFr   r   r   )re   rd   r   r   r   )r'   r   r   r   r   r   r%      s   

zJSONLDirectoryIterator.genstatec              
   C   s   t d| j d| j d| j d|  |d }|d }|d u s$|d u r&d S t|ts-J t|ts4J | jD ]}||kr? nq7t	|| j| jdd| _
| j
| d S )	Nr<   r=   r>   r?   r   r;   Frg   )r3   r@   rT   r   r   rB   rQ   rC   rd   r   re   r8   )r'   rh   r   r;   fpath_candidater   r   r   r8      s4   

z#JSONLDirectoryIterator.set_positionc                 C   s(   | j d u r
d d dS | j j| j  dS )N)r   r;   )re   r   rF   r*   r   r   r   rF      s   
z#JSONLDirectoryIterator.get_positionc                 C   s
   | j d S )rH   r   )r`   r*   r   r   r   rJ      s   
z'JSONLDirectoryIterator.get_example_filec                 C   rG   rL   )rT   r*   r   r   r   rM      rK   zJSONLDirectoryIterator.get_idN)rN   rO   rP   __doc__rQ   rC   rR   r(   r+   r0   r   r   r%   r   r8   rF   rJ   rM   r   r   r   r   rS   o   s$    
rS   c                   @   s`   e Zd Zdi fdededededeeef f
ddZd	d
 Zdd Z	dd Z
dd Zdd ZdS )IterativeJSONLDatasetr   global_rankr   dataset_nameseeddataset_configsc                 C   s0   || _ || _|| | _|| _|| _| jj| _d S r   )_dataset_name_seed_dataset_confrl   r   
annotation	data_path)r'   rl   r   rm   rn   ro   r   r   r   r(      s   
zIterativeJSONLDataset.__init__c                 C   s~   | j | | }| j| }tj| jrt| j||dd| _n
t| j||dd| _|dkr=t	
d| j d| d|  d S d S )NTrg   )rT   r   r   r   r   zInitializing JSONLDataset z on dataloader rank z and world size )rl   r   rZ   r[   isfilert   r   re   rS   r3   r4   rp   )r'   r   r   r   r   r   r   r   worker_init   s0   


z!IterativeJSONLDataset.worker_initc                 C   s    | j  }t|tr|S d|iS )Nsingle_jsonl_position)re   rF   rB   r   )r'   posr   r   r   
state_dict  s   

z IterativeJSONLDataset.state_dictc                 C   sD   d|v r| j |d  n| j | td| j d| d d S )Nrw   zJSONLDataset z resuming from .)re   r8   r3   r4   rp   r'   ry   r   r   r   load_state_dict  s   z%IterativeJSONLDataset.load_state_dictc                 C   r)   r   r   r*   r   r   r   r+     r,   zIterativeJSONLDataset.__iter__c                 C   r-   r   )r/   re   r*   r   r   r   r0     r1   zIterativeJSONLDataset.__next__N)rN   rO   rP   rC   rQ   r   r   r(   rv   ry   r|   r+   r0   r   r   r   r   rk      s&    

rk   c                       sj   e Zd Zdg i fdededededee deeef f fdd	Z	d
d Z
dd Zdd Zdd Z  ZS )DatasetMixerr   mixrl   r   rn   preprocessorsro   c                    s  t    g _g _g _g _|_|_|_d	|
 }|
dD ]Q}|
d\} vr;td djv rHtd dt||| d}	 fdd	|D }
j|	|
f jt| j jd
 q&fdd	jD _d _d S )N ,:zDataset z not found in dataset_configs.z8 already in the mix. Each dataset can only be used once.)rl   r   rm   rn   ro   c                    s   g | ]	}|  d qS ))dataset_configr   )rW   p)ro   dsr   r   rY   K  s    z)DatasetMixer.__init__.<locals>.<listcomp>r   c                    s   g | ]	}|t  j qS r   )sumweights)rW   wr*   r   r   rY   T  s    )superr(   dataset_and_preprocessorsr   dataset_namestotalsrl   r   rn   r\   split
ValueErrorrk   appendfloatrng)r'   r~   rl   r   rn   r   ro   elemweightdataset_preprocessors	__class__)ro   r   r'   r   r(   $  sB   
	


zDatasetMixer.__init__c                 C   sV   dd t | j| jD dd t | j| jD | jd ur'dd | j D dS d dS )Nc                 S   s   i | ]\}\}}||  qS r   )ry   )rW   ds_namer   _r   r   r   
<dictcomp>Y  s    
z+DatasetMixer.state_dict.<locals>.<dictcomp>c                 S   s   i | ]\}}||qS r   r   )rW   r   totalr   r   r   r   _  s    c                 S   s$   g | ]}t |tjr| n|qS r   )rB   npndarraytolistrW   sr   r   r   rY   d  s    z+DatasetMixer.state_dict.<locals>.<listcomp>)datasetsr   r   )zipr   r   r   r   	get_stater*   r   r   r   ry   W  s   
	zDatasetMixer.state_dictc              	   C   s   |d   D ]$\}}|| jv r*| j|}| j| \}}|| |d | | j|< qtd| j d| j dt	  d |d d ur[t
j | _dd	 |d D }| j| d S d S )
Nr   r   zDatasetMixer with datasets z" resuming with total samples seen z on process rz   r   c                 S   s$   g | ]}t |trt|n|qS r   )rB   listr   arrayr   r   r   r   rY   {  s    z0DatasetMixer.load_state_dict.<locals>.<listcomp>)itemsr   indexr   r|   r   r3   r4   rZ   getpidr   randomRandomStater   	set_state)r'   ry   r   sdds_idxr   r   	rng_stater   r   r   r|   m  s"   

zDatasetMixer.load_state_dictc                 C   s8   t jj }| jD ]\}}t|dr|||j q	d S )Nrv   )r   r   r   r
   r   hasattrrv   r   )r'   r   r   r   r   r   r   r   rv     s   
zDatasetMixer.worker_initc           	      c   s    | j d u rt| j| j\}}tj||| jf| _ 	 z:| j jt	| j
| j
d}| j| \}}t|}|D ]}|d urA||| j }q5|d u rHW q| j|  d7  < |V  W n% tyz } ztd| j|  d| dt   W Y d }~nd }~ww q)NT)r   r   z#Error while iterating over dataset z: z
Traceback:
)r   r   rl   r   r   r   r   rn   choicera   r   r   r/   r   	Exceptionr3   errorr   	traceback
format_exc)	r'   r   r   src_idr   r   outpreprocessorer   r   r   r+     s4   

zDatasetMixer.__iter__)rN   rO   rP   rQ   rC   r   r   r   r   r(   ry   r|   rv   r+   __classcell__r   r   r   r   r}   #  s*    
3r}   c                   @   sf   e Zd ZdZ		dddZdd Zdd Zd	d
 Zdd Zdd Z	dd Z
dd Zdd Zdd ZdS )PersistentDataLoaderah  
    A _very_ persistent dataloader.

    Uses StatefulDataLoader to save dataset state (make sure dataset has a state_dict() and load_state_dict() method).
    Also keeps the dataloader iterator and the epoch iterator separate, so that the dataloader workers are persistent.

    Also laughs in the face of torch when it tries to kill the whole job because a worker died. Instead, this dataloader
    will just gracefully restart the underlying iterator and corresponding workers, while additionally loading the state dict
    so that it resumes from where it left off.

    This may or may not be a good idea.
    Nc              
   C   sd   ddl m} |||d||dkrdnd |t|dr|jnd dd| _|d ur*| | t| j| _d S )Nr   )StatefulDataLoaderFforkrv   r   )
batch_sizeshuffler   multiprocessing_context
collate_fnworker_init_fnsnapshot_every_n_steps)torchdata.stateful_dataloaderr   r   rv   
dataloaderr|   r$   _dataloader_iter)r'   r   r   workersr   	positionsr   r   r   r   r(     s   
zPersistentDataLoader.__init__c                 C   s
   | j  S r   )r   ry   r*   r   r   r   ry     r1   zPersistentDataLoader.state_dictc                 C   s   | j | d S r   )r   r|   r{   r   r   r   r|     s   z$PersistentDataLoader.load_state_dictc                 C   s   d S r   r   r*   r   r   r   __del__  r,   zPersistentDataLoader.__del__c                 C   r-   r   )ra   r   r*   r   r   r   __len__  r1   zPersistentDataLoader.__len__c                 C   s   |   | _| S r   )r%   r$   r*   r   r   r   r+     s   
zPersistentDataLoader.__iter__c                 C   r-   r   r.   r*   r   r   r   r0     r1   zPersistentDataLoader.__next__c                 C   s
   d | _ d S r   )r   r*   r   r   r   _refresh_iter  s   
z"PersistentDataLoader._refresh_iterc              
   C   s   | j d u r| j| j  t| j| _ zt| j W S  ttfy$     ty@ } z| j d u r:| 	 W  Y d }~S |d }~ww r   )
r   r   r|   ry   r$   r/   KeyboardInterruptStopIterationr   _get_next_sample)r'   r   r   r   r   r     s   

z%PersistentDataLoader._get_next_samplec                 c   s&    	 z|   V  W n ty    w qr   )r   r   r*   r   r   r   r%     s   zPersistentDataLoader.gen)NN)rN   rO   rP   rj   r(   ry   r|   r   r   r+   r0   r   r   r%   r   r   r   r   r     s    
!r   )rb   r6   loggingrZ   rU   r   typingr   r   r   r   r   r   r   numpyr   r   torch.utils.datar	   r
   	getLoggerrN   r3   r   r   rS   rk   r}   r   r   r   r   r   <module>   s$   $
Jw=}