o wi@s`ddlZddlZddlZddlTddlmZddlmZmZm Z GdddZ Gdd d Z dS) N)*) InferResult)get_cancelled_errorget_error_grpc raise_errorc@sJeZdZdZddZddZdddZd d Zd d Zd dZ ddZ dS) _InferStreamaSupports sending inference requests and receiving corresponding requests on a gRPC bi-directional stream. Parameters ---------- callback : function Python function that is invoked upon receiving response from the underlying stream. The function must reserve the last two arguments (result, error) to hold :py:class:`InferResult` and :py:class:`InferenceServerException` objects respectively which will be provided to the function when executing the callback. The ownership of these objects will be given to the user. The 'error' would be None for a successful inference. verbose : bool Enables verbose mode if set True. cCs2||_||_t|_d|_d|_d|_d|_dS)NFT) _callback_verbosequeueQueue_request_queue_handler _cancelled_active_response_iterator)selfcallbackverboser\/home/ubuntu/sommelier/.venv/lib/python3.10/site-packages/tritonclient/grpc/_infer_stream.py__init__9s  z_InferStream.__init__cCs|jdddS)NT)cancel_requests)closerrrr__del__Bsz_InferStream.__del__FcCsf|r |jr |jd|_|jdur1|js|jd|jr,|j|jr,t dd|_dSdS)aGracefully close underlying gRPC streams. If cancel_requests is set True, then client cancels all the pending requests and closes the stream. If set False, the call blocks till all the pending requests on the stream are processed. TNzstream stopped...) rcancelrrr putis_alivejoinr print)rrrrrrEs       z_InferStream.closecCsH||_|jdur tdtj|jd|_|j|jr"tddSdS)zInitializes the handler to process the response from stream and execute the callbacks. Parameters ---------- response_iterator : iterator The iterator over the gRPC response stream. Nz7Attempted to initialize already initialized InferStream)targetzstream started...) rrr threadingThread_process_responsestartr r )rresponse_iteratorrrr _init_handlerYs   z_InferStream._init_handlercCs"|jr |j|dStddS)zEnqueues the specified request object to be provided in gRPC request stream. Parameters ---------- request : ModelInferRequest The protobuf message holding the ModelInferRequest zThe stream is no longer in valid state, the error detail is reported through provided callback. A new stream should be started after stopping the current stream.N)rr rrrrequestrrr_enqueue_requestls  z_InferStream._enqueue_requestcCs|j}|S)aYReturns the request details in the order they were added. The call to this function will block until the requests are available in the queue. InferStream._enqueue_request adds the request to the queue. Returns ------- protobuf message The ModelInferRequest protobuf message. )r getr(rrr _get_requests z_InferStream._get_requestc Csz,|jD]%}|jr t|d}}|jdkrt|jd}nt|j}|j||dqWdStj y`}z'|j |_ | tj jkrJt|}nt|}|jd|dWYd}~dSd}~ww)ztWorker thread function to iterate through the response stream and executes the provided callbacks. N)msg)resulterror)rr r error_messageInferenceServerExceptionrinfer_responser grpcRpcError is_activercode StatusCode CANCELLEDrdetailsr)rresponser/r0 rpc_errorrrrr$s$    z_InferStream._process_responseN)F) __name__ __module__ __qualname____doc__rrrr'r*r,r$rrrrr's  rc@s(eZdZdZddZddZddZdS) _RequestIteratorzAn iterator class to provide data to gRPC request stream. Parameters ---------- stream : InferStream The InferStream that holds the context to an active stream. cCs ||_dSN)_stream)rstreamrrrrs z_RequestIterator.__init__cCs|SrBrrrrr__iter__sz_RequestIterator.__iter__cCs|j}|dur t|SrB)rCr, StopIterationr(rrr__next__s z_RequestIterator.__next__N)r=r>r?r@rrErGrrrrrAs   rA) r r"r4tritonclient.utils _infer_resultr_utilsrrrrrArrrrs