cua-computer

import asyncio from cua_computer import Computer, Context, State, Task, TaskContext from typing import Dict, Any # Define a simple application state class MyState(State): count: int = 0 message: str = "Hello Cua!" # Define a task that increments the count class IncrementTask(Task): async def run(self, context: TaskContext[MyState]): await context.write("Incrementing count...") context.state.count += 1 await context.sleep(1) # Simulate some work await context.write(f"Count is now: {context.state.count}") context.exit() # Task completes # Define a task that displays the current state and offers an action class DisplayTask(Task): async def run(self, context: TaskContext[MyState]): await context.write(f"Current State: {context.state.message} (Count: {context.state.count})") await context.write("Press 'i' to increment, 'q' to quit.") while True: key = await context.read_key() if key == 'i': await context.spawn(IncrementTask()) elif key == 'q': context.exit() break else: await context.write("Unknown command. Press 'i' or 'q'.") async def main(): initial_state = MyState() computer = Computer(initial_state) # Run the main display task await computer.run(DisplayTask()) await computer.exit() # Ensure cleanup after tasks are done if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: print("\nExiting Cua application.")