#!/usr/bin/env python """A simple example demonstrating calling an async function from a cmd2 app.""" import asyncio import concurrent.futures import threading import cmd2 _event_loop = None _event_lock = threading.Lock() def run_async(coro) -> concurrent.futures.Future: """Await a coroutine from a synchronous function/method.""" global _event_loop # noqa: PLW0603 if _event_loop is None: with _event_lock: if _event_loop is None: _event_loop = asyncio.new_event_loop() thread = threading.Thread( target=_event_loop.run_forever, name="Async Runner", daemon=True, ) thread.start() return asyncio.run_coroutine_threadsafe(coro, _event_loop) async def async_wait(duration: float) -> float: """Example async function that is called from a synchronous cmd2 command.""" await asyncio.sleep(duration) return duration class AsyncCallExample(cmd2.Cmd): """A simple cmd2 application. Demonstrates how to run an async function from a cmd2 command. """ def do_async_wait(self, _: str) -> None: """Waits asynchronously. Example cmd2 command that calls an async function. """ waitable = run_async(async_wait(0.1)) self.poutput("Begin waiting...") # Wait for coroutine to complete and get its return value: res = waitable.result() self.poutput(f"Done waiting: {res}") return def do_hello_world(self, _: str) -> None: """Prints a simple greeting. Just a typical (synchronous) cmd2 command """ self.poutput("Hello World") def main() -> int: """Main entry point for the example.""" app = AsyncCallExample() app.set_window_title("Call to an Async Function Test") return app.cmdloop() if __name__ == "__main__": import sys sys.exit(main())