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

Classes

class ChatApp (context: SSHContext, **kwargs)

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.

Raises

CssPathError
When the supplied CSS path(s) are an unexpected type.
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(),
        )

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)

Yield child widgets for a container.

This method should be implemented in a subclass.

def on_chat(self, message: dict[str, str]) ‑> None
def on_input_submitted(self, event: textual.widgets._input.Input.Submitted) ‑> None

Inherited members

class ChatMessage (**data: Any)

Usage docs: https://docs.pydantic.dev/2.7/concepts/models/

A base class for creating Pydantic models.

Attributes

__class_vars__
The names of classvars defined on the model.
__private_attributes__
Metadata about the private attributes of the model.
__signature__
The signature for instantiating the model.
__pydantic_complete__
Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__
The pydantic-core schema used to build the SchemaValidator and SchemaSerializer.
__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_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_extra__
An instance attribute with the values of extra fields from validation when model_config['extra'] == 'allow'.
__pydantic_fields_set__
An instance attribute with the names of fields explicitly set.
__pydantic_private__
Instance attribute with the 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.

Expand source code 
class ChatMessage(BaseModel):
    user: str | None
    message: str

Ancestors

  • pydantic.main.BaseModel

Class variables

var message : str
var model_computed_fields
var model_config
var model_fields
var user : str | None