o
    /ѹg\                     @   s  d dl Z d dlmZ d dlZd dlmZmZ d dlm	Z	 d dl
mZ d dlmZ d dlmZmZmZmZmZmZ d dlZd dlmZ d dlmZ d d	lmZ d d
lmZmZ d dl m!Z!m"Z" d dl#m$Z$ d dl%m&Z& d dl'm(Z(m)Z)m*Z* d dl+m,Z,m-Z-m.Z. d dl/m0Z0m1Z1 d dl2m3Z3m4Z4 d dl5m6Z6 d dl7m8Z8 ddl9m:Z:m;Z; e4 rd dl<Z<e3 r	 ed Z=ee>e?eddf Z@ee>edf ZAe6 ZBG dd deZCG dd deCZDdd ZEdS )     N)ABCabstractmethod)partial)Pool)Lock)AnyDict	GeneratorListMappingUnion)version)Model)	MsDataset)TASK_OUTPUTSModelOutputBase)TASK_INPUTScheck_input_type)Preprocessor)Config)
FrameworksInvoke	ModelFile)create_devicedevice_placementverify_device)read_configsnapshot_download)is_tf_availableis_torch_available)
get_logger)compile_model   )is_modelis_official_hub_path)ztorch.Tensorz	tf.TensorzImage.Imageznumpy.ndarrayztorch.nn.Modulec                
   @   sz  e Zd ZdZdd Zdee fddZ							d/d
ede	eee f de	e
ee
 f defddZdd ZdefddZde	eee f de	eeef ef fddZdd ZdefddZdd Zdedeeef fddZd d! Zdee deeef fd"d#Zd$d% Zd&d' Zd(edeeef fd)d*Zd(eeef deeef fd+d,Zd(eeef deeef fd-d.ZdS )0PipelinezPipeline base.
    c                 K   sp   t |trtd|  t |tr6t|r6td| d t|r4tj|f| jdt	j
| jd|S |S |S )Nzinitiate model from zinitiate model from location .T)deviceZmodel_prefetchedZ
invoked_by
device_map)
isinstancestrloggerinfor$   r#   r   from_pretraineddevice_namer   ZPIPELINEr(   )selfmodelkwargs r2   ]/Users/admin/.pyenv/versions/3.10.0/lib/python3.10/site-packages/modelscope/pipelines/base.pyinitiate_single_model/   s$   
zPipeline.initiate_single_modelinput_modelsc                 C   s"   g }|D ]
}| | | q|S N)appendr4   )r/   r5   modelsr0   r2   r2   r3   initiate_multiple_models?   s   z!Pipeline.initiate_multiple_modelsNgpuTconfig_filer0   preprocessorr'   c           	      K   s\  |dur|dksJ d|| _ t| || _t|ts+| j|fi || _| jg| _n	d| _| || _t	| jdk| _
|durMt|| _tj|}n| j
sct| jtrZ| j}n| jj}t|| _|du rq| j
sqt|| _n|| _| js| j
r| jd r|  | _nd| _| jtjkrt| j| _d| _t | _|| _ |!dd| _"|!di | _#dS )	aH   Base class for pipeline.

        If config_file is provided, model and preprocessor will be
        instantiated from corresponding config. Otherwise, model
        and preprocessor will be constructed separately.

        Args:
            config_file(str, optional): Filepath to configuration file.
            model: (list of) Model name or model object
            preprocessor: (list of) Preprocessor object
            device (str): device str, should be either cpu, cuda, gpu, gpu:X or cuda:X
            auto_collate (bool): automatically to convert data to tensor or not.
            compile (bool, optional): Compile the model with torch 2.0, default False
            compile_options (dict, optional): The compile options if compile=True,
                default None to use the default params of 'TorchModel.compile'.
        Nr:   z;`device` and `device_map` cannot be input at the same time!r"   r   FcompileZcompile_options)$r(   r   r.   r)   r
   r4   r0   r8   r9   lenhas_multiple_modelsr   	from_filecfgospathdirnamer*   	model_dirr   r   r-   r<   _get_framework	frameworkr   torchr   r'   _model_preparer   _model_prepare_lock_auto_collateget_compile_compile_options)	r/   r;   r0   r<   r'   auto_collater(   r1   rE   r2   r2   r3   __init__E   s@   

zPipeline.__init__c                    s    j jdd  fdd} jsH jtjkrE jr2 jD ]}|| q jr1 fdd jD  _n| j	  jrEt
 j	fi  j _	d _ j   dS )	zQ Place model on certain device for pytorch models before first inference
        iX  )timeoutc                    s`   t | tjjst| dr| j} t | tjjsd S |   ddlm} || r.| 	 j
 d S d S )Nr0   r   )is_on_same_device)r)   rH   nnModulehasattrr0   evalmodelscope.utils.torch_utilsrR   tor'   )r0   rR   r/   r2   r3   _prepare_single   s   z/Pipeline.prepare_model.<locals>._prepare_singlec                    s   g | ]}t |fi  jqS r2   )r!   rN   ).0mrY   r2   r3   
<listcomp>   s    z*Pipeline.prepare_model.<locals>.<listcomp>TN)rJ   acquirerI   rG   r   rH   r?   r8   rM   r0   r!   rN   release)r/   rZ   r\   r2   rY   r3   prepare_model   s&   




zPipeline.prepare_modelreturnc                    s|   g  | j D ]}t|tr|}n|j}t|tj}t	|} 
|j qt fdd D s:td   d S  d S )Nc                 3   s    | ]	}| d  kV  qdS )r   Nr2   )r[   xZ
frameworksr2   r3   	<genexpr>   s    z*Pipeline._get_framework.<locals>.<genexpr>z:got multiple models, but they are in different frameworks r   )r8   r)   r*   rE   ospjoinr   ZCONFIGURATIONr   r@   r7   rG   allr+   warning)r/   r\   rE   Zcfg_filerA   r2   rc   r3   rF      s   


zPipeline._get_frameworkinputc           
      O   s  | j s| jr| jd r| js|   |dd }| jd
i |\}}}||d< ||d< ||d< dt| jv rCt	|t
rCd|i}d|d	< t	|t
rn|d u rbg }|D ]}	|| j|	g|R i | qP| j||fi |}|S t	|tr| j|g|R i |S | j|g|R i |}|S )Nr   
batch_sizepreprocess_paramsforward_paramspostprocess_paramsZLLMPipelinemessagesTZ
is_messager2   )r0   r?   r8   rI   r`   pop_sanitize_parameterstype__name__r)   listr7   _process_single_process_batchr   _process_iterator)
r/   ri   argsr1   rj   rk   rl   rm   outputeler2   r2   r3   __call__   s0   
 
