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"""

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

    redis: Redis
    """Redis connection"""

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

    _log: deque[ChatMessage]
    _exit_event: Event

    def __init__(self, context: SSHContext, **kwargs):
        super().__init__(context, **kwargs)
        self.redis = Resources().cache
        self.pubsub = self.redis.pubsub()
        self.pubsub.subscribe(**{"chat": self.on_chat})
        self._log = deque(maxlen=LIMIT)
        self._exit_event = Event()
        self.run_worker(self._listen, exclusive=True, thread=True)

    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):
        # chat log
        yield VerticalScroll(Static(id="log"))

        # input
        input_widget = Input(
            placeholder="Enter a message or press ESC",
            max_length=MAX_LENGTH,
        )
        input_widget.focus()
        yield input_widget

    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._log.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._log])
            )
        )
        vs = self.query_one(VerticalScroll)
        vs.scroll_end(animate=False)
        input = self.query_one(Input)
        input.value = ""

    def exit(self) -> 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().exit()

    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

Create an instance of an app.

Args

driver_class

Driver class or None to auto-detect. This will be used by some Textual tools.

css_path

Path to CSS or None to use the CSS_PATH class variable. To load multiple CSS files, pass a list of strings or paths which will be loaded in order.

watch_css

Reload CSS if the files changed. This is set automatically if you are using textual run with the dev switch.

ansi_color

Allow ANSI colors if True, or convert ANSI colors to to RGB if False.

Raises

CssPathError

When the supplied CSS path(s) are an unexpected type.

Ancestors

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

Class variables

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):
    # chat log
    yield VerticalScroll(Static(id="log"))

    # input
    input_widget = Input(
        placeholder="Enter a message or press ESC",
        max_length=MAX_LENGTH,
    )
    input_widget.focus()
    yield input_widget

Yield child widgets for a container.

This method should be implemented in a subclass.

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._log.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._log])
        )
    )
    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(),
    )

Inherited members

• XthuluApp:
  • context
  • exit

class ChatMessage (**data: Any)

Expand source code

class ChatMessage(BaseModel):
    user: str | None
    message: str

Usage Documentation

Models

A base class for creating Pydantic models.

Attributes

__class_vars__

The names of the class variables defined on the model.

__private_attributes__

Metadata about the private attributes of the model.

__signature__

The synthesized __init__ [Signature][inspect.Signature] of the model.

__pydantic_complete__

Whether model building is completed, or if there are still undefined fields.

__pydantic_core_schema__

The core schema of the model.

__pydantic_custom_init__

Whether the model has a custom __init__ function.

__pydantic_decorators__

Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.

__pydantic_generic_metadata__

Metadata for generic models; contains data used for a similar purpose to args, origin, parameters in typing-module generics. May eventually be replaced by these.

__pydantic_parent_namespace__

Parent namespace of the model, used for automatic rebuilding of models.

__pydantic_post_init__

The name of the post-init method for the model, if defined.

__pydantic_root_model__

Whether the model is a [RootModel][pydantic.root_model.RootModel].

__pydantic_serializer__

The pydantic-core SchemaSerializer used to dump instances of the model.

__pydantic_validator__

The pydantic-core SchemaValidator used to validate instances of the model.

__pydantic_fields__

A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.

__pydantic_computed_fields__

A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.

__pydantic_extra__

A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra] is set to 'allow'.

__pydantic_fields_set__

The names of fields explicitly set during instantiation.

__pydantic_private__

Values of private attributes set on the model instance.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var message : str

var model_config

var user : str | None