407 lines
14 KiB
Python
407 lines
14 KiB
Python
import logging
|
||
import time
|
||
from typing import Optional
|
||
|
||
import marvin
|
||
import pendulum
|
||
import pyperclip
|
||
from textual.app import App, ComposeResult
|
||
from textual.containers import Container, Horizontal, VerticalScroll
|
||
from textual.events import Enter, Leave
|
||
from textual.logging import TextualHandler
|
||
from textual.message import Message
|
||
from textual.reactive import reactive
|
||
from textual.screen import ModalScreen, Screen
|
||
from textual.widgets import Button, Input, Label, Markdown
|
||
|
||
from marvin_sw_text_adventure.console import console
|
||
from marvin_sw_text_adventure.game import (
|
||
Game,
|
||
StarWarsCharacter,
|
||
create_game,
|
||
did_complete_mission,
|
||
get_next_mission,
|
||
)
|
||
|
||
logging.basicConfig(
|
||
level="NOTSET",
|
||
handlers=[TextualHandler()],
|
||
)
|
||
|
||
|
||
class ResponseHover(Message):
|
||
pass
|
||
|
||
|
||
class ResponseBody(Markdown):
|
||
text: str = ""
|
||
|
||
def update(self, markdown: str):
|
||
self.text = markdown
|
||
super().update(markdown)
|
||
|
||
def on_enter(self):
|
||
self.post_message(ResponseHover())
|
||
|
||
|
||
class Response(Container):
|
||
body = None
|
||
stream_finished: bool = False
|
||
|
||
def __init__(self, message: marvin.models.threads.Message, **kwargs) -> None:
|
||
classes = kwargs.setdefault("classes", "")
|
||
kwargs["classes"] = f"{classes} response".strip()
|
||
self.message = message
|
||
super().__init__(**kwargs)
|
||
|
||
def compose(self) -> ComposeResult:
|
||
self.body = ResponseBody(self.message.content, classes="response-body markdown")
|
||
self.body.border_title = (
|
||
"You" if self.message.role == "user" else self.message.name
|
||
)
|
||
self.body.border_subtitle = (
|
||
pendulum.instance(self.message.timestamp).in_tz("local").format("h:mm:ss A")
|
||
)
|
||
self.border_title = "Edit message"
|
||
|
||
yield self.body
|
||
with Horizontal(classes="edit-buttons-container hidden"):
|
||
yield Button("Copy", variant="default", id="copy-message")
|
||
yield Button("Delete", variant="error", id="delete-message")
|
||
|
||
def on_click(self):
|
||
self.action_toggle_buttons()
|
||
|
||
def on_response_hover(self, event: ResponseHover):
|
||
"""
|
||
This is an "enter" event bubbled up from the ResponseBody, since the
|
||
default "Leave" is triggered when hovering on child widgets. This keeps
|
||
the hover class even when hovering on the child widget.
|
||
"""
|
||
self.add_class("response-hover")
|
||
|
||
def on_enter(self, event: Enter):
|
||
self.add_class("response-hover")
|
||
|
||
def on_leave(self, event: Leave):
|
||
self.remove_class("response-hover")
|
||
|
||
def action_toggle_buttons(self):
|
||
self.toggle_class("show-edit-buttons")
|
||
self.body.toggle_class("show-edit-buttons")
|
||
self.query_one(".edit-buttons-container").toggle_class("hidden")
|
||
|
||
def on_button_pressed(self, event: Button.Pressed):
|
||
if event.button.id == "copy-message":
|
||
pyperclip.copy(self.message.content)
|
||
self.action_toggle_buttons()
|
||
|
||
elif event.button.id == "delete-message":
|
||
self.app.push_screen(ConfirmMessageDeleteScreen(self.message.id))
|
||
|
||
|
||
class UserResponse(Response):
|
||
def __init__(self, message: marvin.models.threads.Message, **kwargs) -> None:
|
||
classes = kwargs.setdefault("classes", "")
|
||
kwargs["classes"] = f"{classes} user-response".strip()
|
||
super().__init__(message=message, **kwargs)
|
||
|
||
|
||
class BotResponse(Response):
|
||
def __init__(self, message: marvin.models.threads.Message, **kwargs) -> None:
|
||
classes = kwargs.setdefault("classes", "")
|
||
kwargs["classes"] = f"{classes} bot-response".strip()
|
||
super().__init__(message=message, **kwargs)
|
||
|
||
|
||
class Hud(Container):
|
||
...
|
||
|
||
def compose(self) -> ComposeResult:
|
||
with VerticalScroll(id="hud"):
|
||
with Container(id="character"):
|
||
yield Label("character name", id="character-name")
|
||
yield Label("character health", id="character-health")
|
||
yield Label("imperial credits", id="imperial-credits")
|
||
with Container(id="ship"):
|
||
yield Label("ship name", id="ship-name")
|
||
yield Label("ship health", id="ship-health")
|
||
yield Label("ship fuel", id="ship-fuel")
|
||
|
||
|
||
class Conversation(Container):
|
||
# bot_name = reactive(None, layout=True)
|
||
|
||
# def watch_bot_name(self, bot_name: str):
|
||
# with self.app.batch_update():
|
||
# bot_name_label = self.query_one("#empty-thread-bot-name", Label)
|
||
# bot_description_label = self.query_one(
|
||
# "#empty-thread-bot-description", Label
|
||
# )
|
||
|
||
# if bot_name:
|
||
# bot_name_label.update(
|
||
# f"Send a message to [bold green]{self.bot_name}[/]!"
|
||
# )
|
||
# if self.app.bot.description:
|
||
# bot_description_label.update(f"{self.app.bot.description}")
|
||
# bot_description_label.remove_class("hidden")
|
||
# else:
|
||
# bot_description_label.add_class("hidden")
|
||
|
||
# else:
|
||
# bot_description_label.add_class("hidden")
|
||
# bot_name_label.update("Send a message to start a thread...")
|
||
|
||
def compose(self) -> ComposeResult:
|
||
input = Input(placeholder="What do you do ❯", id="message-input")
|
||
input.focus()
|
||
yield input
|
||
with VerticalScroll(id="messages"):
|
||
with Container(id="message-top"):
|
||
with Container(id="empty-thread-container"):
|
||
yield Label("", id="empty-thread-bot-name")
|
||
yield Label("", id="empty-thread-bot-description")
|
||
|
||
async def add_response(self, response: Response, scroll: bool = True) -> None:
|
||
messages = self.app.query_one("Conversation #messages", VerticalScroll)
|
||
# wait for the responses to be fully mounted before scrolling
|
||
# to avoid issues with rendering Markdown
|
||
await messages.mount(response)
|
||
if scroll:
|
||
messages.scroll_end(duration=0.2)
|
||
|
||
# show / hide the empty thread message
|
||
empty = self.app.query_one("Conversation #empty-thread-container")
|
||
empty.add_class("hidden")
|
||
|
||
def clear_responses(self) -> None:
|
||
responses = self.app.query("Response")
|
||
for response in responses:
|
||
response.remove()
|
||
# self.bot_name = getattr(self.app.bot, "name")
|
||
empty = self.query_one("Conversation #empty-thread-container")
|
||
empty.remove_class("hidden")
|
||
|
||
|
||
class CreateScreen(ModalScreen):
|
||
def __init__(self, message: str, **kwargs) -> None:
|
||
self.message = message
|
||
super().__init__(**kwargs)
|
||
|
||
def compose(self) -> ComposeResult:
|
||
with Container(id="create"):
|
||
yield Label(self.message, id="create-label")
|
||
|
||
|
||
class MainScreen(Screen):
|
||
BINDINGS = [
|
||
("escape", "focus_threads", "Focus on Threads"),
|
||
("n", "new_thread", "New Thread"),
|
||
("k", "scroll_up_messages", "Scroll Up"),
|
||
("j", "scroll_down_messages", "Scroll Down"),
|
||
("u", "page_up_messages", "Page Up"),
|
||
("d", "page_down_messages", "Page Down"),
|
||
]
|
||
action: Optional[str] = reactive(None, always_update=True, layout=True)
|
||
|
||
def action_focus_threads(self) -> None:
|
||
self.app.query_one("#threads", Threads).focus()
|
||
|
||
def action_focus_message(self) -> None:
|
||
self.app.query_one("#message-input", Input).focus()
|
||
|
||
def action_scroll_up_messages(self) -> None:
|
||
messages = self.query_one("Conversation #messages", VerticalScroll)
|
||
messages.scroll_up(duration=0.1)
|
||
|
||
def action_scroll_down_messages(self) -> None:
|
||
messages = self.query_one("Conversation #messages", VerticalScroll)
|
||
messages.scroll_down(duration=0.1)
|
||
|
||
def action_page_up_messages(self) -> None:
|
||
messages = self.query_one("Conversation #messages", VerticalScroll)
|
||
messages.scroll_page_up(duration=0.1)
|
||
|
||
def action_page_down_messages(self) -> None:
|
||
messages = self.query_one("Conversation #messages", VerticalScroll)
|
||
messages.scroll_page_down(duration=0.1)
|
||
|
||
def compose(self) -> ComposeResult:
|
||
# yield Sidebar(id="sidebar")
|
||
yield Hud(id="hud")
|
||
yield Conversation(id="conversation")
|
||
|
||
async def on_mount(self):
|
||
if not marvin.settings.openai_api_key.get_secret_value():
|
||
self.set_timer(0.5, self.action_show_settings_screen)
|
||
conversation = self.query_one("Conversation", Conversation)
|
||
|
||
await conversation.add_response(
|
||
BotResponse(
|
||
marvin.models.threads.Message(
|
||
role="bot",
|
||
name="sw-text-adventure",
|
||
bot_id=123,
|
||
content=f"{self.app.character.name} from {self.app.character.home_planet} - {self.app.character.backstory}",
|
||
)
|
||
)
|
||
)
|
||
await conversation.add_response(
|
||
BotResponse(
|
||
marvin.models.threads.Message(
|
||
role="bot",
|
||
name="sw-text-adventure",
|
||
bot_id=123,
|
||
content=f"Mission: {self.app.character.mission.description}",
|
||
)
|
||
)
|
||
)
|
||
|
||
async def on_input_submitted(self, event: Input.Submitted) -> None:
|
||
if event.input.disabled:
|
||
return
|
||
elif not event.input.value:
|
||
return
|
||
action = event.input.value
|
||
event.input.value = ""
|
||
conversation = self.query_one("Conversation", Conversation)
|
||
await conversation.add_response(
|
||
UserResponse(
|
||
marvin.models.threads.Message(name="User", role="user", content=action)
|
||
)
|
||
)
|
||
self.action = action
|
||
|
||
async def watch_action(self, action: str) -> None:
|
||
if action is None:
|
||
return
|
||
conversation = self.query_one("Conversation", Conversation)
|
||
|
||
result = did_complete_mission(self.app.character, action)
|
||
self.app.character.previous_missions.append(
|
||
(self.app.character.mission, result)
|
||
)
|
||
self.app.character.previous_missions = self.app.character.previous_missions[-5:]
|
||
self.app.character.imperial_credits -= result.imperial_credits_spent
|
||
self.app.character.imperial_credits += result.imperial_credits_earned
|
||
self.app.character.health -= result.health_lost
|
||
self.app.character.health += result.health_gained
|
||
self.app.character.ship.fuel_level -= result.fuel_used
|
||
message = ""
|
||
message += f"Your mission has been completed"
|
||
message += result.story
|
||
message += f"You earned {result.imperial_credits_earned} imperial credits"
|
||
message += f"You spent {result.imperial_credits_spent} imperial credits"
|
||
message += f"You gained {result.health_gained} health"
|
||
message += f"You lost {result.health_lost} health"
|
||
|
||
await conversation.add_response(
|
||
BotResponse(
|
||
marvin.models.threads.Message(
|
||
role="bot",
|
||
name="sw-text-adventure",
|
||
bot_id=123,
|
||
content=result.story,
|
||
)
|
||
)
|
||
)
|
||
|
||
await conversation.add_response(
|
||
BotResponse(
|
||
marvin.models.threads.Message(
|
||
role="bot",
|
||
name="sw-text-adventure",
|
||
bot_id=123,
|
||
content=f"You earned {result.imperial_credits_earned} imperial credits, spent {result.imperial_credits_spent} imperial credits, gained {result.health_gained} hp, and lost {result.health_lost} hp.",
|
||
)
|
||
)
|
||
)
|
||
|
||
self.app.character.mission = get_next_mission(
|
||
self.app.character, action, result.success
|
||
)
|
||
|
||
await conversation.add_response(
|
||
BotResponse(
|
||
marvin.models.threads.Message(
|
||
role="bot",
|
||
name="sw-text-adventure",
|
||
bot_id=123,
|
||
content=f"Next Mission: {self.app.character.mission.description} imperial credits, spent {result.imperial_credits_spent} imperial credits, gained {result.health_gained} hp, and lost {result.health_lost}",
|
||
)
|
||
)
|
||
)
|
||
|
||
async def on_button_pressed(self, event: Button.Pressed) -> None:
|
||
if event.button.id == "show-settings":
|
||
self.action_show_settings_screen()
|
||
elif event.button.id == "show-bots":
|
||
self.action_show_bots_screen()
|
||
elif event.button.id == "create-new-thread":
|
||
self.action_new_thread()
|
||
elif event.button.id == "delete-thread":
|
||
await self.action_delete_thread()
|
||
elif event.button.id == "quit":
|
||
self.app.exit()
|
||
|
||
|
||
class SWTextAdventure(App):
|
||
CSS_PATH = ["app.css"]
|
||
character: Optional[StarWarsCharacter] = reactive(
|
||
None, always_update=True, layout=True
|
||
)
|
||
game_type: Optional[str] = reactive(None, always_update=True, layout=True)
|
||
game: Optional[Game] = reactive(None, always_update=True, layout=True)
|
||
game_ready: bool = reactive(False)
|
||
is_ready: bool = False
|
||
|
||
def __init__(self, **kwargs):
|
||
super().__init__(**kwargs)
|
||
self.game_type = "star-wars"
|
||
|
||
async def on_ready(self) -> None:
|
||
self.push_screen(MainScreen())
|
||
|
||
async def watch_game_type(self) -> None:
|
||
self.push_screen(CreateScreen(message="starting"))
|
||
time.sleep(1)
|
||
if self.character is None:
|
||
self.push_screen(CreateScreen(message="creating character"))
|
||
self.character = StarWarsCharacter.parse_file("character.json")
|
||
# self.character=create_character()
|
||
if self.game is None:
|
||
self.push_screen(
|
||
CreateScreen(message=f"creating game for {self.character.name}")
|
||
)
|
||
self.game = create_game # (self.character)
|
||
# self.pop_screen()
|
||
self.game_ready = True
|
||
self.is_ready = True
|
||
|
||
async def watch_action(self) -> None:
|
||
...
|
||
|
||
async def watch_game_ready(self, game_ready: bool) -> None:
|
||
console.log(f"Game ready: {game_ready}")
|
||
|
||
|
||
def run_app():
|
||
import os
|
||
import sys
|
||
|
||
from textual.features import parse_features
|
||
|
||
dev = "--dev" in sys.argv
|
||
features = set(parse_features(os.environ.get("TEXTUAL", "")))
|
||
if dev:
|
||
features.add("debug")
|
||
features.add("devtools")
|
||
|
||
os.environ["TEXTUAL"] = ",".join(sorted(features))
|
||
app = SWTextAdventure()
|
||
app.run()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
run_app()
|