o j @sdZddlmZddlZddlZddlmZddlmZmZddl m Z m Z m Z ddl Z ddlmZmZddlmZdd lmZmZdd lmZd d gZGd d d e jZGdd d e jZdS)zI Utility for running a prompt_toolkit application in an asyncssh server. ) annotationsN)get_running_loop)Callable Coroutine)AnyTextIOcast) AppSessioncreate_app_session)Size) PipeInputcreate_pipe_input) Vt100_OutputPromptToolkitSSHSessionPromptToolkitSSHServerc@s\eZdZd$ddZd%d d Zd&ddZd'ddZd(ddZd(ddZd)ddZ d*d!d"Z d#S)+rinteract>Callable[[PromptToolkitSSHSession], Coroutine[Any, Any, None]] enable_cprboolreturnNonecsN|_|_d_d_d_d_d_Gfddd}tt|_ dS)Ncs@eZdZdfdd Zddd Zdd d Zedfd d ZdS)z0PromptToolkitSSHSession.__init__..Stdoutdatastrrrcs@zjdurj|ddWdSWdStyYdSw)N z )_chanwritereplaceBrokenPipeError)srself/root/aizidognhua/tmp/workspace/projects/ec89d86c-575f-41c9-af57-ac45cbdbf775/venv/lib/python3.10/site-packages/prompt_toolkit/contrib/ssh/server.pyr-s  z6PromptToolkitSSHSession.__init__..Stdout.writercSdSNTr!rr!r!r"isatty4z7PromptToolkitSSHSession.__init__..Stdout.isattycSsdSNr!r%r!r!r"flush7r'z6PromptToolkitSSHSession.__init__..Stdout.flushcs"jdusJtjjdS)Nr)rr _orig_chan get_encodingr%rr!r"encoding:sz9PromptToolkitSSHSession.__init__..Stdout.encodingN)rrrrrrrr)rr)__name__ __module__ __qualname__rr&r)propertyr,r!rr!r"Stdout,s   r3) rr interact_taskr app_session_input_outputrrstdout)r rrr3r!rr"__init__sz PromptToolkitSSHSession.__init__r cCs4|jdur tdddS|j\}}}}t||dS)zU Callable that returns the current `Size`, required by Vt100_Output. NO)rowscolumns)rr get_terminal_sizer widthheightpixwidth pixheightr!r!r" _get_sizeAs   z!PromptToolkitSSHSession._get_sizechanrcCs ||_dSr()r)r rEr!r!r"connection_madeKs z'PromptToolkitSSHSession.connection_madecCr#r$r!rr!r!r"shell_requestedNr'z'PromptToolkitSSHSession.shell_requestedcCst||_dSr()r create_task _interactr4rr!r!r"session_startedQsz'PromptToolkitSSHSession.session_startedc s.|jdur tdt|jdr|jjdur|jd|j}t|j|j||j d|_ t ]|_ t |j |j d;}||_z$z ||IdHWn tyVtYnwW|j|j n |j|j wWdn1sxwYWddSWddS1swYdS)Nz,`_interact` called before `connection_made`. set_line_modeF)termr)inputoutput)r Exceptionhasattr_editorrKget_terminal_typerr8rDrr7r r6r r5r BaseException traceback print_excclose)r rLsessionr!r!r"rITs4         "z!PromptToolkitSSHSession._interactr@intrArBobjectrCcCs&|jr|jjr|jjdSdSdSr()r5app _on_resizer?r!r!r"terminal_size_changedpsz-PromptToolkitSSHSession.terminal_size_changedrrdatatypecCs|jdurdS|j|dSr()r6 send_text)r rr]r!r!r" data_receivedws z%PromptToolkitSSHSession.data_receivedNrrrrrr)rr )rErrrr-r.) r@rXrArXrBrYrCrYrr)rrr]rYrr) r/r0r1r9rDrFrGrJrIr\r_r!r!r!r"rs  )    c@s4eZdZdZdddd d ZdddZdddZdS)raZ Run a prompt_toolkit application over an asyncssh server. This takes one argument, an `interact` function, which is called for each connection. This should be an asynchronous function that runs the prompt_toolkit applications. This function runs in an `AppSession`, which means that we can have multiple UI interactions concurrently. Example usage: .. code:: python async def interact(ssh_session: PromptToolkitSSHSession) -> None: await yes_no_dialog("my title", "my text").run_async() prompt_session = PromptSession() text = await prompt_session.prompt_async("Type something: ") print_formatted_text('You said: ', text) server = PromptToolkitSSHServer(interact=interact) loop = get_running_loop() loop.run_until_complete( asyncssh.create_server( lambda: MySSHServer(interact), "", port, server_host_keys=["/etc/ssh/..."], ) ) loop.run_forever() :param enable_cpr: When `True`, the default, try to detect whether the SSH client runs in a terminal that responds to "cursor position requests". That way, we can properly determine how much space there is available for the UI (especially for drop down menus) to render. TrrrrrrrcCs||_||_dSr()rr)r rrr!r!r"r9s zPromptToolkitSSHServer.__init__usernamercCr#)NFr!)r rbr!r!r" begin_authsz!PromptToolkitSSHServer.begin_authrcCst|j|jdS)Nra)rrrrr!r!r"session_requestedsz(PromptToolkitSSHServer.session_requestedNr`)rbrrr)rr)r/r0r1__doc__r9rcrdr!r!r!r"rs ) )re __future__rasynciorTrcollections.abcrrtypingrrrasyncssh"prompt_toolkit.application.currentr r prompt_toolkit.data_structuresr prompt_toolkit.inputr r prompt_toolkit.output.vt100r__all__SSHServerSessionr SSHServerrr!r!r!r"s    h