a Q\a@sdZddlZddlZddlmZmZmZmZmZm Z ddl Z ddl m Z m Z ddlmZddlmZddlmZdd gZGd dde jZGd d d e jZdS) zI Utility for running a prompt_toolkit application in an asyncssh server. N)Any AwaitableCallableOptionalTextIOcast) AppSessioncreate_app_session)Size)create_pipe_input) Vt100_OutputPromptToolkitSSHSessionPromptToolkitSSHServerc@seZdZedgedfddddZedddZeddd d Z e dd d Z ddd dZ ddddZ eeeeddddZeeddddZdS)r NinteractreturncsJ|_d_d_d_t_d_Gfddd}tt|_ dS)NcsReZdZeddfdd ZedddZdddd Zeedfd d Z dS) z0PromptToolkitSSHSession.__init__..StdoutN)datarcs:z"jdur j|ddWnty4Yn0dS)N z )_chanwritereplaceBrokenPipeError)srself@lib/python3.9/site-packages/prompt_toolkit/contrib/ssh/server.pyr$s   z6PromptToolkitSSHSession.__init__..Stdout.writercSsdSNTrrrrrisatty+sz7PromptToolkitSSHSession.__init__..Stdout.isattycSsdSNrrrrrflush.sz6PromptToolkitSSHSession.__init__..Stdout.flushcs"jdusJtjjdS)Nr)rstrZ _orig_chanZ get_encodingrrrrencoding1sz9PromptToolkitSSHSession.__init__..Stdout.encoding) __name__ __module__ __qualname__r#rboolr r"propertyr$rrrrStdout#s r*) r interact_taskr app_sessionr _input_outputrrstdout)rrr*rrr__init__sz PromptToolkitSSHSession.__init__rcCs8|jdurtdddS|j\}}}}t||dSdS)zU Callable that returns the current `Size`, required by Vt100_Output. NO)Zrowscolumns)rr get_terminal_sizerwidthheightpixwidth pixheightrrr _get_size8s  z!PromptToolkitSSHSession._get_size)chanrcCs ||_dSr!)r)rr;rrrconnection_madeBsz'PromptToolkitSSHSession.connection_madecCsdSrrrrrrshell_requestedEsz'PromptToolkitSSHSession.shell_requestedcCst||_dSr!)asyncioZget_event_loopZ create_task _interactr+rrrrsession_startedHsz'PromptToolkitSSHSession.session_startedc s|jdurtdt|jdr6|jjdur6|jd|j}t|j|j|dd|_ t |j |j dv}||_ zHz| |IdHWntytYn0W|j|j n|j|j 0Wdn1s0YdS)Nz,`_interact` called before `connection_made`. set_line_modeF)termZ write_binary)inputoutput)r ExceptionhasattrZ_editorrAZget_terminal_typer r/r:r.r r-r,r BaseException traceback print_excclose)rrBZsessionrrrr?Ks$       z!PromptToolkitSSHSession._interact)r6r7r8r9rcCs|jr|jjr|jjdSr!)r,ZappZ _on_resizer5rrrterminal_size_changedesz-PromptToolkitSSHSession.terminal_size_changed)rdatatypercCs|j|dSr!)r-Z send_text)rrrLrrr data_receivedlsz%PromptToolkitSSHSession.data_received)r%r&r'rrr0r r:rr<r(r=r@r?intobjectrKr#rMrrrrr s %  c@sLeZdZdZeegedfddddZee dddZ ed d d Z dS) ra, 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_event_loop() loop.run_until_complete( asyncssh.create_server( lambda: MySSHServer(interact), "", port, server_host_keys=["/etc/ssh/..."], ) ) loop.run_forever() NrcCs ||_dSr!)r)rrrrrr0szPromptToolkitSSHServer.__init__)usernamercCsdS)NFr)rrPrrr begin_authsz!PromptToolkitSSHServer.begin_authrcCs t|jSr!)r rrrrrsession_requestedsz(PromptToolkitSSHServer.session_requested) r%r&r'__doc__rr rr0r#r(rQrRrrrrrps ! )rSr>rHtypingrrrrrrZasyncsshZ"prompt_toolkit.application.currentrr Zprompt_toolkit.data_structuresr Zprompt_toolkit.inputr Zprompt_toolkit.output.vt100r __all__ZSSHServerSessionr Z SSHServerrrrrrs    ^