zPipeline.__call__c                 K   s
   i i |fS )a  
        this method should sanitize the keyword args to preprocessor params,
        forward params and postprocess params on '__call__' or '_process_single' method
        considered to be a normal classmethod with default implementation / output

        Default Returns:
            Dict[str, str]:  preprocess_params = {}
            Dict[str, str]:  forward_params = {}
            Dict[str, str]:  postprocess_params = pipeline_parameters
        Nr2   )r/   Zpipeline_parametersr2   r2   r3   rp      s   
zPipeline._sanitize_parametersc                 o   s*    |D ]}| j |g|R i |V  qd S r6   )rt   )r/   ri   rw   r1   ry   r2   r2   r3   rv      s   zPipeline._process_iteratorc                 C   s   t || jS r6   )
collate_fnr'   )r/   datar2   r2   r3   _collate_fn   s   zPipeline._collate_fnc              	   O   s   | di }| di }| di }| | | j|fi |}t| j| j= | jtjkrTt  | j	r;| 
|}| j|fi |}W d    n1 sNw   Y  n	| j|fi |}W d    n1 sgw   Y  | j|fi |}| | |S )Nrk   rl   rm   )rL   _check_input
preprocessr   rG   r.   r   rH   no_gradrK   r}   forwardpostprocess_check_output)r/   ri   rw   r1   rk   rl   rm   outr2   r2   r3   rt      s$   


	
zPipeline._process_singlec                 C   sv   i }|D ]}|  D ]\}}||g }|| |||< q
q| D ]}t|| d tjr8t|| ||< q#|S )Nr   )itemsrL   r7   keysr)   rH   Tensorcat)r/   Z	data_listZ
batch_dataZsample_preprocessedkvZ
value_listr2   r2   r3   _batch  s   

zPipeline._batchc              
      s  | d| d}| d}g }tdt||D ]}t|| t|}|| }	fdd||| D }
tjjG jtjkrot	  
|
}jrV|}j|fi |}W d    n1 siw   Y  n
|
}j|fi |}W d    n1 sw   Y  t|	D ]T i }| D ]8\}}|d urt|ttfrt|d tjrt| fdd|D ||< q|  ||< q|  d	  ||< qj|fi |}| || qq|S )
Nrk   rl   rm   r   c                    s   g | ]}j |fi  qS r2   )r   )r[   i)rk   r/   r2   r3   r]   %  s    z+Pipeline._process_batch.<locals>.<listcomp>c                 3   s     | ]}|  d   V  qdS )r"   Nr2   )r[   e)	batch_idxr2   r3   rd   ;  s
    
