Module userland.scripts.chat

Node chat script

Global variables

var LIMIT

Total number of messages to keep in backlog

var MAX_LENGTH

Maximum length of individual messages

Functions

async def main(cx: SSHContext) ‑> None

Expand source code

async def main(cx: SSHContext) -> None:
    cx.console.set_window_title("chat")
    await ChatApp(cx).run_async()

Classes

class ChatApp (context: SSHContext,
**kwargs)

Expand source code

class ChatApp(XthuluApp):
    """Node chat Textual app"""

    AUTO_FOCUS = "Input"
    BINDINGS = [Binding("escape", "quit", show=False)]

    redis: Redis
    """Redis connection"""

    pubsub: PubSub
    """Redis PubSub connection"""

    _chatlog: deque[ChatMessage]
    _exit_event: Event

    def __init__(self, context: SSHContext, **kwargs):
        ""  # empty docstring
        super(ChatApp, self).__init__(context, **kwargs)
        self.redis = Resources().cache
        self.pubsub = self.redis.pubsub()
        self.pubsub.subscribe(**{"chat": self.on_chat})
        self._chatlog = deque(maxlen=LIMIT)
        self._exit_event = Event()

    def _listen(self) -> None:
        self.redis.publish(
            "chat",
            ChatMessage(
                user=None, message=f"{self.context.username} has joined"
            ).model_dump_json(),
        )

        while not self._exit_event.is_set():
            self.pubsub.get_message(True, 0.01)

    def compose(self):
        yield VerticalScroll(Static(id="log"))
        yield Input(
            placeholder="Enter a message or press ESC",
            max_length=MAX_LENGTH,
        )

    def on_chat(self, message: dict[str, str]) -> None:
        def format_message(msg: ChatMessage):
            if msg.user:
                return (
                    f"\n[bright_white on blue]<{msg.user}>[/] "
                    f"{escape(msg.message)}"
                )

            return (
                "\n[bright_white on red]<*>[/] "
                f"[italic][white]{msg.message}[/][/]"
            )

        msg = ChatMessage(**json.loads(message["data"]))
        self._chatlog.append(msg)
        l: Static = self.get_widget_by_id("log")  # type: ignore
        l.update(
            self.console.render_str(
                "".join([format_message(m) for m in self._chatlog])
            )
        )
        vs = self.query_one(VerticalScroll)
        vs.scroll_end(animate=False)
        input = self.query_one(Input)
        input.value = ""

    def exit(self, **kwargs) -> None:
        msg = ChatMessage(
            user=None, message=f"{self.context.username} has left"
        )
        self.redis.publish("chat", msg.model_dump_json())
        self._exit_event.set()
        self.workers.cancel_all()
        super(ChatApp, self).exit()

    async def on_ready(self) -> None:
        self.run_worker(self._listen, exclusive=True, thread=True)

    def on_input_submitted(self, event: Input.Submitted) -> None:
        val = event.input.value.strip()

        if val == "":
            return

        self.redis.publish(
            "chat",
            ChatMessage(
                user=self.context.username, message=val
            ).model_dump_json(),
        )

Node chat Textual app

Ancestors

• XthuluApp
• textual.app.App
• typing.Generic
• textual.dom.DOMNode
• textual.message_pump.MessagePump

Class variables

var AUTO_FOCUS

var BINDINGS

var pubsub : redis.client.PubSub

Redis PubSub connection

var redis : redis.client.Redis

Redis connection

Methods

def compose(self)

Expand source code

def compose(self):
    yield VerticalScroll(Static(id="log"))
    yield Input(
        placeholder="Enter a message or press ESC",
        max_length=MAX_LENGTH,
    )

Yield child widgets for a container.

This method should be implemented in a subclass.

def exit(self, **kwargs) ‑> None

Expand source code

def exit(self, **kwargs) -> None:
    msg = ChatMessage(
        user=None, message=f"{self.context.username} has left"
    )
    self.redis.publish("chat", msg.model_dump_json())
    self._exit_event.set()
    self.workers.cancel_all()
    super(ChatApp, self).exit()

def on_chat(self, message: dict[str, str]) ‑> None

Expand source code

def on_chat(self, message: dict[str, str]) -> None:
    def format_message(msg: ChatMessage):
        if msg.user:
            return (
                f"\n[bright_white on blue]<{msg.user}>[/] "
                f"{escape(msg.message)}"
            )

        return (
            "\n[bright_white on red]<*>[/] "
            f"[italic][white]{msg.message}[/][/]"
        )

    msg = ChatMessage(**json.loads(message["data"]))
    self._chatlog.append(msg)
    l: Static = self.get_widget_by_id("log")  # type: ignore
    l.update(
        self.console.render_str(
            "".join([format_message(m) for m in self._chatlog])
        )
    )
    vs = self.query_one(VerticalScroll)
    vs.scroll_end(animate=False)
    input = self.query_one(Input)
    input.value = ""

def on_input_submitted(self, event: textual.widgets._input.Input.Submitted) ‑> None

Expand source code

def on_input_submitted(self, event: Input.Submitted) -> None:
    val = event.input.value.strip()

    if val == "":
        return

    self.redis.publish(
        "chat",
        ChatMessage(
            user=self.context.username, message=val
        ).model_dump_json(),
    )

async def on_ready(self) ‑> None

Expand source code

async def on_ready(self) -> None:
    self.run_worker(self._listen, exclusive=True, thread=True)

Inherited members

• XthuluApp:
  • ENABLE_COMMAND_PALETTE
  • context

class ChatMessage (**data: Any)

Expand source code

class ChatMessage(BaseModel):
    """Posted chat message"""

    message: str
    """Message text"""

    user: str | None
    """Message author (or `None` if system)"""

    def __init__(self, **data: Any):
        ""  # empty docstring
        super(ChatMessage, self).__init__(**data)

Posted chat message

Ancestors

• pydantic.main.BaseModel

Class variables

var message : str

Message text

var model_config

var user : str | None

Message author (or None if system)