o ٷi:/@sdZddlZddlZddlmZmZddlmZddlm Z ddl m Z ddl m Z ddlmZmZdd lmZdd lmZdd lmZmZdd lmZdd lmZe eZGdddZdS)z Async entrypoint for vLLM-Omni diffusion model inference. Provides an asynchronous interface for running diffusion models, enabling concurrent request handling and streaming generation. N)AsyncGeneratorIterable)ThreadPoolExecutor)Any) init_logger)get_hf_file_to_dict)OmniDiffusionConfigTransformerConfig)DiffusionEngine)OmniDiffusionRequest)OmniDiffusionSamplingParamsOmniPromptType) LoRARequest)OmniRequestOutputc @sNeZdZdZ d,dededBdefddZ  d-ded e d edBd e dBd e f d dZ d,ded edBded e e dffddZd.ddZd.ddZd.ddZd eeeBd dfddZed efddZed efddZded efddZd e d efd d!Zd eefd"d#Zd$ed efd%d&Zd,d'edBd dfd(d)Zd efd*d+Z dS)/AsyncOmniDiffusionaAsync entry point for vLLM-Omni diffusion model inference. This class provides an asynchronous interface for running diffusion models, enabling concurrent request handling. It wraps the DiffusionEngine and provides async methods for image generation. Args: model: Model name or path to load od_config: Optional OmniDiffusionConfig. If not provided, it will be created from kwargs **kwargs: Additional keyword arguments passed to OmniDiffusionConfig Example: >>> async_diffusion = AsyncOmniDiffusion(model="Qwen/Qwen-Image") >>> result = await async_diffusion.generate( ... prompt="A beautiful sunset over the ocean", ... request_id="req-1", ... ) >>> print(result.images) Nmodel od_configkwargsc Ks||_|d}|d}|durtjdd|i|}nt|tr;|dur*|d}|dur3|d}tjdi|}||_|durJ|jjd||durV|jjd|z(t d|j}|dury|dd|_ | t d|j}t ||_ntdWncttttfyt d|j}|durtd |j|d } |d pg} | d ksd | vrd|_ t |_| n$| dkr|j durd|_ t |_| n| rt| dkr| d|_ nYnwt||_tdd|_d|_td|dS)Nstage_idengine_input_sourcerzmodel_index.json _class_nameztransformer/config.jsonzmodel_index.json not foundz config.jsonz9Could not find config.json or model_index.json for model model_type architecturesbagelBagelForConditionalGeneration BagelPipelinenextstepNextStep11Pipeliner) max_workersFz-AsyncOmniDiffusion initialized with model: %s)rgetr from_kwargs isinstancedictromni_kv_config setdefaultrmodel_class_nameupdate_multimodal_supportr from_dicttf_model_configFileNotFoundErrorAttributeErrorOSError ValueErrorlenr make_engineenginer _executor_closedloggerinfo) selfrrrrr config_dicttf_config_dictcfgrrr r ^/home/ubuntu/.local/lib/python3.10/site-packages/vllm_omni/entrypoints/async_omni_diffusion.py__init__4sb               zAsyncOmniDiffusion.__init__promptsampling_params request_id lora_requestreturnc s|durdtjdd}|jrd|_|dur||_t|g||gd}td|t }z| |j |j j|IdH}|d}Wnty_}ztd||td ||d}~ww|jsf||_|S) aGenerate images asynchronously from a text prompt. Args: prompt: Text prompt describing the desired image sampling_params: Sampling parameters request_id: Optional unique identifier for tracking the request Returns: OmniRequestOutput containing generated images Raises: RuntimeError: If generation fails Nzdiff-T)promptsr= request_idsz"Starting generation for request %srz$Generation failed for request %s: %szDiffusion generation failed: )uuiduuid4hexguidance_scaleguidance_scale_providedr?r r4debugasyncioget_event_looprun_in_executorr2r1step Exceptionerror RuntimeErrorr>) r6r<r=r>r?requestloopresulter r r:generates:  zAsyncOmniDiffusion.generatecKs(|jd||d|IdH}|VdS)a(Generate images with streaming progress updates. Currently, diffusion models don't support true streaming, so this yields a single result after generation completes. Future implementations may support step-by-step progress updates. Args: prompt: Text prompt describing the desired image request_id: Optional unique identifier for tracking the request **kwargs: Additional generation parameters Yields: OmniRequestOutput with generation progress/results )r<r>Nr )rU)r6r<r>rrSr r r:generate_streams z"AsyncOmniDiffusion.generate_streamc Cs|jrdSd|_z|jWnty'}z td|WYd}~nd}~wwz |jjddWntyI}z td|WYd}~nd}~wwtddS)z{Close the engine and release resources. Should be called when done using the AsyncOmniDiffusion instance. NTz"Error closing diffusion engine: %sF)waitz Error shutting down executor: %szAsyncOmniDiffusion closed) r3r1closerNr4warningr2shutdownr5)r6rTr r r:rXs zAsyncOmniDiffusion.closecCs |dS)zAlias for close() method.N)rXr6r r r:rZs zAsyncOmniDiffusion.shutdowncCs$z|WdStyYdSw)z Best-effort cleanup on deletion.N)rXrNr[r r r:__del__s  zAsyncOmniDiffusion.__del__cs|j|dS)zAbort a request.N)r1abort)r6r>r r r:r]szAsyncOmniDiffusion.abortcCs|j S)zCheck if the engine is running.r3r[r r r: is_runningszAsyncOmniDiffusion.is_runningcCs|jS)zCheck if the engine is stopped.r^r[r r r: is_stoppedszAsyncOmniDiffusion.is_stopped adapter_idc sDt}||j|jjdd|fidIdH}t|tr t|S|S)z Remove a LoRA remove_loraN rJrKrLr2r1collective_rpcr#listall)r6rarRresultsr r r:rbs  zAsyncOmniDiffusion.remove_lorac Ft}||j|jjdddd|idIdH}t|tr!t|S|S)zAdd a LoRA adapteradd_loraNr r?rc)r6r?rRrgr r r:ri   zAsyncOmniDiffusion.add_lorac sdt}||j|jjdddidIdH}t|ts|pgSt}|D] }| |p+gq$t |S)z%List all registered LoRA adapter IDs. list_lorasNr ) rJrKrLr2r1rdr#resetupdatesorted)r6rRrgmergedpartr r r:rks" zAsyncOmniDiffusion.list_loraslora_idc rh)z&Prevent an adapter from being evicted.pin_loraNr rarc)r6rqrRrgr r r:rr,rjzAsyncOmniDiffusion.pin_loratrace_filenamecs(t}||j|jj|IdHdS)zStart profiling for the diffusion model. Args: trace_filename: Optional base filename for trace files. If None, a timestamp-based name will be generated. N)rJrKrLr2r1 start_profile)r6rsrRr r r:rt:sz AsyncOmniDiffusion.start_profilecs"t}||j|jjIdHS)zStop profiling and return profiling results. Returns: Dictionary containing paths to trace and table files. N)rJrKrLr2r1 stop_profile)r6rRr r r:ruHs  zAsyncOmniDiffusion.stop_profile)N)NN)r@N)!__name__ __module__ __qualname____doc__strrrr;r r rrrUrrVrXrZr\rr]propertyboolr_r`intrbrirerkrrrtr$rur r r r:rs^ O ;    r)ryrJrDcollections.abcrrconcurrent.futuresrtypingr vllm.loggerrvllm.transformers_utils.configrvllm_omni.diffusion.datarr $vllm_omni.diffusion.diffusion_enginer vllm_omni.diffusion.requestr vllm_omni.inputs.datar r vllm_omni.lora.requestrvllm_omni.outputsrrvr4rr r r r:s