|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +import asyncio |
| 4 | +import contextlib |
| 5 | +import typing |
| 6 | +import webbrowser |
| 7 | + |
| 8 | +import anyio |
| 9 | +import click |
| 10 | +from jumpstarter_driver_composite.client import CompositeClient |
| 11 | +from jumpstarter_driver_network.adapters.novnc import NovncAdapter |
| 12 | + |
| 13 | +from jumpstarter.client.decorators import driver_click_group |
| 14 | + |
| 15 | +if typing.TYPE_CHECKING: |
| 16 | + from jumpstarter_driver_network.client import TCPClient |
| 17 | + |
| 18 | + |
| 19 | +class VNClient(CompositeClient): |
| 20 | + """Client for interacting with a VNC server.""" |
| 21 | + |
| 22 | + @property |
| 23 | + def tcp(self) -> TCPClient: |
| 24 | + """Get the TCP client.""" |
| 25 | + return self.children["tcp"] |
| 26 | + |
| 27 | + @contextlib.contextmanager |
| 28 | + def session(self) -> typing.Iterator[str]: |
| 29 | + """Create a new VNC session.""" |
| 30 | + with NovncAdapter(client=self.tcp, method="connect") as adapter: |
| 31 | + yield adapter |
| 32 | + |
| 33 | + def cli(self) -> click.Command: |
| 34 | + """Return a click command handler for this driver.""" |
| 35 | + |
| 36 | + @driver_click_group(self) |
| 37 | + def vnc(): |
| 38 | + """Open a VNC session.""" |
| 39 | + |
| 40 | + @vnc.command() |
| 41 | + @click.option("--browser/--no-browser", default=True, help="Open the session in a web browser.") |
| 42 | + def session(browser: bool): |
| 43 | + """Open a VNC session.""" |
| 44 | + # The NovncAdapter is a blocking context manager that runs in a thread. |
| 45 | + # We can enter it, open the browser, and then just wait for the user |
| 46 | + # to press Ctrl+C to exit. The adapter handles the background work. |
| 47 | + with self.session() as url: |
| 48 | + click.echo(f"To connect, please visit: {url}") |
| 49 | + if browser: |
| 50 | + webbrowser.open(url) |
| 51 | + click.echo("Press Ctrl+C to close the VNC session.") |
| 52 | + try: |
| 53 | + # Use the client's own portal to wait for cancellation. |
| 54 | + self.portal.call(asyncio.Event().wait) |
| 55 | + except (KeyboardInterrupt, anyio.get_cancelled_exc_class()): |
| 56 | + click.echo("\nClosing VNC session.") |
| 57 | + |
| 58 | + return vnc |
0 commit comments