o
    bi                     @   sx   d dl Z d dlZd dlZd dlZd dlmZmZ d dlZd dlm	Z	m
Z
mZ d dlmZ d dlmZ G dd deZdS )    N)OptionalUnion)_apply_list_default_encoder_make_iterable)BlockAccessor)BlockBasedFileDatasinkc                       sV   e Zd Z	ddddedeeeeeef  def fddZ	d	e
d
dfddZ  ZS )WebDatasetDatasinkTtar)file_formatpathencoderr   c                   s"   t  j|fddi| || _d S )Nr   r
   )super__init__r   )selfr   r   r   file_datasink_kwargs	__class__ e/home/ubuntu/.local/lib/python3.10/site-packages/ray/data/_internal/datasource/webdataset_datasink.pyr      s   
zWebDatasetDatasink.__init__blockfilezpyarrow.NativeFilec           
      C   s  t j|dd}t|}|D ]u}t|ts| }| jd ur%t| j|td}d|vr0t	
 j|d< |d }| D ]I\}}|d u sE|drFq8t|tsRt|tsRJ t|ts\|d}t | d| }	t||	_t |	_d\|	_|	_|	_||	t| q8q|  d S )	Nzw|)fileobjmode)default__key____zutf-8.)i  datar   )tarfileopenr   
isinstancedict	as_pydictr   r   r   uuiduuid4hexitems
startswithbytesstrencodeTarInfolensizetimemtimer   unamegnameaddfileioBytesIOclose)
r   r   r   streamsamplessamplekeykvtir   r   r   write_block_to_file   s.   





z&WebDatasetDatasink.write_block_to_file)T)__name__
__module____qualname__r*   r   r   boolcallablelistr   r   r>   __classcell__r   r   r   r   r	      s    r	   )r4   r   r/   r$   typingr   r   pyarrow3ray.data._internal.datasource.webdataset_datasourcer   r   r   ray.data.blockr   !ray.data.datasource.file_datasinkr   r	   r   r   r   r   <module>   s    