z*Pipeline._process_batch.<locals>.<genexpr>r"   )rL   ranger>   minr   rG   r.   r   rH   r   r   rK   r}   r   r   r)   tuplers   r   rq   r   r   r7   )r/   ri   rj   r1   rl   rm   Zoutput_listr   endZreal_batch_sizeZpreprocessed_listZbatched_outr   r   elementr2   )r   rk   r/   r3   ru     sR   









zPipeline._process_batchc           	      C   sh  | j }|tv rt| }t|trLd }|D ]}t|ttfr*t|t|kr)|} q4qt|tr3|} q4q|d u rJd}|D ]	}|| d7 }q<t||}t|trXt	|| d S t|trut|tsfJ dt
||D ].\}}t	|| qkt|tr| D ]}t|tr||v rt	|| ||  q~td| d S d S t| ddstd| d d	| _d S d S )
NzDinput data format for current pipeline should be one of following: 

zinput should be a tuplezinvalid input_type definition _input_has_warnedFtask z input definition is missingT)	group_keyr   r)   rs   dictr   rq   r*   
ValueErrorr   zipr   getattrr+   rh   r   )	r/   ri   	task_nameZ
input_typeZmatched_typeterr_msgZ	input_eler   r2   r2   r3   r~   I  sN   





zPipeline._check_inputc                 C   s   | j }|tvrt| ddstd| d d| _d S t| }g }t|ttfr,|	 n|}|D ]}t|ttfrB||vrB|
| q0t|dkrTtd| d| d	d S )
N_output_has_warnedFr   z output keys are missingTr   zexpected output keys are z, those z are missing)r   r   r   r+   rh   r   r)   r   r   r   r7   r>   r   )r/   ri   r   Zoutput_keysZmissing_keysr   r2   r2   r3   r   s  s,   


zPipeline._check_outputinputsc                 K   s8   | j dus	J dt| j trJ d| j |fi |S )z\ Provide default implementation based on preprocess_cfg and user can reimplement it
        Nz'preprocess method should be implementedzEdefault implementation does not support using multiple preprocessors.)r<   r)   r
   )r/   r   rk   r2   r2   r3   r     s
   zPipeline.preprocessc                 K   s2   | j dus	J d| jrJ d| j |fi |S )zU Provide default implementation using self.model and user can reimplement it
        Nz$forward method should be implementedzFdefault implementation does not support multiple models in a pipeline.)r0   r?   )r/   r   rl   r2   r2   r3   r     s   zPipeline.forwardc                 K   s   t d)ac   If current pipeline support model reuse, common postprocess
            code should be write here.

        Args:
            inputs:  input data
            post_params:   post process parameters

        Return:
            dict of results:  a dict containing outputs of model, each
                output should have the standard output name.
        r   N)NotImplementedError)r/   r   Zpost_paramsr2   r2   r3   r     s   zPipeline.postprocess)NNNr:   TN)rr   
__module____qualname____doc__r4   r
   
