o Эjy+@sddlmZddlZddlZddlZddlmZmZddlm Z m Z m Z m Z m Z mZddlmZddlmZmZmZmZddlmZd gZed Ze d ZGd d d e eZGdd d ZdS)) annotationsN) AsyncIteratorIterable)AnyCallableGenericLiteralTypeVaroverload)ConcurrencyError) OP_BINARYOP_CONTOP_TEXTFrame)Data Assemblerzutf-8Tc@sNeZdZdZdddZdddZdd d ZddddZdddZdddZ dS) SimpleQueuez Simplified version of :class:`asyncio.Queue`. Provides only the subset of functionality needed by :class:`Assembler`. returnNonecCst|_d|_t|_dSN)asyncioget_running_looploop get_waiter collectionsdequequeueselfr!X/var/www/html/CropPilot/venv/lib/python3.10/site-packages/websockets/asyncio/messages.py__init__s zSimpleQueue.__init__intcCs t|jSr)lenrrr!r!r"__len__"s zSimpleQueue.__len__itemrcCs8|j||jdur|js|jddSdSdS)zPut an item into the queue.N)rappendrdone set_result)r r'r!r!r"put%s zSimpleQueue.putTblockboolcsp|js3|s td|jdusJd|j|_z|jIdHW|jd|_n |jd|_w|jS)z?Remove and return an item from the queue, waiting if necessary.stream of frames endedNzcannot call get() concurrently)rEOFErrorrr create_futurecancelpopleft)r r,r!r!r"get+s    zSimpleQueue.getitems Iterable[T]cCs0|jdus Jd|jrJd|j|dS)z)Put back items into an empty, idle queue.Nz%cannot reset() while get() is runningz&cannot reset() while queue isn't empty)rrextend)r r4r!r!r"reset9szSimpleQueue.resetcCs0|jdur|js|jtddSdSdS)z8Close the queue, raising EOFError in get() if necessary.Nr.)rr) set_exceptionr/rr!r!r"abort?szSimpleQueue.abortNrr)rr$)r'rrr)T)r,r-rr)r4r5rr) __name__ __module__ __qualname____doc__r#r&r+r3r7r9r!r!r!r"rs     rc@seZdZdZddddddfd.ddZed/ddZed0ddZed1d2ddZd1d2ddZed3ddZed4d dZed1d5d"dZd1d5d#dZd6d&d'Zd7d(d)Z d7d*d+Z d7d,d-Z dS)8ra Assemble messages from frames. :class:`Assembler` expects only data frames. The stream of frames must respect the protocol; if it doesn't, the behavior is undefined. Args: pause: Called when the buffer of frames goes above the high water mark; should pause reading from the network. resume: Called when the buffer of frames goes below the low water mark; should resume reading from the network. NcCdSrr!r!r!r!r"XzAssembler.cCr?rr!r!r!r!r"r@YrAhigh int | NonelowpauseCallable[[], Any]resumerrcCst|_|dur|dur|d}|dur|dur|d}|dur4|dur4|dkr,td||kr4td|||_|_||_||_d|_d|_d|_ dS)Nrz%low must be positive or equal to zeroz)high must be greater than or equal to lowF) rframes ValueErrorrBrDrErGpausedget_in_progressclosed)r rBrDrErGr!r!r"r#Ts  zAssembler.__init__decode Literal[True]strcdSrr!r rNr!r!r"r3vz Assembler.getLiteral[False]bytescrQrr!rRr!r!r"r3yrS bool | NonercrQrr!rRr!r!r"r3|rScs|jrtdd|_z_|j|j IdH}||jtus'|jtus'J|dur0|jtu}|g}|j sfz |j|j IdH}Wnt j yR|j |w||jt us^J|||j r6Wd|_nd|_wddd|D}|r|S|S)a0 Read the next message. :meth:`get` returns a single :class:`str` or :class:`bytes`. If the message is fragmented, :meth:`get` waits until the last frame is received, then it reassembles the message and returns it. To receive messages frame by frame, use :meth:`get_iter` instead. Args: decode: :obj:`False` disables UTF-8 decoding of text frames and returns :class:`bytes`. :obj:`True` forces UTF-8 decoding of binary frames and returns :class:`str`. Raises: EOFError: If the stream of frames has ended. UnicodeDecodeError: If a text frame contains invalid UTF-8. ConcurrencyError: If two coroutines run :meth:`get` or :meth:`get_iter` concurrently. &get() or get_iter() is already runningTNFcss|]}|jVqdSr)data).0framer!r!r" sz Assembler.get..)rLr rIr3rM maybe_resumeopcoderr finrCancelledErrorr7rr(joinrN)r rNr[rIrYr!r!r"r3s8    AsyncIterator[str]cCr?rr!rRr!r!r"get_iterzAssembler.get_iterAsyncIterator[bytes]cCr?rr!rRr!r!r"rcrdAsyncIterator[Data]cCr?rr!rRr!r!r"rcrdcCs |jrtdd|_z |j|j IdH}Wn tjy$d|_w||jt us5|jt us5J|dur>|jt u}|rMt }| |j |jVnt|j V|js|j|j IdH}||jtuslJ|rx| |j |jVnt|j V|jrVd|_dS)a Stream the next message. Iterating the return value of :meth:`get_iter` asynchronously yields a :class:`str` or :class:`bytes` for each frame in the message. The iterator must be fully consumed before calling :meth:`get_iter` or :meth:`get` again. Else, :exc:`ConcurrencyError` is raised. This method only makes sense for fragmented messages. If messages aren't fragmented, use :meth:`get` instead. Args: decode: :obj:`False` disables UTF-8 decoding of text frames and returns :class:`bytes`. :obj:`True` forces UTF-8 decoding of binary frames and returns :class:`str`. Raises: EOFError: If the stream of frames has ended. UnicodeDecodeError: If a text frame contains invalid UTF-8. ConcurrencyError: If two coroutines run :meth:`get` or :meth:`get_iter` concurrently. rWTNF)rLr rIr3rMrr`r]r^rr UTF8DecoderrNrYr_rUr)r rNr[decoderr!r!r"rcs6     r[rcCs&|jrtd|j||dS)z Add ``frame`` to the next message. Raises: EOFError: If the stream of frames has ended. r.N)rMr/rIr+ maybe_pause)r r[r!r!r"r+ s  z Assembler.putcCs>|jdurdSt|j|jkr|jsd|_|dSdSdS)z7Pause the writer if queue is above the high water mark.NT)rBr%rIrKrErr!r!r"ri  zAssembler.maybe_pausecCs>|jdurdSt|j|jkr|jrd|_|dSdSdS)z7Resume the writer if queue is below the low water mark.NF)rDr%rIrKrGrr!r!r"r]#rjzAssembler.maybe_resumecCs|jrdSd|_|jdS)z End the stream of frames. Calling :meth:`close` concurrently with :meth:`get`, :meth:`get_iter`, or :meth:`put` is safe. They will raise :exc:`EOFError`. NT)rMrIr9rr!r!r"close.szAssembler.close) rBrCrDrCrErFrGrFrr)rNrOrrP)rNrTrrUr)rNrVrr)rNrOrrb)rNrTrre)rNrVrrf)r[rrrr:) r;r<r=r>r#r r3rcr+rir]rkr!r!r!r"rEs2 "   =    E   ) __future__rrcodecsrcollections.abcrrtypingrrrrr r exceptionsr rIr rrrr__all__getincrementaldecoderrgrrrr!r!r!r"s     0