InputModelr9   r*   r   r   rP   r`   rF   Inputr   r   r	   rz   rp   rv   r}   rt   r   ru   r~   r   r   r   r   r2   r2   r2   r3   r%   +   sR    
C#
)


/*


r%   c                   @   s   e Zd ZdZ			ddedeeee f fddZdd	 Z	d
d Z
edd Zdeeef deeef fddZedd ZdedefddZdS )DistributedPipelinea  This pipeline is used to load multi gpu models.

    What will this class do:
    1. Read the global config from the configuration.json
    2. Set the multiprocessing method to spawn
    3. Open a multiprocessing pool of the world_size to instantiate model pieces.
    4. Set the master port and ip
    5. Call _instantiate_one to instantiate one model piece,
    This method should be implemented by the derived class.
    6. After the forward method is called, do preprocess in main process and
    call _forward_one to collect results, and do post process in main process.

    NOTE: _instantiate_one and _forward_one are class methods, any derived class should implement them and
    store the model handler in the class field.
    NTr0   r<   c           	      K   sP  || _ d| _t | _|| _tj|r|| _nt	|| _t
| j| _| | j| _d | _d| _t| j| _d| _| jj| _tjjddd tt| j}t| j| _d|vrZd|d< d|v rdt|d ntd	d
}ddlm}m} ||sy| }t ||d< |d tj!d< |d tj!d< | j"t#| j$j%fd| ji| jj&|| g | _'d S )NFcpuspawnT)forceZ	master_ipz	127.0.0.1master_porti<s  iL  r   )_find_free_port_is_free_portZMASTER_ADDRZMASTER_PORTrE   )(r<   rI   r   rJ   rK   rB   rC   existsrE   r   r   rA   _get_world_size
world_size
model_poolr.   r   r'   r?   rG   rH   multiprocessingZset_start_methodrs   r   r   intrandomrandintrW   r   r   r*   environmapr   	__class___instantiate_oner0   r8   )	r/   r0   r<   rO   r1   Zranksr   r   r   r2   r2   r3   rP     sX   


zDistributedPipeline.__init__c                 C   sB   t | dr| jd urz| j  W d S  ty   Y d S w d S d S )Nr   )rU   r   	terminateAttributeErrorrY   r2   r2   r3   __del__  s   zDistributedPipeline.__del__c                 C   s    | j  }|d= |d= |d= |S )Nr   r<   rJ   )__dict__copy)r/   Z	self_dictr2   r2   r3   __getstate__  s
   
z DistributedPipeline.__getstate__c                 K      dS )a  Instantiate one model piece.

        Args:
            rank: The model rank.
            model_dir: The model_dir in the node.
            kwargs: Any extra args.

        Returns:
            None. The model handler should be kept in the class field.
        Nr2   )clsZrankrE   r1   r2   r2   r3   r        z$DistributedPipeline._instantiate_oner   ra   c                 K   s,   ||d}| j | jj|g| j }|d S )N)r   rl   r   )r   r   r   _forward_oner   )r/   r   rl   resr2   r2   r3   r     s   
zDistributedPipeline.forwardc                 C   r   )zForward the inputs to one model piece.

        Use the model handler kept in the class field to forward.

        Args:
            inputs: The inputs after the preprocessing.

        Returns:
            The forward results.
        Nr2   )r   r   r2   r2   r3   r     r   z DistributedPipeline._forward_onerA   c                 C   s    | d}|d u r| dS |S )Nzmegatron.world_sizezmodel.world_size)Zsafe_get)r/   rA   Zm_world_sizer2   r2   r3   r     s   

z#DistributedPipeline._get_world_size)NNT)rr   r   r   r   r*   r   r   r
   rP   r   r   classmethodr   r   r   r   r   r   r   r   r2   r2   r2   r3   r     s(    
1




r   c              	      s8  ddl m} dd }t| tst| tr#t|  fdd|  D S t| ttfrRdt	| kr5t
g S t| d ttfrE||  S t|  fdd| D S t| tjri| jjtju ra| S tt
|  S t| t
jrt|  S t| ttttttd	fr| S || d
kr| S || dkr| S tdt|  )a3  Prepare the input just before the forward function.
    This method will move the tensors to the right device.
    Usually this method does not need to be overridden.

    Args:
        data: The data out of the dataloader.
        device: The device to move data to.

    Returns: The processed data.

    r   )default_collatec                 S   s   | j jS r6   )r   rr   )objr2   r2   r3   get_class_name2  s   z"collate_fn.<locals>.get_class_namec                    s(   i | ]\}}||d krt | n|qS )Z	img_metasr{   )r[   r   r   r'   r2   r3   
<dictcomp>7  s    zcollate_fn.<locals>.<dictcomp>c                 3   s    | ]}t | V  qd S r6   r   )r[   r   r   r2   r3   rd   A  s    zcollate_fn.<locals>.<genexpr>NZInputFeaturesZDataContainerzUnsupported data type )Ztorch.utils.data.dataloaderr   r)   r   r   rq   r   r   rs   r>   rH   r   r   floatrX   npZndarrayZdtypeZstr_r{   Z
from_numpybytesr*   boolr   )r|   r'   r   r   r2   r   r3   r{   $  s2   

r{   )FrB   os.pathrC   re   r   abcr   r   	functoolsr   r   r   	threadingr   typingr   r   r	   r
   r   r   Znumpyr   	packagingr   Zmodelscope.models.baser   Zmodelscope.msdatasetsr   Zmodelscope.outputsr   r   Zmodelscope.pipeline_inputsr   r   Zmodelscope.preprocessorsr   Zmodelscope.utils.configr   Zmodelscope.utils.constantr   r   r   Zmodelscope.utils.devicer   r   r   Zmodelscope.utils.hubr   r   Zmodelscope.utils.import_utilsr   r   Zmodelscope.utils.loggerr    rW   r!   utilr#   r$   rH   r   r*   r   r   r   r+   r%   r   r{   r2   r2   r2   r3   <module>   sH      ~}