rich
233dfc813c
CI / lint (push) Waiting to run
CI / test macos-latest py3.14 (push) Waiting to run
CI / test ubuntu-24.04 py3.10 (push) Waiting to run
CI / test ubuntu-latest py3.11 (push) Waiting to run
CI / test ubuntu-latest py3.12 (push) Waiting to run
CI / test ubuntu-latest py3.13 (push) Waiting to run
CI / test ubuntu-latest py3.14 (push) Waiting to run
CI / test windows-latest py3.10 (push) Waiting to run
CI / test windows-latest py3.14 (push) Waiting to run
CI / build (push) Waiting to run
Issue trailers / check (push) Waiting to run
Bring in the 65 upstream commits since 0.11.1, with ^K lock and the self chip. Both sides kept their features. React gives way to Lock in two places. On the PicoCalc lane, Lock keeps the Shift half of F2 and React is ^E only. In the hint, ^E react is the first atom to go when ^K lock has no room. The long-message code now refills the compose line with MentionEditor, which replaced LineEditor upstream.
2431 lines
114 KiB
Python
2431 lines
114 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
||
"""The live chat screen and the function that opens it.
|
||
|
||
This is the interactive, full-screen chat: a transcript that scrolls, with an input line
|
||
pinned at the bottom. The messages that you send and the messages that come in over the
|
||
mesh show together in real time. This module is in the UI layer, the same as the device
|
||
picker and the config editor, but it can depend on the context and the services. It
|
||
connects the send path of :class:`~meshterm.services.chat_service.ChatService` and the
|
||
event hub, which always runs, to a :class:`~meshterm.ui.tui.screen.Screen`.
|
||
|
||
While the screen is open, it subscribes to the hub. Thus the inbound messages for the
|
||
current conversation are appended live. Separately,
|
||
:class:`~meshterm.services.chat_service.ChatService` stores each inbound message. Thus
|
||
the history is complete, also when the screen is not open.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import re
|
||
import unicodedata
|
||
from collections.abc import Awaitable, Callable
|
||
from functools import lru_cache
|
||
from typing import TYPE_CHECKING, Any
|
||
|
||
from rich.cells import cell_len
|
||
from rich.console import Group, RenderableType
|
||
from rich.text import Text
|
||
|
||
from ..core import links, parts, reactions
|
||
from ..core.channels import MENTION, split_channel_sender
|
||
from ..core.connection import ContactNotOnDeviceError
|
||
from ..core.events import EventKind, MeshEvent
|
||
from ..core.models import ChatMessage, Contact, Conversation, Message, utcnow
|
||
from ..core.regions import WILDCARD
|
||
from ..platforms import Platform, on_platform
|
||
from .compose import MentionEditor
|
||
from .pathline import PATH_INK, PathHop
|
||
from .qr import QrScreen, qr_strip
|
||
from .theme import emoticon, name_style, snr_style
|
||
from .tui.prompt import (
|
||
CHANNEL_BYTE_LIMIT,
|
||
DM_BYTE_LIMIT,
|
||
byte_counter,
|
||
)
|
||
from .tui.render import render_hanging, render_lines, right_aligned_tail
|
||
from .tui.screen import CANCEL, Screen
|
||
from .tui.spinner import Spinner, spinner_interval
|
||
from .widgets import (
|
||
DESTINATION,
|
||
ORIGIN,
|
||
author_chip,
|
||
hop_chip,
|
||
mention_hop,
|
||
mention_mark,
|
||
picked_hop,
|
||
self_chip,
|
||
)
|
||
|
||
if TYPE_CHECKING:
|
||
from ..context import AppContext
|
||
from .map_render import MapMarker
|
||
from .map_screen import Locator
|
||
from .node_detail_screen import ContactPages
|
||
|
||
|
||
#: The delivery marks for a *resolved* outbound direct message, at the end of its line:
|
||
#: acknowledged, or transmitted but not acknowledged (^R can retry it). These are the
|
||
#: ``✓``/``✗`` status marks of the app in their ok/err styles, not the ✅/❌ emoji, which
|
||
#: are for the lane of packet-class icons. While the ack is still pending, the line shows an
|
||
#: animated spinner instead (refer to :meth:`ChatScreen._delivery_glyph`).
|
||
_DELIVERED = ("✓", "ok")
|
||
_FAILED = ("✗", "err")
|
||
|
||
#: A word between whitespace. The wrap moves a word only as a whole. Thus the run that a
|
||
#: wrap must not cut a URL in is the word that has the URL, with a bracket or a full stop
|
||
#: next to it.
|
||
_WORD = re.compile(r"\S+")
|
||
|
||
#: The column where a URL starts when it is too long to hang under its body: the column
|
||
#: after the ``❯`` that is in front of the first line of the selected message. A message
|
||
#: that names us has its mark in this column, and the timestamp follows it.
|
||
_URL_GUTTER = 2
|
||
|
||
#: Whether the URLs of a message are drawn as QR codes under it (``Platform.url_codes``,
|
||
#: only on the PicoCalc). Bound for each platform, never asked at each frame.
|
||
_URL_CODES = False
|
||
|
||
#: The mark in front of the count of a reaction. Where the font has no emoji, an emoticon
|
||
#: such as ``+1`` or ``<3`` stands in for the emoji, and a bare count after it reads as part
|
||
#: of it (``+1 3``). There, the count is ``x3``. Bound for each platform.
|
||
_COUNT_MARK = ""
|
||
|
||
|
||
@on_platform
|
||
def _bind(platform: Platform) -> None:
|
||
"""Bind the transcript to the platform (now and at each switch).
|
||
|
||
The platform says whether codes hang under URLs, and how a reaction count is marked.
|
||
"""
|
||
global _URL_CODES, _COUNT_MARK
|
||
_URL_CODES = platform.url_codes
|
||
_COUNT_MARK = "" if platform.emoji else "x"
|
||
|
||
|
||
#: The maximum width of a footer hint, in cells (refer to the UX standards).
|
||
_HINT_CELLS = 72
|
||
|
||
|
||
def _fit_hint(atoms: list[str], spare: tuple[str, ...] = ()) -> str:
|
||
"""Join the hint atoms that are not empty, and remove ``spare`` atoms until the hint fits.
|
||
|
||
The ``spare`` atoms go in the given order, and only while the hint is wider than
|
||
:data:`_HINT_CELLS`. Thus a key that the line can hold stays on it.
|
||
"""
|
||
kept = [atom for atom in atoms if atom]
|
||
for atom in spare:
|
||
if cell_len(" · ".join(kept)) <= _HINT_CELLS:
|
||
break
|
||
if atom in kept:
|
||
kept.remove(atom)
|
||
return " · ".join(kept)
|
||
|
||
|
||
#: The long words of a hint atom, and the short words that the same atom has when the hint
|
||
#: needs the room. The two forms count as one atom. The short form leaves out a note or a
|
||
#: second key, as the hint already does for ^U, ^L, and ^O.
|
||
_SHORT_ATOMS = {"Enter reply (@mention)": "Enter reply", "^End/Esc cancel": "Esc cancel"}
|
||
|
||
|
||
def _atom_set(hint: str) -> set[str]:
|
||
"""The atoms of a hint, each in its short form, to compare two hints by their atoms."""
|
||
return {_SHORT_ATOMS.get(atom, atom) for atom in hint.split(" · ")}
|
||
|
||
|
||
class ChatScreen(Screen):
|
||
"""A live conversation: a transcript that scrolls, above a pinned input line.
|
||
|
||
The transcript automatically stays on the newest message (and goes back to the bottom
|
||
each time you type or send). ↑ selects a message. The selection moves with
|
||
↑↓/PgUp/PgDn, and the viewport moves with it. ^End (or Esc) puts the focus on the
|
||
compose line again. Enter sends the current line, or does an action on a selected
|
||
message: in a channel, it starts a reply with an ``@mention``, and in a direct chat, it
|
||
opens the delivery paths of the message. In the two types of chat, ^P opens the paths
|
||
of the selected message. A path belongs to one message, so when no message is
|
||
selected, there is nothing to show. ^U shows the links of the selected message as QR
|
||
codes, on the share screen. ^L opens the map on the node of the author of the selected
|
||
message, when that node has a location. ^K locks that author against the archive
|
||
sweep, or unlocks it. ^E reacts to the selected message with an emoji (refer to
|
||
:mod:`~meshterm.core.reactions`). When no message is selected, Esc leaves the chat.
|
||
|
||
A reaction is a message on the wire, but the transcript does not show it as one. It shows
|
||
each reaction under the message that it reacts to: the emoji and a count, or the names of
|
||
the nodes that reacted while that message is selected. A room has no reactions, so a
|
||
room shows each message as it arrived.
|
||
|
||
A channel with a send scope shows its region in the title (``#ops · yul``). There, ^R
|
||
sends a message again *unscoped*: the selected message, or the newest when no message is
|
||
selected. This is the solution for a scoped message that no repeater in range relays.
|
||
It comes after an amber confirm, because an unscoped flood gets to each repeater that
|
||
the scope kept it from.
|
||
"""
|
||
|
||
floating = False
|
||
#: Home and End move the caret of the compose line here. Thus edge scroll does not use
|
||
#: them.
|
||
home_end_jumps = False
|
||
|
||
@property
|
||
def picocalc_lyra_lane(self):
|
||
"""The PicoCalc lane of a conversation: the same lane in a channel, a DM, and a room.
|
||
|
||
The plain bank of F1 to F3 has the verbs that all three screens share, in the order
|
||
of the hint: ``Paths`` (^P), ``URL→QR`` (^U), and ``Map`` (^L). The Shift bank of F1
|
||
and F2 has the verbs of one screen only. ``Retry`` (^R) is on the Shift half of F1 in
|
||
a DM and in a room, and ``Resend`` (the same ^R, unscoped) is there in a channel. A
|
||
room adds ``Log in`` (^G) on the Shift half of F2 (refer to
|
||
:class:`~meshterm.ui.room.RoomScreen`). Thus a key does the same thing on each of
|
||
the three screens, and the user finds a verb in the same place (JP, 2026-10-10).
|
||
|
||
The shared verbs act *on* the selected message. ``Paths`` is lit while a message is
|
||
selected. ``URL→QR`` shows the codes of the links of the selection on a share
|
||
screen, one code at a time with its full URL, and is lit while the selection has a
|
||
URL (refer to :meth:`_picked_urls`). On the Cardputer it is the only way to a code.
|
||
On the PicoCalc, the transcript already draws each code under its message in the
|
||
same size, so there the chip only shows one code alone. ``Map`` is lit while the
|
||
author of the selection has a location (refer to :meth:`_locate_target`).
|
||
|
||
``Author`` (^O, the node page of the author of the selection) is the fourth shared
|
||
verb, and the plain bank has only three slots. Thus it is on the Shift half of F3,
|
||
behind ``Map``, the other verb about the author. It is lit while the author has a
|
||
page (refer to :meth:`_author_page`). Each of the two halves of F3 is empty when
|
||
the screen has nothing to find that node with.
|
||
|
||
``Lock`` (^K) is the fifth verb about the author, and it is on the Shift half of F2.
|
||
It toggles the lock of the author, so it reads ``Unlock`` on a locked author, and
|
||
it is lit while the author can have a lock (refer to :meth:`_lock_target`). That
|
||
slot is free in a channel and in a DM. In a room, ``Log in`` has it, so there the
|
||
user locks from the ``Author`` page, which a conversation opens with its lock row
|
||
(JP, 2026-10-19). The slot is empty where MeshTerm cannot store a lock.
|
||
|
||
``React`` (^E) has no slot. It was on the Shift half of F2 before ``Lock`` came, and
|
||
the Shift bank has no free slot now. ``Lock`` keeps the slot (Rich, 2026-10-22), so a
|
||
reaction is ^E only, and the hint names it when there is room.
|
||
|
||
The Shift half of F1 follows the two rules of the lane. In a channel with no
|
||
resend connected, it is empty, because a channel message is never acknowledged and
|
||
there is nothing to retry. In a DM with nothing outstanding, it is only dim.
|
||
``Resend`` sends the selected message (or the newest) again, unscoped (refer to
|
||
:meth:`_retry_target`), and it is dim until that message is one of ours that went
|
||
out under a region.
|
||
|
||
The day jump (``ctrl_pageup``/``ctrl_pagedown``) has no chip. ``Day ↑`` and
|
||
``Day ↓`` were on F1 and F2, and JP removed them (2026-10-10). The desktop keeps
|
||
the jump on ^PgUp and ^PgDn.
|
||
|
||
F4/F5 keep the shared paging pair. The Shift bank of the shared lane sends the plain
|
||
``end``/``home`` actions. But here the cursor of the compose line already uses them
|
||
(refer to :meth:`handle`). Thus, without a change, the two keyboard keys go to the
|
||
same fallback ("clear the selection, stay at the tail"), and each one seems to
|
||
scroll to the bottom. Instead, the two Shift slots send ``ctrl_end``/``ctrl_home``,
|
||
and their labels say what these actions do to a conversation: go back to the
|
||
*latest* message and the compose line, or go back to the *oldest*. Each jump is
|
||
still behind the page key that goes in the same direction (refer to
|
||
:data:`~meshterm.ui.tui.fkeys.DEFAULT_LANE`). From the outside, the transcript moves
|
||
by one screen, also when it is the selection that moves.
|
||
"""
|
||
from .tui.fkeys import FPair, default_lane
|
||
|
||
lane = list(default_lane(nav=bool(self._messages)))
|
||
live = bool(self._messages)
|
||
if not self._is_channel:
|
||
again = ("Retry", "retry")
|
||
elif self._resend_unscoped is not None:
|
||
again = ("Resend", "retry")
|
||
else:
|
||
again = ("", "")
|
||
lane[0] = FPair(
|
||
"Paths",
|
||
"paths",
|
||
*again,
|
||
enabled=self._selected is not None and self._paths is not None,
|
||
opp_enabled=self._retry_target() is not None,
|
||
)
|
||
lock = self._lock_target()
|
||
if self._pages is not None and self._pages.can_lock:
|
||
toggle = ("Unlock" if lock is not None and lock[1] else "Lock", "lock")
|
||
else:
|
||
toggle = ("", "")
|
||
lane[1] = FPair(
|
||
"URL→QR",
|
||
"url_code",
|
||
*toggle,
|
||
enabled=bool(self._picked_urls()),
|
||
opp_enabled=lock is not None,
|
||
)
|
||
if self._locator is not None or self._pages is not None:
|
||
where = ("Map", "locate") if self._locator is not None else ("", "")
|
||
who = ("Author", "author_page") if self._pages is not None else ("", "")
|
||
lane[2] = FPair(
|
||
*where,
|
||
*who,
|
||
enabled=self._locate_target() is not None,
|
||
opp_enabled=self._author_page() is not None,
|
||
)
|
||
lane[3] = FPair("Page ↓", "pagedown", "Latest", "ctrl_end", enabled=live, opp_enabled=live)
|
||
lane[4] = FPair("Page ↑", "pageup", "Oldest", "ctrl_home", enabled=live, opp_enabled=live)
|
||
return lane
|
||
|
||
def __init__(
|
||
self,
|
||
conversation: Conversation,
|
||
messages: list[ChatMessage],
|
||
*,
|
||
send: Callable[[str], Awaitable[ChatMessage | None]],
|
||
names: dict[str, str],
|
||
session, # noqa: ANN001 - a TuiSession, imported late to prevent an import cycle
|
||
resend: Callable[[ChatMessage], Awaitable[ChatMessage]] | None = None,
|
||
paths: Callable[[ChatMessage], Awaitable[None]] | None = None,
|
||
key_of: Callable[[str], str | None] | None = None,
|
||
scope: str | None = None,
|
||
resend_unscoped: Callable[[ChatMessage], Awaitable[ChatMessage | None]] | None = None,
|
||
locator: Locator | None = None,
|
||
pages: ContactPages | None = None,
|
||
self_name: str | None = None,
|
||
reaction_set: list[str] | None = None,
|
||
max_parts: int = 1,
|
||
part_gap_s: float = 0.0,
|
||
self_key: str | None = None,
|
||
) -> None:
|
||
"""Build the chat screen.
|
||
|
||
Args:
|
||
conversation: The thread that the screen shows (its label is the title of the
|
||
screen).
|
||
messages: The initial transcript (history), the oldest first.
|
||
send: An async callable that sends a line and returns the stored outbound
|
||
message (or ``None`` if nothing was sent).
|
||
names: A map from contact key prefix to friendly name, for the labels of
|
||
inbound direct messages.
|
||
session: The running :class:`~meshterm.ui.tui.session.TuiSession`. The screen
|
||
uses it to ask for a paint when messages arrive or a send completes.
|
||
resend: An async callable that tries again to deliver a direct message that
|
||
is not acknowledged, and changes the message in place (direct chats only,
|
||
``None`` for channels).
|
||
paths: An async callable that shows the delivery paths of the selected
|
||
message (the ^P view). ``None`` makes this action do nothing, with no
|
||
message.
|
||
key_of: Finds the key of the node of a sender from its display name (refer to
|
||
:func:`~meshterm.services.trace_runner.make_name_key_resolver`). The key
|
||
is the source of the hue of the sender. With ``None`` (or for a name that
|
||
it cannot find), the sender gets the grey of an unknown node
|
||
(``node.unknown``), because colour is only for identities that have a key.
|
||
scope: The send scope of the channel, shown in the title as a bare ``·`` atom
|
||
(``#ops · yul``: the title of a channel has no other atom that the user
|
||
can think is a region name). ``None`` for a channel that sends under the
|
||
default scope of the device, and always ``None`` for a direct chat.
|
||
resend_unscoped: An async callable that sends the text of a channel message
|
||
again, unscoped, and returns the stored message (channels only. With
|
||
``None``, ^R does nothing there).
|
||
locator: The located nodes, for ^L: the map on the author of the selected
|
||
message (refer to :class:`~meshterm.ui.map_screen.Locator`). With ``None``,
|
||
^L does nothing.
|
||
pages: The contacts, for ^O: the node page of the author of the selected
|
||
message (refer to :class:`~meshterm.ui.node_detail_screen.ContactPages`).
|
||
With ``None``, ^O does nothing.
|
||
self_name: The name of our node (refer to :func:`our_node`). Our messages and
|
||
the mentions of our name show it (:func:`~meshterm.ui.widgets.self_chip`).
|
||
A channel reaction names the author of its target, and our node is the
|
||
author of each message that we sent. With ``None``, our channel messages
|
||
get no reactions.
|
||
reaction_set: The emoji that ^E offers, in order. ``None`` offers
|
||
:data:`~meshterm.core.reactions.DEFAULT_REACTIONS`.
|
||
max_parts: The largest number of parts for a message that is too long for one
|
||
packet (refer to :mod:`meshterm.core.parts`). With 1, a message that is
|
||
too long is refused.
|
||
part_gap_s: The time to wait between two parts of a channel message, so that
|
||
the parts do not go out as a burst. A direct part waits for its
|
||
acknowledgement, and does not wait this time.
|
||
self_key: Our public key, which gives the hue of that chip.
|
||
"""
|
||
super().__init__()
|
||
self._self_name = self_name
|
||
self._self_key = self_key
|
||
self._scope = scope if conversation.is_channel else None
|
||
self.title = f"{conversation.label} · {self._scope}" if self._scope else conversation.label
|
||
self._is_channel = conversation.is_channel
|
||
# Many voices or one. The user reads a channel and a room in the same way: each
|
||
# message under its author, and Enter on a message starts a reply with an
|
||
# @mention. A direct chat has two persons, and Enter on a message opens its paths.
|
||
self._multiparty = conversation.is_channel or conversation.is_room
|
||
# Whether typing goes to the compose line. It is off where no typed text can be
|
||
# sent (a room that keeps nothing from us). Thus the keyboard keys never fill a
|
||
# line that the user cannot see.
|
||
self._composing = True
|
||
self._key_of: Callable[[str], str | None] = key_of or (lambda name: None)
|
||
# The one remote sender of a direct thread is the peer. Its key gives the colour of
|
||
# the header, also when the resolver cannot find the display name.
|
||
contact = conversation.contact
|
||
self._peer_key = (
|
||
""
|
||
if conversation.is_channel or contact is None
|
||
else (contact.public_key or contact.key_prefix or "")
|
||
)
|
||
# The reactions of the conversation, folded onto their targets. A room has none.
|
||
self._fold = (
|
||
None
|
||
if conversation.is_room
|
||
else reactions.ReactionFold(channel=conversation.is_channel, self_name=self_name)
|
||
)
|
||
self._reaction_set = reaction_set or reactions.DEFAULT_REACTIONS.split()
|
||
self._peer_label = conversation.label
|
||
self._messages = [m for m in messages if not self._folds(m)]
|
||
self._send = send
|
||
self._resend = resend
|
||
self._resend_unscoped = resend_unscoped if conversation.is_channel else None
|
||
self._max_parts = max(1, max_parts)
|
||
self._part_gap_s = max(0.0, part_gap_s)
|
||
self._paths = paths
|
||
self._locator = locator
|
||
self._pages = pages
|
||
self._names = names
|
||
self._session = session
|
||
self._editor = MentionEditor()
|
||
self._sending = False
|
||
# It turns while a direct message is in flight, so that the glyph at the end of the
|
||
# message spins (instead of a static hourglass) until the ack resolves. All messages
|
||
# share it: only one send or retry is ever in flight at a time (``_sending``
|
||
# controls the two).
|
||
self._spinner = Spinner()
|
||
self._status = ""
|
||
self._stick = True # keep the newest message visible until the user scrolls up
|
||
self._paths_open = False # one paths dialog at a time
|
||
self._map_open = False # one map at a time
|
||
self._page_open = False # one node page at a time
|
||
self._codes_open = False # one share screen of links at a time
|
||
self._paste_open = False # one paste-confirm dialog at a time
|
||
self._resend_open = False # one resend-unscoped confirm at a time
|
||
self._react_open = False # one reaction dialog at a time
|
||
# The selection: the index of the highlighted message (or None when the compose
|
||
# line has the focus), and the body line where it rendered, so that the frame keeps
|
||
# it visible.
|
||
self._selected: int | None = None
|
||
self._selected_line: int | None = None
|
||
# One line after the last rendered row of the selected message, so that its full
|
||
# block stays visible, not only its head (refer to render_body).
|
||
self._selected_end: int | None = None
|
||
# The cache of the output of _render_grouped for each message. Thus a paint that
|
||
# something outside the transcript causes (a key press, the 1 s idle tick, an
|
||
# animation step of a spinner) renders again only the rows that changed. Refer to
|
||
# _render_grouped.
|
||
self._render_cache: dict | None = None
|
||
# Strong references to the sends, resends, and spinner tickers that a key handler
|
||
# starts. The event loop holds only a *weak* reference. Thus the garbage collector
|
||
# can remove a task that nothing else names while it runs: the result is a message
|
||
# that is only half transmitted, or a spinner that stops. Each task removes itself
|
||
# from the set when it finishes.
|
||
self._tasks: set[asyncio.Task] = set()
|
||
|
||
def _spawn(self, coro) -> asyncio.Task: # noqa: ANN001 - any coroutine of this screen
|
||
"""Start ``coro`` on the loop and keep a reference to it until it finishes."""
|
||
task = asyncio.ensure_future(coro)
|
||
self._tasks.add(task)
|
||
task.add_done_callback(self._tasks.discard)
|
||
return task
|
||
|
||
@property
|
||
def footer_hint(self) -> str:
|
||
"""The key hint. It shows whether a message is selected and what Enter does to it.
|
||
|
||
Only the keys that do something show. ``^P paths`` must have a selected message to
|
||
show the paths *of*. ``^U QR`` must have a selected message with a URL in it
|
||
(refer to :meth:`_picked_urls`). ``^L map`` must have a selected message whose
|
||
author has a location (refer to :meth:`_locate_target`). ``^O author`` must have a
|
||
selected message whose author has a node page (refer to :meth:`_author_page`).
|
||
``^R retry failed`` must have something to retry (refer to :meth:`_retry_target`).
|
||
The F-key lane uses the same rules to dim its slots.
|
||
|
||
``^K lock`` (or ``^K unlock``) must have a selected message whose author can have a
|
||
lock (refer to :meth:`_lock_target`). It goes in only when there is room for it
|
||
(JP, 2026-10-19: "If there's room, put it."): the hint with it must stay in 72 cells
|
||
and keep each atom of the hint without it. It can make the words of an atom shorter
|
||
(:data:`_SHORT_ATOMS`), as ^U, ^L, and ^O already do. When it does not go in, the
|
||
status line still tells what the key did.
|
||
"""
|
||
hint = self._hint_with("")
|
||
lock = self._lock_target()
|
||
if lock is not None:
|
||
atom = "^K unlock" if lock[1] else "^K lock"
|
||
longer = self._hint_with(atom)
|
||
# ^E react gives way to ^K, as React gives way to Lock on the Shift half of F2.
|
||
if (
|
||
cell_len(longer) <= _HINT_CELLS
|
||
and atom in longer.split(" · ")
|
||
and _atom_set(hint) - {"^E react"} <= _atom_set(longer)
|
||
):
|
||
return longer
|
||
return hint
|
||
|
||
def _hint_with(self, lock: str) -> str:
|
||
"""The key hint, with the ``lock`` atom after ``^O author`` (or no atom for ``""``)."""
|
||
if self._selected is not None:
|
||
code = "^U QR" if self._picked_urls() else ""
|
||
where = "^L map" if self._locate_target() is not None else ""
|
||
who = "^O author" if self._author_page() is not None else ""
|
||
react = "^E react" if self._can_react() else ""
|
||
react_first = ("^E react",) if lock else ()
|
||
if self._is_channel and self._retry_target() is not None:
|
||
# A selected scoped message of ours can be sent again unscoped, and the hint
|
||
# must name that key. The "(@mention)" of the reply and ^End are removed to
|
||
# keep the line inside 72 cells (Esc still cancels the selection, and Enter
|
||
# still says what it does). If the line is still too long, ↑↓ goes first,
|
||
# then ^P, which the hint of each other selected message names, then ^O.
|
||
# The message is ours here, so ^O only opens our own page.
|
||
return _fit_hint(
|
||
[
|
||
"Enter reply",
|
||
"^P paths",
|
||
code,
|
||
where,
|
||
who,
|
||
lock,
|
||
react,
|
||
"^R resend unscoped",
|
||
"↑↓ pick",
|
||
"Esc cancel",
|
||
],
|
||
spare=(*react_first, "↑↓ pick", "^P paths", "^O author", "^E react"),
|
||
)
|
||
if self._replies:
|
||
# With all the keys of a selection, the long words do not fit in 72 cells.
|
||
# Then the hint uses the short words of the resend hint above.
|
||
hint = _fit_hint(
|
||
[
|
||
"Enter reply (@mention)",
|
||
"^P paths",
|
||
code,
|
||
where,
|
||
who,
|
||
lock,
|
||
react,
|
||
"↑↓ pick",
|
||
"^End/Esc cancel",
|
||
],
|
||
spare=("↑↓ pick",),
|
||
)
|
||
if cell_len(hint) <= _HINT_CELLS:
|
||
return hint
|
||
# ^E goes after ↑↓ when even the short words do not fit. The keys of the
|
||
# selection that were here before ^E keep their places.
|
||
return _fit_hint(
|
||
[
|
||
"Enter reply",
|
||
"^P paths",
|
||
code,
|
||
where,
|
||
who,
|
||
lock,
|
||
react,
|
||
"↑↓ pick",
|
||
"Esc cancel",
|
||
],
|
||
spare=(*react_first, "↑↓ pick", "^E react"),
|
||
)
|
||
return _fit_hint(
|
||
["Enter paths", code, where, who, lock, react, "↑↓ pick", "^End/Esc cancel"],
|
||
spare=(*react_first, "↑↓ pick", "^E react"),
|
||
)
|
||
if self._retry_target() is not None:
|
||
if self._is_channel:
|
||
return "Enter send · ↑ pick a message · ^R resend unscoped · Esc back"
|
||
return "Enter send · ↑ pick a message · ^R retry failed · Esc back"
|
||
return "Enter send · ↑ pick a message · Esc back"
|
||
|
||
# --- live updates --------------------------------------------------------
|
||
|
||
def append(self, message: ChatMessage) -> None:
|
||
"""Append an inbound message to the transcript and paint.
|
||
|
||
A reaction goes onto the message that it reacts to, and the transcript does not
|
||
move for it.
|
||
"""
|
||
if self._folds(message):
|
||
self._session.invalidate()
|
||
return
|
||
self._messages.append(message)
|
||
# Do not pull the viewport to the tail while the user selects a message to reply to.
|
||
if self._selected is None:
|
||
self._stick = True
|
||
self._session.invalidate()
|
||
|
||
# --- rendering -----------------------------------------------------------
|
||
|
||
def render_body(self, width: int) -> list[str]:
|
||
"""Render the transcript, a divider, the input line, and the status, if there is one."""
|
||
self._selected_line = None
|
||
self._selected_end = None
|
||
if not self._messages:
|
||
lines = render_lines(Text(self._empty_text, style="muted"), width)
|
||
else:
|
||
# Direct threads and channel threads use the same grouped transcript, in the
|
||
# style of Discord or Slack: consecutive messages from one sender share a
|
||
# coloured header, with day dividers between them. Because the render is for
|
||
# each message, we can also store where the selected message is (refer to
|
||
# _selected_line / cursor_line). The user can select in the two types of thread.
|
||
# The only difference is what Enter then does with the selection: a reply in a
|
||
# channel, or the delivery paths in a direct thread.
|
||
if self._selected is not None:
|
||
self._selected = max(0, min(self._selected, len(self._messages) - 1))
|
||
lines = self._render_grouped(width)
|
||
|
||
compose = self._compose_line(width)
|
||
footer_parts: list[RenderableType] = [
|
||
Text("─" * width, style="muted"),
|
||
compose,
|
||
]
|
||
if self._selected is not None:
|
||
footer_parts.append(self._pick_banner())
|
||
if self._status:
|
||
footer_parts.append(Text(self._status, style="muted"))
|
||
lines += render_lines(Group(*footer_parts), width)
|
||
|
||
# Stay at the bottom: give the frame an offset that is too large, so that it clamps
|
||
# the viewport to the last lines (input and latest messages). A scroll up clears
|
||
# the stick.
|
||
if self._stick:
|
||
self.scroll = len(lines)
|
||
elif self._selected_end is not None:
|
||
# The selection keeps its *full* message visible, with the codes under its text.
|
||
# The frame keeps only the one line that cursor_line names, the head of the
|
||
# message. With only that line, a selection that moves down stops the head on
|
||
# the bottom row, and the codes are below the viewport. Scroll far enough to
|
||
# show the last row of the message (one extra line for the day divider pinned
|
||
# at the top). The frame then moves the viewport up again only if that pushes
|
||
# the head off the top.
|
||
self.scroll = max(self.scroll, self._selected_end - self._scroll_viewport + 1)
|
||
return lines
|
||
|
||
def cursor_line(self) -> int | None:
|
||
"""Keep the selected reply target visible.
|
||
|
||
When nothing is selected, the scroll is free (the stick controls it).
|
||
"""
|
||
return self._selected_line
|
||
|
||
#: The text of an empty transcript.
|
||
_empty_text = "No messages yet — say hello!"
|
||
|
||
@property
|
||
def _replies(self) -> bool:
|
||
"""Whether Enter on a selected message replies to it (or else opens its paths).
|
||
|
||
This is true where there are many voices and an @mention says which one gets the
|
||
answer, and only while there is a compose line for the reply.
|
||
"""
|
||
return self._multiparty and self._composing
|
||
|
||
def _compose_line(self, width: int) -> RenderableType:
|
||
"""The compose line under the transcript: the input, with its byte limit.
|
||
|
||
The byte counter is pinned to the right edge of the *last* line of the input. Thus,
|
||
when the text wraps onto a second line, the counter stays in the bottom-right
|
||
corner and does not follow the cursor down the wrap. Only when the last line is
|
||
already full to the edge does the counter go onto its own right-aligned line.
|
||
|
||
Each complete mention in the line shows as its chip, the same chip as in the
|
||
transcript (refer to :class:`~meshterm.ui.compose.MentionEditor`). A chip never
|
||
breaks across two lines.
|
||
"""
|
||
limit = self._byte_limit()
|
||
input_line = self._editor.render_line(
|
||
self._mention_hop, overflow_at=self._overflow_at(limit)
|
||
)
|
||
whole = [(s.start, s.end) for s in input_line.spans if str(s.style) == PATH_INK]
|
||
return right_aligned_tail(input_line, self._byte_counter(limit), width, whole=whole)
|
||
|
||
# --- outgoing byte limit -------------------------------------------------
|
||
|
||
def _byte_limit(self) -> int:
|
||
"""The UTF-8 byte limit for a message in this conversation (channel or direct)."""
|
||
return CHANNEL_BYTE_LIMIT if self._is_channel else DM_BYTE_LIMIT
|
||
|
||
def _used_bytes(self) -> int:
|
||
"""The UTF-8 byte length of the current compose buffer: what counts against the limit."""
|
||
return len(self._editor.text.encode("utf-8"))
|
||
|
||
def _overflow_at(self, limit: int) -> int | None:
|
||
"""The index of the first compose character that cannot be sent, or ``None``.
|
||
|
||
With no parts, this is the first character with bytes past ``limit``. With parts,
|
||
it is the first character that does not fit in the last part (refer to
|
||
:func:`~meshterm.core.parts.overflow_at`). The count goes by character (not by
|
||
byte), so multibyte input stays complete: an emoji or an accented letter is fully
|
||
under or fully over the line, never divided in the middle of its sequence.
|
||
"""
|
||
return parts.overflow_at(self._editor.text, limit, self._max_parts)
|
||
|
||
def _parts(self, text: str) -> list[str] | None:
|
||
"""The messages that ``text`` goes out as, or ``None`` when it has too many parts."""
|
||
return parts.split(text, self._byte_limit(), self._max_parts)
|
||
|
||
def _byte_counter(self, limit: int) -> Text:
|
||
"""The inline ``used/limit`` counter: the shared compose gauge (:func:`byte_counter`).
|
||
|
||
When the text goes out as more than one part, the counter starts with the number
|
||
of parts, and the gauge shows the bytes of the last part, with its part number.
|
||
Thus the gauge tells how much the user can add before one more part starts.
|
||
"""
|
||
cut = self._parts(self._editor.text.strip())
|
||
if cut is None or len(cut) < 2:
|
||
return byte_counter(self._used_bytes(), limit)
|
||
counter = Text(f"{len(cut)} parts · ", style="muted")
|
||
counter.append_text(byte_counter(len(cut[-1].encode("utf-8")), limit))
|
||
return counter
|
||
|
||
def _name(self, peer: str | None) -> str | None:
|
||
"""Find the contact name for a sender key prefix (exact match, then prefix match)."""
|
||
if not peer:
|
||
return None
|
||
needle = peer.lower()
|
||
if needle in self._names:
|
||
return self._names[needle]
|
||
for prefix, name in self._names.items():
|
||
if prefix.startswith(needle) or needle.startswith(prefix):
|
||
return name
|
||
return None
|
||
|
||
# --- grouped rendering ---------------------------------------------------
|
||
|
||
def _render_grouped(self, width: int) -> list[str]:
|
||
"""Render the transcript to ANSI lines, and keep the row of the selected message.
|
||
|
||
This serves direct threads and channel threads. Consecutive messages from the same
|
||
sender on the same day share one coloured header, with each message body indented
|
||
below it. A divider marks each new day, and a blank line separates two different
|
||
sender groups. The render goes one message at a time (not as one Group). Thus we
|
||
can store the body line of the selected message in :attr:`_selected_line`, so that
|
||
the frame can scroll it into the viewport. The user can select in the two types of
|
||
thread (refer to :meth:`handle`).
|
||
|
||
Things that change nothing here cause a paint all the time: the idle tick, a key
|
||
press, the ack spinner. Thus this method caches the rendered lines of each message
|
||
in :attr:`_render_cache`, and for each message it puts in the cached slice instead
|
||
of a new render. A message renders again only when this is necessary:
|
||
|
||
* it still waits for its ack (the spinner draws its last glyph again at each
|
||
tick),
|
||
* it is the selected message (the selection sets its style, not the message), or
|
||
* its identity or ``acked`` is different from the cache (it was appended, it was
|
||
replaced in place when a pending bubble resolved, or it was removed), or
|
||
* its number of reactions is different from the cache (a reaction arrived).
|
||
|
||
A change of the locks draws all of the transcript again, because the sender chip of
|
||
a locked contact has a padlock (refer to :meth:`_group_header`). The locks change
|
||
rarely: ^K here, or the lock row of the node page of an author.
|
||
|
||
Most importantly, this works for each message. It is not a cached prefix that
|
||
stops at the first such row. Thus a message stays cheap to draw again, however far
|
||
back it is in a long transcript. With a prefix that stops, a page up through the
|
||
history causes a full render again at each key press. With this method, it does
|
||
not. The grouping context (day and sender headers) is cheap to calculate again in
|
||
all cases. Thus it is calculated new each time, and never taken from the cache.
|
||
"""
|
||
messages = self._messages
|
||
total = len(messages)
|
||
cache = self._render_cache
|
||
spacing = self.section_spacing()
|
||
locks = self._pages.locked_keys() if self._pages is not None else frozenset()
|
||
if (
|
||
cache is not None
|
||
and cache["width"] == width
|
||
and cache["spacing"] == spacing
|
||
and cache["locks"] == locks
|
||
):
|
||
cached_ids = cache["ids"]
|
||
cached_acked = cache["acked"]
|
||
cached_selected = cache["selected"]
|
||
cached_reacts = cache["reacts"]
|
||
cached_boundaries = cache["boundaries"]
|
||
cached_lines = cache["lines"]
|
||
cache_count = len(cached_ids)
|
||
else:
|
||
cached_ids = cached_acked = cached_selected = cached_reacts = cached_boundaries = ()
|
||
cached_lines = []
|
||
cache_count = 0
|
||
|
||
lines: list[str] = []
|
||
# Store each day divider as a sticky block of one row. Thus, when the divider of
|
||
# the top visible message scrolls off, it is pinned to the top row again. This is
|
||
# the same base Screen.sticky_block that the conversation picker uses for its
|
||
# section headings. A day has nothing to say except its date, so the block is only
|
||
# the divider. The heading of a select list can also have its description.
|
||
self._sticky_headers = []
|
||
ids: list[int] = []
|
||
ackeds: list[bool | None] = []
|
||
selecteds: list[bool] = []
|
||
reacts: list[int] = []
|
||
boundaries: list[int] = []
|
||
prev_group: tuple[bool, str] | None = None
|
||
prev_day = None
|
||
# The first message of the group of the selected message. Its slice has the author
|
||
# chip of the group, which is white while the selection is in the group.
|
||
picked_head = self._picked_head()
|
||
|
||
for idx in range(total):
|
||
message = messages[idx]
|
||
stamp = message.created_at.astimezone()
|
||
day = stamp.date()
|
||
sender, body = self._sender_and_body(message)
|
||
# Group by (are-we-the-sender, display name), not by the name alone. Thus our
|
||
# messages never join with those of a remote sender that has the same name.
|
||
group = (message.outbound, sender)
|
||
new_day = day != prev_day
|
||
selected = idx == self._selected
|
||
# A slice that the selection changes renders again, now and on the next paint
|
||
# after the selection leaves it: the selected message, and the head of its group.
|
||
marked = selected or idx == picked_head
|
||
pending = message.acked is None and message.outbound and not message.is_channel
|
||
react_count = self._fold.count(message) if self._fold is not None else 0
|
||
reusable = (
|
||
not marked
|
||
and not pending
|
||
and idx < cache_count
|
||
and not cached_selected[idx]
|
||
and cached_ids[idx] == id(message)
|
||
and cached_acked[idx] == message.acked
|
||
and cached_reacts[idx] == react_count
|
||
)
|
||
# A blank line closes the messages of an author and the messages of a day, where
|
||
# the preferences ask for it (:meth:`~meshterm.ui.tui.screen.Screen.section_spacing`,
|
||
# the same test as the section headings of a list). It is the first line of the
|
||
# slice of the message that starts the next group, so a cached slice brings it
|
||
# back with it.
|
||
gap = spacing and bool(lines) and (new_day or group != prev_group)
|
||
if reusable:
|
||
start = cached_boundaries[idx - 1] if idx > 0 else 0
|
||
slice_lines = cached_lines[start : cached_boundaries[idx]]
|
||
if new_day:
|
||
at = int(gap) # the divider comes after the gap
|
||
self._sticky_headers.append((len(lines) + at, [slice_lines[at]]))
|
||
lines += slice_lines
|
||
else:
|
||
if gap:
|
||
lines.append("")
|
||
if new_day:
|
||
# A day divider is the section heading of this transcript: it pins to the
|
||
# top row while its day scrolls under it, and ^PgUp/^PgDn step by it.
|
||
# Thus it has the heading style of the app (``── Label ──`` in the bold
|
||
# ``heading`` grey, refer to menus.section_heading). The rule above the
|
||
# compose line is plain ``muted``, because that rule is chrome.
|
||
divider = render_lines(
|
||
Text(f"── {stamp:%a} {stamp:%b} {stamp.day} ──", style="heading"), width
|
||
)
|
||
self._sticky_headers.append((len(lines), [divider[0]]))
|
||
lines += divider
|
||
if new_day or group != prev_group:
|
||
header = self._group_header(sender, message, picked=idx == picked_head)
|
||
lines += render_lines(header, width)
|
||
if selected:
|
||
# The first message has the head of the transcript: nothing comes before
|
||
# its day divider. Thus its selection anchors at line 0, not at its own
|
||
# body. When the anchor was on the body, the viewport at the very top
|
||
# was scrolled down two lines: the divider was only pinned, the sender
|
||
# chip was not visible, and the ↑ of the title bar was lit over a
|
||
# transcript with nothing above it.
|
||
self._selected_line = 0 if idx == 0 else len(lines)
|
||
lines += self._body_lines(body, message, width, selected=selected)
|
||
if selected:
|
||
self._selected_end = len(lines)
|
||
prev_group, prev_day = group, day
|
||
ids.append(id(message))
|
||
ackeds.append(message.acked)
|
||
selecteds.append(marked)
|
||
reacts.append(react_count)
|
||
boundaries.append(len(lines))
|
||
|
||
self._render_cache = {
|
||
"width": width,
|
||
"spacing": spacing,
|
||
"locks": locks,
|
||
"ids": ids,
|
||
"acked": ackeds,
|
||
"selected": selecteds,
|
||
"reacts": reacts,
|
||
"lines": lines,
|
||
"boundaries": boundaries,
|
||
}
|
||
return lines
|
||
|
||
def _sender_and_body(self, message: ChatMessage) -> tuple[str, str]:
|
||
"""Return the display sender and the clean body of a message.
|
||
|
||
Our messages are ``you``. An inbound channel message has a ``Name: `` prefix, which
|
||
we move into the sender (``·`` when there is no prefix). An inbound direct message
|
||
gets the sender from the resolved contact name (or the raw key, or ``?``).
|
||
"""
|
||
if message.outbound:
|
||
return "you", message.text
|
||
if message.is_channel:
|
||
name, body = split_channel_sender(message.text)
|
||
return (name or "·"), body
|
||
return (self._name(message.peer) or message.peer or "?"), message.text
|
||
|
||
def _picked_head(self) -> int | None:
|
||
"""The index of the first message of the group of the selected message, or ``None``.
|
||
|
||
A group is a run of messages from one sender on one day, under one author chip
|
||
(refer to :meth:`_render_grouped`). The chip is drawn with the first message of the
|
||
run, so that message is the one to draw again when the selection is in the run.
|
||
"""
|
||
selected, messages = self._selected, self._messages
|
||
if selected is None or not 0 <= selected < len(messages):
|
||
return None
|
||
|
||
def group_of(message: ChatMessage) -> tuple:
|
||
day = message.created_at.astimezone().date()
|
||
return (day, message.outbound, self._sender_and_body(message)[0])
|
||
|
||
group = group_of(messages[selected])
|
||
head = selected
|
||
while head > 0 and group_of(messages[head - 1]) == group:
|
||
head -= 1
|
||
return head
|
||
|
||
def _group_header(self, sender: str, message: ChatMessage, *, picked: bool = False) -> Text:
|
||
"""Build the sender header that starts a group: the name, drawn as a chip.
|
||
|
||
The chip makes a *label* different from the prose under it. A name alone above its
|
||
messages then reads as a tag, not as a coloured word that a person typed. It is the
|
||
chip of a path line (refer to :func:`~meshterm.ui.widgets.name_chip`). Thus a
|
||
sender looks exactly like that node as a hop in a route.
|
||
|
||
Our messages are the exception. They go under our name, as ``★ Name`` on a chip
|
||
in white (refer to :func:`~meshterm.ui.widgets.self_chip`). A route
|
||
shows our end as the bare ``★``, but a chat is a conversation between names.
|
||
Our chip comes from the direction of the message (``outbound``), not from the
|
||
``"you"`` label. Thus a remote sender with the name ``you`` still gets the hue of
|
||
its key, and does not look like our node.
|
||
|
||
The sender is the origin of the route of each message under it. Thus the chip
|
||
closes on the point, the mark of a route that goes on. An ``@mention`` in a body is
|
||
the other end, and it opens on the notch (refer to :meth:`_render_mentions`).
|
||
|
||
A sender that is a locked contact has the padlock before its name, in the place of
|
||
the ``★`` of our chip (JP, 2026-10-19). Thus the chip shows the state that ^K
|
||
toggles.
|
||
|
||
With ``picked``, the selected message is in this group, and the label of the chip
|
||
is white, as the text of that message is (refer to
|
||
:func:`~meshterm.ui.widgets.picked_hop`, JP, 2026-10-19). Our chip already has a
|
||
white label.
|
||
"""
|
||
if message.outbound:
|
||
return self_chip(self._self_name, self._self_key, role=ORIGIN)
|
||
# ``·`` is a channel line that arrived with no sender prefix: there is no key to use.
|
||
key = None if sender == "·" else self._header_key(sender, message)
|
||
pages = self._pages
|
||
locked = key is not None and pages is not None and pages.lock_state(pages.find(key)) is True
|
||
return author_chip(sender, key, locked=locked, picked=picked)
|
||
|
||
def _header_key(self, sender: str, message: ChatMessage) -> str | None:
|
||
"""The key that gives the colour of the sender chip of a group: the key of the name.
|
||
|
||
This is a hook, not a call, because a room knows more than a name. Each post has the
|
||
key prefix of its author, which gives the colour of the chip with no name lookup
|
||
(refer to :class:`~meshterm.ui.room.RoomScreen`).
|
||
"""
|
||
return self._sender_key(sender)
|
||
|
||
def _body_lines(
|
||
self, body: str, message: ChatMessage, width: int, *, selected: bool = False
|
||
) -> list[str]:
|
||
"""Render one message to ANSI lines, with its own time and a hanging indent.
|
||
|
||
The timestamp of each message is here (not on the group header), so that each
|
||
message shows the time when it was sent, also when one sender has many messages in
|
||
a group. Without this, a run of messages from the same sender seems to share one
|
||
time. A long body wraps with a hanging indent, so that the next lines align under
|
||
the body and not under the timestamp gutter. When ``selected`` is true, the line
|
||
is marked as the reply target (the same as the ``❯`` pointer and the highlight of
|
||
the select screen).
|
||
|
||
A message that mentions our name has a mark at the left of its time: the white
|
||
``★`` of our node (refer to :func:`~meshterm.ui.widgets.mention_mark`, JP,
|
||
2026-10-13). The chip of the mention in the body has our white label, but no star
|
||
(refer to :func:`~meshterm.ui.widgets.mention_hop`). Thus the mark of a message that
|
||
names us is in one column, where the eye finds it when it goes down the transcript.
|
||
Each time has the two cells of that column in front of it, so that all the times stay
|
||
in one column too.
|
||
|
||
A URL is never cut where the screen can hold it. A URL that is too long for the
|
||
body lane moves out to the column after the ``❯``, on its own line (refer to
|
||
:func:`~meshterm.ui.tui.render.render_hanging`). This is because a terminal opens
|
||
a link, and a user copies one, only while it is on one line.
|
||
|
||
Where the platform draws them (:data:`_URL_CODES`, the PicoCalc), each URL in the
|
||
body gets a QR code under it. The codes are side by side, at the indent of the
|
||
body. Thus the user can open a link from the handheld on a phone, and does not
|
||
have to type it.
|
||
"""
|
||
stamp = message.created_at.astimezone()
|
||
prefix = Text()
|
||
prefix.append("❯ " if selected else " ", style="cursor" if selected else None)
|
||
if self._names_us(body):
|
||
prefix.append_text(mention_mark(self._self_key))
|
||
prefix.append(" ")
|
||
else:
|
||
prefix.append(" ")
|
||
prefix.append(f"{stamp:%H:%M} ", style="cursor" if selected else "muted")
|
||
body_text = self._body_text(body, message, selected=selected)
|
||
whole = [m.span() for m in _WORD.finditer(body_text.plain) if links.urls(m.group())]
|
||
# A mention is a chip, and a chip never breaks across two lines.
|
||
whole += [(s.start, s.end) for s in body_text.spans if str(s.style) == PATH_INK]
|
||
lines = render_hanging(
|
||
prefix, body_text, width, indent=prefix.cell_len, whole=whole, gutter=_URL_GUTTER
|
||
)
|
||
urls = links.urls(body) if _URL_CODES else []
|
||
if urls:
|
||
lines += render_lines(qr_strip(urls, width, indent=prefix.cell_len), width)
|
||
badges = self._fold.badges(message) if self._fold is not None else []
|
||
if badges:
|
||
row, runs = self._badge_text(badges, selected=selected)
|
||
lines += render_hanging(
|
||
Text(" " * prefix.cell_len), row, width, indent=prefix.cell_len, whole=runs
|
||
)
|
||
return lines
|
||
|
||
def _badge_text(
|
||
self, badges: list[reactions.Badge], *, selected: bool
|
||
) -> tuple[Text, list[tuple[int, int]]]:
|
||
"""The reactions on one message, as one line under its body, and the runs to keep whole.
|
||
|
||
Each emoji has its count after it. The count is ``accent`` when our node is one of
|
||
the nodes that reacted, else ``muted``. On the selected message, the names of the
|
||
nodes take the place of the count, each in the hue of its node, and our node is the
|
||
white ``you``. Where the font has no emoji, an emoticon stands in for each default
|
||
reaction (refer to :func:`~meshterm.ui.theme.emoticon`).
|
||
"""
|
||
text = Text()
|
||
runs: list[tuple[int, int]] = []
|
||
for index, badge in enumerate(badges):
|
||
if index:
|
||
text.append(" ")
|
||
start = len(text)
|
||
text.append(emoticon(badge.emoji))
|
||
text.append(" ")
|
||
if not selected:
|
||
count = f"{_COUNT_MARK}{len(badge.reactors)}"
|
||
text.append(count, style="accent" if badge.ours else "muted")
|
||
runs.append((start, len(text)))
|
||
continue
|
||
for number, reactor in enumerate(badge.reactors):
|
||
if number:
|
||
text.append(", ", style="muted")
|
||
if reactor.ours:
|
||
text.append("you", style="you")
|
||
else:
|
||
name = reactor.name or self._peer_label
|
||
key = self._peer_key if reactor.name is None else self._sender_key(name)
|
||
text.append(name, style=name_style(name, key or None))
|
||
if not number:
|
||
runs.append((start, len(text))) # the emoji never ends a line alone
|
||
return text, runs
|
||
|
||
def _body_text(self, body: str, message: ChatMessage, *, selected: bool) -> Text:
|
||
"""Build the styled body of a message: mentions in colour, then the glyphs at the end."""
|
||
text = self._render_mentions(body, selected=selected)
|
||
if message.outbound:
|
||
# A direct message shows its own delivery (spinner → ✓/✗, and a retry is
|
||
# possible). A channel broadcast has no ack. Thus mark only a channel message
|
||
# that did not leave the companion.
|
||
if message.is_channel:
|
||
if message.acked is False:
|
||
text.append(" ⚠ no ack", style="warn")
|
||
text.append_text(self._scope_note(message))
|
||
else:
|
||
text.append(" ")
|
||
text.append_text(self._delivery_glyph(message.acked))
|
||
if message.snr is not None:
|
||
text.append(f" {message.snr:+.0f} dB", style=snr_style(message.snr))
|
||
return text
|
||
|
||
def _scope_note(self, message: ChatMessage) -> Text:
|
||
"""A muted note after a sent message that went out under a scope not in the title.
|
||
|
||
This note shows only in a channel whose title names a scope, and only for the
|
||
exceptions to that scope: an unscoped resend, or a message sent before the scope
|
||
was set or changed. On each other line, a note only repeats the title. In
|
||
a channel with no scope of its own, the default scope of its messages belongs to
|
||
the device. It is not news to show on each line.
|
||
"""
|
||
if not self._scope or not message.scope or message.scope == self._scope:
|
||
return Text()
|
||
if message.scope == WILDCARD:
|
||
return Text(" · unscoped", style="muted")
|
||
return Text(f" · scope {message.scope}", style="muted")
|
||
|
||
def _render_mentions(self, body: str, *, selected: bool) -> Text:
|
||
"""Render body text, and change each ``@[Name]`` token to the chip of that node.
|
||
|
||
The reply flow puts an ``@[Name]`` token in the compose line (refer to
|
||
:meth:`_begin_reply`). Here it shows as the name in a chip of the hue of that node.
|
||
Thus the user sees that a mention belongs to the node that it names. The text
|
||
around the mentions keeps the base style of the line (``cursor`` when the message
|
||
is the selected reply target, or else no style). A mention of our own name has our
|
||
chip, in white, and the line has our white ``★`` at the left of its time
|
||
(refer to :meth:`_body_lines`). Thus a message that names us is easy to see.
|
||
|
||
The chip opens on the notch, because the node that a mention names is the
|
||
destination of a route that came from the sender (refer to :meth:`_group_header`,
|
||
whose chip closes on the point). The chip has no ``@``, because the chip already
|
||
says that the name is a mention (JP, 2026-10-13). Arrow mode has no chip, so there a
|
||
mention keeps its ``@`` (refer to :func:`~meshterm.ui.widgets.mention_hop`).
|
||
|
||
On the selected message, the label of each chip is white, as the text around it is
|
||
(refer to :func:`~meshterm.ui.widgets.picked_hop`, JP, 2026-10-19).
|
||
"""
|
||
base = "cursor" if selected else None
|
||
text = Text()
|
||
pos = 0
|
||
for match in MENTION.finditer(body):
|
||
if match.start() > pos:
|
||
text.append(body[pos : match.start()], style=base)
|
||
hop = self._mention_hop(match.group(1))
|
||
if selected:
|
||
hop = picked_hop(hop) # the chips turn white with the text around them
|
||
text.append_text(hop_chip(hop, role=DESTINATION))
|
||
pos = match.end()
|
||
if pos < len(body):
|
||
text.append(body[pos:], style=base)
|
||
return text
|
||
|
||
def _mention_hop(self, name: str) -> PathHop:
|
||
"""The hop of a mention of ``name``: in the transcript, and in the compose line.
|
||
|
||
The two use this one method, so a chip in the compose line is the chip that the
|
||
message will show after the send. A mention of our name has our key and the chip
|
||
with the white label. Each other name has the key that the names of the senders
|
||
give (refer to :meth:`_sender_key`).
|
||
"""
|
||
if self._self_name and name == self._self_name:
|
||
return mention_hop(name, self._self_key, you=True)
|
||
return mention_hop(name, self._sender_key(name, mention=True))
|
||
|
||
def _names_us(self, body: str) -> bool:
|
||
"""Whether ``body`` has a mention of our name (it gets a mark before its time)."""
|
||
if not self._self_name:
|
||
return False
|
||
return any(match.group(1) == self._self_name for match in MENTION.finditer(body))
|
||
|
||
def _delivery_glyph(self, acked: bool | None) -> Text:
|
||
"""Map the ``acked`` state of an outbound direct message to the mark at its end.
|
||
|
||
A message that still waits for its ack (``acked is None``) shows the current
|
||
animation step of the spinner. :meth:`_spin_while` animates it while the send is in
|
||
flight. A resolved message shows the delivered ``✓`` (ok) or the unacknowledged
|
||
``✗`` (err).
|
||
"""
|
||
if acked is None:
|
||
return self._spinner.text()
|
||
glyph, style = _DELIVERED if acked else _FAILED
|
||
return Text(glyph, style=style)
|
||
|
||
def _pick_banner(self) -> Text:
|
||
"""A one-line note above the input while a message is selected.
|
||
|
||
In a channel, it starts with the reply (what Enter does there). In a direct chat,
|
||
it starts with the paths view (what Enter does there). The two say what the
|
||
selection is for. Thus the user never sees a highlight that has no explanation.
|
||
"""
|
||
message = self._messages[self._selected]
|
||
sender, _ = self._sender_and_body(message)
|
||
who = "this message" if message.outbound or sender == "·" else sender
|
||
if self._replies:
|
||
return Text(
|
||
f"↩ Enter to reply to {who} with an @mention · End to cancel",
|
||
style="accent",
|
||
)
|
||
return Text("Enter to see the paths this message took · End to cancel", style="accent")
|
||
|
||
def _sender_key(self, sender: str, *, mention: bool = False) -> str | None:
|
||
"""The key of a sender name, or ``None`` when nothing gives a key for it.
|
||
|
||
The contacts first, then the names that the recorder stored. In a *direct* thread,
|
||
a sender label with no key falls back to the key of the peer, because the only two
|
||
persons in the conversation are we and the peer. A mention names any node, so
|
||
that fallback must not apply to a mention. Thus a mention that is not resolved
|
||
stays grey in a direct chat, exactly as it does in a channel.
|
||
|
||
A name with no key is drawn in the *node* grey, never in ``muted`` (the chip does
|
||
this, refer to :func:`~meshterm.ui.widgets.name_chip`). An unidentified sender is
|
||
content that you can still do actions on (select the message, open its paths),
|
||
not chrome. On the console, ``muted`` is the dark slot and ``node.unknown`` is the
|
||
light one (JP, 2026-08-12).
|
||
|
||
The sender chip (:meth:`_group_header`) and the mention chips
|
||
(:meth:`_render_mentions`) share this method. Thus the two cannot disagree about
|
||
the identity of a name.
|
||
"""
|
||
key = self._key_of(sender)
|
||
if not key and not self._multiparty and not mention:
|
||
key = self._peer_key or None
|
||
return key
|
||
|
||
# --- input ---------------------------------------------------------------
|
||
|
||
def handle(self, action: str, data: str = "") -> None:
|
||
"""Dispatch a key: select messages and do actions on them, edit the compose line, or leave.
|
||
|
||
The two types of chat share one model. When no message is selected, Enter sends,
|
||
and typing edits the compose line. An amber dialog asks the user to confirm a paste
|
||
(Ctrl-V, or the bracketed paste of a terminal) before the paste goes into the line
|
||
(refer to :meth:`_begin_paste`). ↑ selects the newest message. The selection then
|
||
moves with ↑↓, PgUp/PgDn (one screen), Ctrl+Home (the very first message), and
|
||
Ctrl+PgUp/PgDn (day dividers), and the viewport moves with it. Enter on a selected
|
||
message starts a reply with an ``@mention`` in a channel, and opens the delivery
|
||
paths in a direct chat. In the two types of chat, ^P opens the paths of the
|
||
selected message, ^U the QR codes of its links, ^L the map on its author, and ^E the
|
||
reaction dialog. ^K locks or unlocks its author. These keys do nothing when no
|
||
message is selected. ^End (or a move past the newest message) goes back to the
|
||
compose line. The first Esc clears the selection, and the second Esc leaves the
|
||
screen.
|
||
"""
|
||
if action == "enter":
|
||
if self._selected is not None:
|
||
if self._replies:
|
||
self._begin_reply()
|
||
else:
|
||
self._open_paths(self._selected)
|
||
else:
|
||
self._submit()
|
||
elif action == "paste":
|
||
self._begin_paste(data)
|
||
elif action == "paths":
|
||
self._open_paths(self._selected)
|
||
elif action == "url_code":
|
||
self._open_codes()
|
||
elif action == "locate":
|
||
self._open_map()
|
||
elif action == "author_page":
|
||
self._open_author_page()
|
||
elif action == "react":
|
||
self._begin_react()
|
||
elif action == "lock":
|
||
self._toggle_lock()
|
||
elif action == "retry":
|
||
if self._is_channel:
|
||
self._begin_resend_unscoped()
|
||
else:
|
||
self._retry()
|
||
elif action == "escape":
|
||
if self._selected is not None:
|
||
self._clear_selection() # the first Esc clears the selection, the next leaves
|
||
self._session.invalidate()
|
||
else:
|
||
self.resolve(CANCEL)
|
||
elif action == "up":
|
||
self._move_selection(-1)
|
||
elif action == "pageup":
|
||
self._move_selection(-self._page_step)
|
||
elif action == "down":
|
||
self._move_selection(1)
|
||
elif action == "pagedown":
|
||
self._move_selection(self._page_step)
|
||
elif action == "ctrl_home":
|
||
if self._messages:
|
||
self._selected = 0
|
||
self._stick = False
|
||
elif action == "ctrl_pageup":
|
||
self._select_section(-1)
|
||
elif action == "ctrl_pagedown":
|
||
self._select_section(1)
|
||
elif action == "ctrl_end":
|
||
self._clear_selection()
|
||
self._stick = True # jump back to the live tail and the compose line
|
||
else:
|
||
if self._composing and self._editor.edit(action, data):
|
||
# An edit of the compose line puts the focus there. Nothing stays selected.
|
||
self._status = "" # an edit clears the "too long" notice
|
||
self._clear_selection()
|
||
self._stick = True
|
||
|
||
def _open_paths(self, index: int | None) -> None:
|
||
"""Float the delivery-paths view for the selected message (one dialog at a time).
|
||
|
||
``index`` is the selection. Thus ``None`` (nothing selected) opens nothing. A path
|
||
is a route that one *specific* message walked. When the code guessed the latest
|
||
message, it showed the paths of the message that was at the bottom of the
|
||
transcript at that time.
|
||
"""
|
||
if index is None or not self._messages or self._paths is None or self._paths_open:
|
||
return
|
||
message = self._messages[max(0, min(index, len(self._messages) - 1))]
|
||
self._paths_open = True
|
||
|
||
async def run() -> None:
|
||
try:
|
||
await self._paths(message)
|
||
finally:
|
||
self._paths_open = False
|
||
self._session.invalidate()
|
||
|
||
self._session.run_detached(run())
|
||
|
||
def _locate_target(self) -> MapMarker | None:
|
||
"""The marker of the author of the selected message, or ``None`` when ^L does nothing.
|
||
|
||
The author is the node whose key gives the colour of the chip of the message (refer
|
||
to :meth:`_header_key`). Our own message has our node as its author. A message with
|
||
no sender, or an author that has no location, has no marker.
|
||
"""
|
||
if self._locator is None or self._selected is None or not self._messages:
|
||
return None
|
||
message = self._messages[max(0, min(self._selected, len(self._messages) - 1))]
|
||
if message.outbound:
|
||
return self._locator.you()
|
||
sender, _body = self._sender_and_body(message)
|
||
if sender == "·":
|
||
return None
|
||
return self._locator.find(self._header_key(sender, message))
|
||
|
||
def _open_map(self) -> None:
|
||
"""Open the map on the author of the selected message (^L). One map at a time.
|
||
|
||
The map is a screen, and it goes on the stack above the chat. Esc on the map comes
|
||
back to the chat, with the same message selected.
|
||
"""
|
||
target = self._locate_target()
|
||
if target is None or self._locator is None or self._map_open:
|
||
return
|
||
locator = self._locator
|
||
self._map_open = True
|
||
|
||
async def run() -> None:
|
||
try:
|
||
await locator.show(target)
|
||
finally:
|
||
self._map_open = False
|
||
self._session.invalidate()
|
||
|
||
self._session.run_detached(run())
|
||
|
||
def _author_page(self) -> Callable[[], Awaitable[None]] | None:
|
||
"""The node page that ^O opens for the selected message, or ``None`` when there is none.
|
||
|
||
The page belongs to the author of the message: our node for a message of ours, or
|
||
the contact that the key of the author finds (refer to :meth:`_header_key`, the
|
||
same key that gives the colour of the sender chip). A message with no sender, or an
|
||
author that is not a contact, has no page. The hint and the F-key lane ask this at
|
||
each paint, so that they name ^O only when it acts.
|
||
"""
|
||
pages = self._pages
|
||
if pages is None or self._selected is None or not self._messages:
|
||
return None
|
||
message = self._messages[max(0, min(self._selected, len(self._messages) - 1))]
|
||
if message.outbound:
|
||
return lambda: pages.show(None)
|
||
contact = self._author_contact()
|
||
return None if contact is None else (lambda: pages.show(contact))
|
||
|
||
def _author_contact(self) -> Contact | None:
|
||
"""The contact that wrote the selected message, or ``None``.
|
||
|
||
The key of the author finds it (refer to :meth:`_header_key`, the same key that
|
||
gives the colour of the sender chip). A message of ours, a message with no sender,
|
||
and an author that is not a contact have none.
|
||
"""
|
||
pages = self._pages
|
||
if pages is None or self._selected is None or not self._messages:
|
||
return None
|
||
message = self._messages[max(0, min(self._selected, len(self._messages) - 1))]
|
||
if message.outbound:
|
||
return None
|
||
sender, _body = self._sender_and_body(message)
|
||
if sender == "·":
|
||
return None
|
||
return pages.find(self._header_key(sender, message))
|
||
|
||
def _lock_target(self) -> tuple[Contact, bool] | None:
|
||
"""The author that ^K locks or unlocks, and whether it is locked now.
|
||
|
||
``None`` when ^K does nothing: no selected message, a message of ours, an author
|
||
that is not a contact, or a contact that cannot have a lock (refer to
|
||
:meth:`~meshterm.ui.node_detail_screen.ContactPages.lock_state`). The hint and the
|
||
F-key lane ask at each paint, and the store answers from memory.
|
||
"""
|
||
contact = self._author_contact()
|
||
if contact is None or self._pages is None:
|
||
return None
|
||
locked = self._pages.lock_state(contact)
|
||
return None if locked is None else (contact, locked)
|
||
|
||
def _toggle_lock(self) -> None:
|
||
"""Lock the author of the selected message, or unlock it when it is locked (^K).
|
||
|
||
There is no confirm, because the next ^K reverses it, as the row on the node page
|
||
does. The hint atom and the lane chip change to their opposite. The status line also
|
||
tells what the key did, because the hint has no room for the atom on each message.
|
||
"""
|
||
target = self._lock_target()
|
||
if target is None or self._pages is None:
|
||
return
|
||
contact, locked = target
|
||
self._pages.set_locked(contact, not locked)
|
||
name = contact.name or contact.key_prefix
|
||
self._status = f"Unlocked {name}" if locked else f"Locked {name} — no sweep archives it"
|
||
self._session.invalidate()
|
||
|
||
def _open_author_page(self) -> None:
|
||
"""Open the node page of the author of the selected message (^O). One page at a time.
|
||
|
||
The page is a screen, and it goes on the stack above the chat. Esc on the page comes
|
||
back to the chat, with the same message selected.
|
||
"""
|
||
page = self._author_page()
|
||
if page is None or self._page_open:
|
||
return
|
||
self._page_open = True
|
||
|
||
async def run() -> None:
|
||
try:
|
||
await page()
|
||
finally:
|
||
self._page_open = False
|
||
self._session.invalidate()
|
||
|
||
self._session.run_detached(run())
|
||
|
||
def _picked_urls(self) -> list[str]:
|
||
"""The URLs in the body of the selected message, in order. None if nothing is selected."""
|
||
if self._selected is None or not self._messages:
|
||
return []
|
||
message = self._messages[max(0, min(self._selected, len(self._messages) - 1))]
|
||
return links.urls(self._sender_and_body(message)[1])
|
||
|
||
def _open_codes(self) -> None:
|
||
"""Show the links of the selected message as QR codes on the share screen (^U).
|
||
|
||
This is the share screen of a channel or of a contact card
|
||
(:class:`~meshterm.ui.qr.QrScreen`), full-frame. Thus a phone can read a link from
|
||
here and open it. Many links are on one screen that ←→ step through, in the order of
|
||
the message. The same as ^P, it acts on the *selection*: a link belongs to one
|
||
message, and when nothing is selected, the screen cannot know which message. One
|
||
share screen at a time.
|
||
"""
|
||
urls = self._picked_urls()
|
||
if not urls or self._codes_open:
|
||
return
|
||
self._codes_open = True
|
||
|
||
async def run() -> None:
|
||
try:
|
||
await self._session.run_screen(QrScreen(*urls, title="Links"))
|
||
finally:
|
||
self._codes_open = False
|
||
self._session.invalidate()
|
||
|
||
self._session.run_detached(run())
|
||
|
||
def _begin_paste(self, data: str) -> None:
|
||
"""Confirm a clipboard paste on an amber dialog, then insert it into the compose line.
|
||
|
||
A chat message goes out over the air. A paste can be much larger than a key press,
|
||
or it can have content that the user did not want to broadcast. Thus a paste must
|
||
go through a Cancel/Paste confirm (the amber ``danger`` tier: disruptive, not a loss
|
||
of data), and does not go directly into the line. Newlines and control characters
|
||
become spaces (a message is one line). An empty result never opens the dialog. One
|
||
paste dialog at a time, scheduled from the key handler, the same as the paths view.
|
||
"""
|
||
if self._paste_open or not self._composing:
|
||
return
|
||
clean = "".join(ch if ch.isprintable() else " " for ch in data).strip()
|
||
if not clean:
|
||
return
|
||
self._paste_open = True
|
||
count = len(clean)
|
||
|
||
async def run() -> None:
|
||
try:
|
||
confirmed = await self._session.button_dialog(
|
||
Text(
|
||
f"Paste {count} character{'s' if count != 1 else ''} into your message?",
|
||
style="warn",
|
||
),
|
||
[("Cancel", False), ("Paste", True)],
|
||
title="Paste",
|
||
default=1,
|
||
border_style="warn",
|
||
footer_hint="←→ choose · Enter select · Esc cancel",
|
||
)
|
||
finally:
|
||
self._paste_open = False
|
||
if confirmed:
|
||
# Put it in the compose line: clear the selection, go back to the tail, and
|
||
# insert the run through the shared editor. If the result is over the limit,
|
||
# the byte gauge marks it, exactly as it marks typing past the limit.
|
||
self._clear_selection()
|
||
self._stick = True
|
||
self._editor.edit("text", clean)
|
||
self._status = ""
|
||
self._session.invalidate()
|
||
|
||
self._spawn(run())
|
||
|
||
def _move_selection(self, delta: int) -> None:
|
||
"""Move the reply selection by ``delta`` messages (negative = toward older).
|
||
|
||
The selection starts from the compose line only on a move up (``delta < 0``). A
|
||
move past the newest message clears the selection and stays at the tail again.
|
||
"""
|
||
if not self._messages:
|
||
return
|
||
last = len(self._messages) - 1
|
||
if self._selected is None:
|
||
if delta < 0:
|
||
self._selected = last
|
||
self._stick = False
|
||
return
|
||
target = self._selected + delta
|
||
if target > last:
|
||
self._clear_selection()
|
||
self._stick = True
|
||
else:
|
||
self._selected = max(0, target)
|
||
self._stick = False
|
||
|
||
def _select_section(self, direction: int) -> None:
|
||
"""Move the reply selection to the first message of the previous or next day.
|
||
|
||
This is the day jump of the transcript (Ctrl+PageUp/PageDown), for the selection.
|
||
Down moves to the first message of the next day (or to the tail when there is no
|
||
later day). Up moves to the first message of the current day, or to the first
|
||
message of the previous day when the selection is already on a first message.
|
||
"""
|
||
if not self._messages:
|
||
return
|
||
starts = self._day_start_indices()
|
||
current = self._selected if self._selected is not None else len(self._messages) - 1
|
||
if direction > 0:
|
||
target = next((s for s in starts if s > current), None)
|
||
if target is None:
|
||
self._clear_selection()
|
||
self._stick = True
|
||
return
|
||
else:
|
||
governing = max((s for s in starts if s <= current), default=0)
|
||
target = (
|
||
governing
|
||
if governing < current
|
||
else max((s for s in starts if s < current), default=0)
|
||
)
|
||
self._selected = target
|
||
self._stick = False
|
||
|
||
def _day_start_indices(self) -> list[int]:
|
||
"""The indexes of the messages that start a new local day: the transcript dividers."""
|
||
starts: list[int] = []
|
||
prev_day = None
|
||
for i, message in enumerate(self._messages):
|
||
day = message.created_at.astimezone().date()
|
||
if day != prev_day:
|
||
starts.append(i)
|
||
prev_day = day
|
||
return starts
|
||
|
||
def _clear_selection(self) -> None:
|
||
"""Clear the reply selection (the focus goes back to the compose line)."""
|
||
self._selected = None
|
||
self._selected_line = None
|
||
|
||
def _begin_reply(self) -> None:
|
||
"""Put an ``@mention`` of the sender of the selected message at the cursor.
|
||
|
||
The mention goes in at the cursor of the compose line, and the text that the user
|
||
already typed stays. Thus a message can name more than one node: select a message,
|
||
press Enter, select another, and press Enter again. A space comes before the
|
||
mention when it would touch the word in front of it, and after it when nothing
|
||
follows or a word would touch it. The mention is a chip from that moment, and the
|
||
cursor goes after it (refer to :class:`~meshterm.ui.compose.MentionEditor`).
|
||
|
||
A message of ours, or one with no sender, gives no name to mention. There the
|
||
line gets a bare ``@``, and the user types the name.
|
||
"""
|
||
message = self._messages[self._selected]
|
||
sender, _ = self._sender_and_body(message)
|
||
named = not message.outbound and sender != "·"
|
||
text, cursor = self._editor.text, self._editor.cursor
|
||
before = " " if cursor > 0 and not text[cursor - 1].isspace() else ""
|
||
if named:
|
||
after = " " if cursor == len(text) or not text[cursor].isspace() else ""
|
||
self._editor.insert(f"{before}@[{sender}]{after}")
|
||
else:
|
||
self._editor.insert(f"{before}@")
|
||
self._clear_selection()
|
||
self._stick = True
|
||
self._session.invalidate()
|
||
|
||
def _submit(self) -> None:
|
||
"""Send the current input line as a message (scheduled from the key handler).
|
||
|
||
A line that is too long for one packet goes out as parts, each with its part number
|
||
(refer to :mod:`meshterm.core.parts`), when the ``message_parts_max`` preference
|
||
permits more than one part. A line that is too long for all the parts is refused.
|
||
The buffer stays the same (so that the user can make it shorter), and the status
|
||
shows why, the same as the red overflow that the compose bar already shows.
|
||
"""
|
||
text = self._editor.text.strip()
|
||
if not text or self._sending:
|
||
return
|
||
cut = self._parts(text)
|
||
if cut is None:
|
||
if self._max_parts < 2:
|
||
over = self._used_bytes() - self._byte_limit()
|
||
self._status = f"Too long by {over} byte{'s' if over != 1 else ''} — trim to send."
|
||
else:
|
||
self._status = f"Too long for {self._max_parts} parts — trim to send."
|
||
self._session.invalidate()
|
||
return
|
||
self._editor = MentionEditor()
|
||
self._sending = True
|
||
self._stick = True
|
||
if self._is_channel:
|
||
self._status = "sending…"
|
||
self._session.invalidate()
|
||
self._spawn(self._send_channel(cut))
|
||
else:
|
||
self._status = ""
|
||
self._session.invalidate()
|
||
self._spawn(self._send_direct(cut))
|
||
|
||
def _sending_part(self, index: int, total: int) -> None:
|
||
"""Show which part goes out now, when a message has more than one."""
|
||
if total > 1:
|
||
self._status = f"sending part {index} of {total}…"
|
||
self._session.invalidate()
|
||
|
||
def _keep_unsent(self, cut: list[str], first: int) -> None:
|
||
"""Put the text of the parts that did not go out back in the compose line.
|
||
|
||
``first`` is the index in ``cut`` of the first part that did not go out. The text
|
||
goes back without its part numbers, so that Enter cuts it again. If the user typed
|
||
something new in the meantime, the line stays as it is, and the text is lost.
|
||
"""
|
||
if self._editor.text or first >= len(cut):
|
||
return
|
||
total = len(cut)
|
||
rest = " ".join(parts.unnumbered(cut[i], i + 1, total) for i in range(first, total))
|
||
self._editor = MentionEditor(rest)
|
||
|
||
async def _send_channel(self, cut: list[str]) -> None:
|
||
"""Broadcast each part of a channel message, and append each when the device takes it.
|
||
|
||
Between two parts, the send waits ``part_gap_s``, so that the floods do not go out
|
||
as a burst. If a part fails, the parts after it do not go out, and their text goes
|
||
back in the compose line.
|
||
"""
|
||
sent = 0
|
||
try:
|
||
for i, text in enumerate(cut):
|
||
if i:
|
||
await asyncio.sleep(self._part_gap_s)
|
||
self._sending_part(i + 1, len(cut))
|
||
message = await self._send(text)
|
||
sent += 1
|
||
if message is not None:
|
||
self._admit(message)
|
||
self._session.invalidate()
|
||
self._status = ""
|
||
except Exception as exc: # noqa: BLE001 - show the error inline, keep the chat open
|
||
self._status = f"send failed: {exc}"
|
||
self._keep_unsent(cut, sent)
|
||
finally:
|
||
self._sending = False
|
||
self._stick = True
|
||
self._session.invalidate()
|
||
|
||
async def _spin_while(self, coro: Awaitable[Any]) -> Any:
|
||
"""Await ``coro`` while the delivery spinner animates on the message in flight.
|
||
|
||
A background timer moves the shared spinner forward and paints every
|
||
:func:`~meshterm.ui.tui.spinner.spinner_interval` seconds (the spin rate of the
|
||
platform). Thus the glyph at the end of a pending message spins until the ack
|
||
resolves. Before the method returns, it always cancels the timer (and awaits it, so
|
||
that it cannot stay after the send as a stray pending task). The spinner is only
|
||
for the look. Thus the method ignores each problem in the animation, so that the
|
||
problem cannot break the send.
|
||
"""
|
||
self._spinner.reset()
|
||
|
||
async def animate() -> None:
|
||
while True:
|
||
await asyncio.sleep(spinner_interval())
|
||
self._spinner.tick()
|
||
self._session.invalidate()
|
||
|
||
ticker = self._spawn(animate())
|
||
try:
|
||
return await coro
|
||
finally:
|
||
ticker.cancel()
|
||
try:
|
||
await ticker
|
||
except asyncio.CancelledError:
|
||
pass
|
||
except Exception: # noqa: BLE001 - a spinner problem must never break a send
|
||
pass
|
||
|
||
async def _send_direct(self, cut: list[str]) -> None:
|
||
"""Send each part of a direct message, and show each as an optimistic bubble.
|
||
|
||
Each part shows as a pending bubble while its delivery is in flight. On success,
|
||
the stored message replaces the bubble. The stored message has its resolved
|
||
``acked`` state and its row id, so that the user can retry a ✗ later in place. A
|
||
part goes out when the part before it is acknowledged or its wait ends. A part with
|
||
no acknowledgement can still have arrived, so the parts after it go out too, and ^R
|
||
retries each ✗ in turn. A hard failure (the companion refuses the send) removes the
|
||
bubble, shows the error inline, and stops the parts after it. Their text goes back
|
||
in the compose line.
|
||
"""
|
||
sent = 0
|
||
try:
|
||
for i, text in enumerate(cut):
|
||
self._sending_part(i + 1, len(cut))
|
||
pending = ChatMessage(text=text, outbound=True, created_at=utcnow())
|
||
self._messages.append(pending)
|
||
self._session.invalidate()
|
||
try:
|
||
message = await self._spin_while(self._send(text))
|
||
except Exception:
|
||
self._discard(pending)
|
||
raise
|
||
if message is None: # nothing went out (the contact was not restored)
|
||
self._discard(pending)
|
||
self._keep_unsent(cut, sent)
|
||
break
|
||
sent += 1
|
||
self._swap(pending, message)
|
||
self._status = ""
|
||
except Exception as exc: # noqa: BLE001 - show the error inline, keep the chat open
|
||
self._status = f"send failed: {exc}"
|
||
self._keep_unsent(cut, sent)
|
||
finally:
|
||
self._sending = False
|
||
self._stick = True
|
||
self._session.invalidate()
|
||
|
||
def _retry_target(self) -> ChatMessage | None:
|
||
"""The message that ^R sends again, or ``None`` when there is nothing to retry.
|
||
|
||
In a direct chat, this is the newest outbound message that went out and was never
|
||
acknowledged. In a channel, nothing is ever acknowledged. There, it is a message
|
||
that *we* sent under a region, and ^R sends it again unscoped: the **selected**
|
||
message when a message is selected (the user pointed at it), or else the newest
|
||
message that we sent. With a selection, only the selection counts. If the selected
|
||
message is not a scoped message of ours, ^R offers nothing, and does not go past it
|
||
to another message without a sign. With no selection, only the newest counts.
|
||
When its unscoped copy has gone, that copy is the newest, and nothing remains to
|
||
offer. In the two cases, this is true only while a resend can start: no send is
|
||
already in flight, and a resend path is connected.
|
||
"""
|
||
if self._is_channel:
|
||
if self._sending or self._resend_unscoped is None:
|
||
return None
|
||
if self._selected is not None:
|
||
candidate: ChatMessage | None = self._messages[self._selected]
|
||
else:
|
||
candidate = next((m for m in reversed(self._messages) if m.outbound), None)
|
||
if (
|
||
candidate is None
|
||
or not candidate.outbound
|
||
or not candidate.scope
|
||
or candidate.scope == WILDCARD
|
||
):
|
||
return None
|
||
return candidate
|
||
if self._sending or self._resend is None:
|
||
return None
|
||
return next(
|
||
(m for m in reversed(self._messages) if m.outbound and m.acked is False),
|
||
None,
|
||
)
|
||
|
||
def _retry(self) -> None:
|
||
"""Try again to deliver the most recent unacknowledged direct message (Ctrl-R)."""
|
||
target = self._retry_target()
|
||
if target is None:
|
||
return
|
||
self._sending = True
|
||
target.acked = None # back to the spinner while the retry is in flight
|
||
self._status = "retrying…"
|
||
self._stick = True
|
||
self._session.invalidate()
|
||
self._spawn(self._resend_message(target))
|
||
|
||
def _begin_resend_unscoped(self) -> None:
|
||
"""Confirm, then send a scoped channel message again, unscoped (^R).
|
||
|
||
The dialog is amber (``danger``). Nothing is lost, but each repeater that permits
|
||
an unscoped flood relays it: all the mesh that the scope kept the message out of.
|
||
Thus it is a choice made on purpose, and the dialog names the region that the
|
||
message leaves. The resend is a new message on the air (a channel message has no
|
||
identity to deliver again), so it is appended as a new message. A radio that
|
||
cannot send unscoped refuses before anything goes out, and the status line says
|
||
why.
|
||
"""
|
||
target = self._retry_target()
|
||
if target is None or self._resend_open:
|
||
return
|
||
self._resend_open = True
|
||
preview = target.text if len(target.text) <= 32 else target.text[:31] + "…"
|
||
|
||
async def run() -> None:
|
||
try:
|
||
confirmed = await self._session.button_dialog(
|
||
Text(
|
||
f"Send “{preview}” again, unscoped? Every repeater that relays "
|
||
f"unscoped floods will carry it, not only those in {target.scope}.",
|
||
style="warn",
|
||
),
|
||
[("Cancel", False), ("Resend", True)],
|
||
title="Resend unscoped",
|
||
default=1,
|
||
border_style="warn",
|
||
footer_hint="←→ choose · Enter select · Esc cancel",
|
||
)
|
||
finally:
|
||
self._resend_open = False
|
||
if not confirmed or self._sending or self._resend_unscoped is None:
|
||
self._session.invalidate()
|
||
return
|
||
self._sending = True
|
||
self._status = "resending unscoped…"
|
||
# The resend goes to the tail as a new message. The selection that chose it is
|
||
# done.
|
||
self._clear_selection()
|
||
self._stick = True
|
||
self._session.invalidate()
|
||
try:
|
||
message = await self._resend_unscoped(target)
|
||
if message is not None:
|
||
self._admit(message)
|
||
self._status = ""
|
||
except Exception as exc: # noqa: BLE001 - show the error inline, keep the chat open
|
||
self._status = f"resend failed: {exc}"
|
||
finally:
|
||
self._sending = False
|
||
self._stick = True
|
||
self._session.invalidate()
|
||
|
||
self._spawn(run())
|
||
|
||
async def _resend_message(self, message: ChatMessage) -> None:
|
||
"""Run a retry to its end, and refresh the delivery state of the message in place."""
|
||
assert self._resend is not None
|
||
try:
|
||
await self._spin_while(self._resend(message))
|
||
except Exception as exc: # noqa: BLE001 - show the error inline, keep the chat open
|
||
message.acked = False
|
||
self._status = f"retry failed: {exc}"
|
||
else:
|
||
self._status = "" if message.acked else "still no ack — ^R to retry"
|
||
finally:
|
||
self._sending = False
|
||
self._stick = True
|
||
self._session.invalidate()
|
||
|
||
def _swap(self, old: ChatMessage, new: ChatMessage) -> None:
|
||
"""Replace an optimistic bubble with its stored message, in place.
|
||
|
||
The stored message is the one that a reaction can name, because it has the
|
||
timestamp that went out. Thus the fold takes it here, and not the bubble.
|
||
"""
|
||
if self._folds(new):
|
||
self._discard(old)
|
||
return
|
||
try:
|
||
self._messages[self._messages.index(old)] = new
|
||
except ValueError: # pragma: no cover - the bubble is always still there
|
||
self._messages.append(new)
|
||
|
||
def _folds(self, message: ChatMessage) -> bool:
|
||
"""Give ``message`` to the fold. ``True`` for a reaction, which the transcript omits."""
|
||
return self._fold is not None and self._fold.take(message)
|
||
|
||
def _admit(self, message: ChatMessage) -> None:
|
||
"""Append a message that we sent, unless it is a reaction (the fold shows that one)."""
|
||
if not self._folds(message):
|
||
self._messages.append(message)
|
||
|
||
# --- reactions -------------------------------------------------------------
|
||
|
||
def _react_target(self) -> tuple[ChatMessage, str | None, str, int] | None:
|
||
"""The selected message, and the author, body, and timestamp that a reaction names it by.
|
||
|
||
``None`` when nothing is selected, or when a reaction cannot name the selection: a
|
||
room, a message that is still in flight, or a channel message with no author (a
|
||
line with no ``Name: `` prefix, or one of ours when our name is not known).
|
||
"""
|
||
if self._fold is None or self._selected is None or not self._messages:
|
||
return None
|
||
message = self._messages[min(self._selected, len(self._messages) - 1)]
|
||
named = self._fold.target_of(message)
|
||
if named is None:
|
||
return None
|
||
author, body, timestamp = named
|
||
if self._is_channel and not author:
|
||
return None
|
||
return message, author, body, timestamp
|
||
|
||
def _can_react(self) -> bool:
|
||
"""Whether ^E can start a reaction: a selection that a reaction can name, and no send."""
|
||
return not self._sending and self._composing and self._react_target() is not None
|
||
|
||
def _begin_react(self) -> None:
|
||
"""Ask which emoji, then send the reaction to the selected message (^E).
|
||
|
||
The dialog is a question, so it floats as a box over the chat. The emoji of the
|
||
``reaction_emoji`` preference come first, then each other emoji (refer to
|
||
:func:`emoji_catalog`). Each row has the name of its emoji, so a typed word such as
|
||
``fire`` or ``heart`` filters the list to the emoji that it names. A row that our
|
||
node already sent to this message says so. A second copy of it would only count one
|
||
time, so the selection of that row sends nothing. The reaction goes out as a usual
|
||
message in the conversation. When it is stored, the fold puts it under its target
|
||
(refer to :meth:`_admit`).
|
||
"""
|
||
target = self._react_target()
|
||
if target is None or self._sending or self._react_open or self._fold is None:
|
||
return
|
||
message, author, body, timestamp = target
|
||
fold = self._fold
|
||
self._react_open = True
|
||
sender, _ = self._sender_and_body(message)
|
||
who = "your message" if message.outbound else sender
|
||
|
||
async def run() -> None:
|
||
from .menus import menu_rows, section_heading
|
||
|
||
def row(emoji: str) -> tuple[str, str, str]:
|
||
name = reactions.emoji_name(emoji)
|
||
sent = fold.reacted(message, emoji)
|
||
return emoticon(emoji), f"{name} · sent" if sent else name, emoji
|
||
|
||
mine = list(self._reaction_set)
|
||
others = [emoji for emoji in emoji_catalog() if emoji not in mine]
|
||
rows = menu_rows([row(emoji) for emoji in mine + others])
|
||
items = [
|
||
section_heading("Your reactions"),
|
||
*rows[: len(mine)],
|
||
section_heading("All emoji"),
|
||
*rows[len(mine) :],
|
||
]
|
||
try:
|
||
emoji = await self._session.select(
|
||
f"React — {who}", items, filterable=True, floating=True
|
||
)
|
||
finally:
|
||
self._react_open = False
|
||
if not emoji or self._sending:
|
||
self._session.invalidate()
|
||
return
|
||
if fold.reacted(message, emoji):
|
||
self._status = "you already reacted with that"
|
||
self._session.invalidate()
|
||
return
|
||
if self._is_channel:
|
||
text = reactions.channel_text(emoji, author or "", body, timestamp)
|
||
else:
|
||
text = reactions.direct_text(emoji, body, timestamp)
|
||
self._sending = True
|
||
self._status = "reacting…"
|
||
self._session.invalidate()
|
||
try:
|
||
sent_message = await self._send(text)
|
||
if sent_message is not None:
|
||
self._admit(sent_message)
|
||
no_ack = not self._is_channel and sent_message.acked is False
|
||
self._status = "reaction sent · no ack" if no_ack else ""
|
||
else:
|
||
self._status = ""
|
||
except Exception as exc: # noqa: BLE001 - show the error inline, keep the chat open
|
||
self._status = f"reaction failed: {exc}"
|
||
finally:
|
||
self._sending = False
|
||
self._session.invalidate()
|
||
|
||
self._spawn(run())
|
||
|
||
def _discard(self, message: ChatMessage) -> None:
|
||
"""Remove an optimistic bubble that never became a real message."""
|
||
try:
|
||
self._messages.remove(message)
|
||
except ValueError: # pragma: no cover - a defensive check
|
||
pass
|
||
|
||
|
||
#: The blocks of Unicode where an emoji with the emoji presentation can be. A wide symbol
|
||
#: between them is a CJK symbol or radical, not an emoji.
|
||
_EMOJI_BLOCKS = ((0x2000, 0x2E7F), (0x1F000, 0x1FAFF))
|
||
|
||
#: The I Ching symbols, which Unicode 16 draws wide. They are symbols, not emoji: the
|
||
#: firmware and the other apps never offer them.
|
||
_NOT_EMOJI = frozenset(range(0x268A, 0x2690)) | frozenset(range(0x2630, 0x2638))
|
||
|
||
|
||
@lru_cache(maxsize=1)
|
||
def emoji_catalog() -> tuple[str, ...]:
|
||
"""Each single emoji that the reaction dialog offers, in the order of Unicode.
|
||
|
||
The standard library has no list of emoji. Thus the catalog takes two sets. The first is
|
||
each symbol in the emoji blocks that Unicode draws wide by default (``🔥``, ``⌚``). The
|
||
second is the emoji that are text by default (``❤``), which the table of Rich lists.
|
||
Each of these gets the variation selector that asks for the emoji form (``❤️``), as the
|
||
apps send it. A flag, a skin tone, and a joined sequence are not in the catalog, but a
|
||
reaction with one still shows when it arrives. The names come from the Unicode database
|
||
of the Python that runs MeshTerm, so a newer Python offers newer emoji.
|
||
"""
|
||
from rich._unicode_data import load
|
||
|
||
found: dict[int, str] = {}
|
||
for first, last in _EMOJI_BLOCKS:
|
||
for cp in range(first, last + 1):
|
||
char = chr(cp)
|
||
if cp in _NOT_EMOJI:
|
||
continue
|
||
if unicodedata.category(char) == "So" and unicodedata.east_asian_width(char) == "W":
|
||
found[cp] = char
|
||
for char in load("auto").narrow_to_wide:
|
||
if not char.isascii() and len(char) == 1:
|
||
found.setdefault(ord(char), char + "\ufe0f")
|
||
return tuple(
|
||
emoji
|
||
for _, emoji in sorted(found.items())
|
||
if not reactions._is_regional(emoji[0]) and reactions.is_reaction_emoji(emoji)
|
||
)
|
||
|
||
|
||
async def our_node(ctx: AppContext) -> tuple[str | None, str | None]:
|
||
"""The name and the public key of our node, from the cache of the session.
|
||
|
||
A chat shows them on our messages (refer to :func:`~meshterm.ui.widgets.self_chip`),
|
||
and a direct chat names one end of each of its packets with the key. The read uses the
|
||
cache, so it does not ask the device again.
|
||
|
||
Returns:
|
||
``(name, key)``. An item is ``None`` when the device did not report it, or when the
|
||
read failed. Our node with no name only keeps the bare ``★`` in a chat.
|
||
"""
|
||
try:
|
||
info = await ctx.devstate.self_info()
|
||
except Exception: # noqa: BLE001 - our node with no name keeps the bare star
|
||
return None, None
|
||
return str(info.get("name") or "") or None, str(info.get("public_key") or "") or None
|
||
|
||
|
||
async def open_chat(ctx: AppContext, conversation: Conversation) -> int:
|
||
"""Open the live chat screen for ``conversation`` and run it until the user closes it.
|
||
|
||
This function makes sure that the listener and the message recorder run, reads the
|
||
stored transcript, subscribes to the hub so that inbound messages for this thread are
|
||
appended live, and pushes the screen. At exit, it always removes the subscription and
|
||
the marker of the active conversation.
|
||
|
||
Args:
|
||
ctx: The shared application context (it must run the interactive TUI surface).
|
||
conversation: The channel or contact conversation to open.
|
||
|
||
Returns:
|
||
The number of messages in the transcript when the screen closed.
|
||
|
||
Raises:
|
||
RuntimeError: If the call is not from the interactive menu (no full-screen
|
||
session).
|
||
"""
|
||
from .surface import TuiUi
|
||
|
||
if not isinstance(ctx.ui, TuiUi): # pragma: no cover - the caller is only in the menu
|
||
raise RuntimeError("live chat is only available in the interactive menu")
|
||
if conversation.is_room:
|
||
from .room import open_room
|
||
|
||
return await open_room(ctx, conversation)
|
||
session = ctx.ui.session
|
||
|
||
device = await ctx.device()
|
||
try:
|
||
await ctx.chat.start() # start to store inbound messages, if it did not already
|
||
except Exception: # noqa: BLE001 - the hub may already run. The recorder is best-effort
|
||
pass
|
||
|
||
from ..services import trace_runner
|
||
|
||
history = ctx.repo.recent_chat_messages(
|
||
is_channel=conversation.is_channel,
|
||
channel_id=conversation.channel_id,
|
||
peer=conversation.peer,
|
||
limit=ctx.preferences.chat_history_limit,
|
||
)
|
||
contacts = await ctx.devstate.contacts()
|
||
names = _contact_names(contacts)
|
||
stored_names, named_at = ctx.repo.node_names_with_times()
|
||
key_of = trace_runner.make_name_key_resolver(contacts, stored_names, heard=named_at)
|
||
|
||
async def send(text: str) -> ChatMessage | None:
|
||
if conversation.is_channel:
|
||
assert conversation.channel_idx is not None
|
||
return await ctx.chat.send_channel(
|
||
conversation.channel_idx, text, label=conversation.label
|
||
)
|
||
assert conversation.contact is not None
|
||
return await _with_restore(ctx, lambda: ctx.chat.send_direct(conversation.contact, text))
|
||
|
||
resend: Callable[[ChatMessage], Awaitable[ChatMessage]] | None = None
|
||
resend_unscoped: Callable[[ChatMessage], Awaitable[ChatMessage | None]] | None = None
|
||
scope: str | None = None
|
||
if conversation.is_channel:
|
||
scope = ctx.chat.channel_scope(conversation.channel_id)
|
||
|
||
async def resend_unscoped(message: ChatMessage) -> ChatMessage | None:
|
||
assert conversation.channel_idx is not None
|
||
return await ctx.chat.send_channel(
|
||
conversation.channel_idx, message.text, label=conversation.label, scope=WILDCARD
|
||
)
|
||
|
||
else:
|
||
|
||
async def resend(message: ChatMessage) -> ChatMessage:
|
||
assert conversation.contact is not None
|
||
return await _with_restore(
|
||
ctx, lambda: ctx.chat.resend_direct(conversation.contact, message)
|
||
)
|
||
|
||
from .map_screen import Locator
|
||
from .node_detail_screen import ContactPages
|
||
|
||
paths = await _make_paths_presenter(ctx, conversation, device)
|
||
self_name, self_key = await our_node(ctx)
|
||
screen = ChatScreen(
|
||
conversation,
|
||
history,
|
||
send=send,
|
||
names=names,
|
||
session=session,
|
||
resend=resend,
|
||
paths=paths,
|
||
key_of=key_of,
|
||
scope=scope,
|
||
resend_unscoped=resend_unscoped,
|
||
locator=await Locator.gather(ctx),
|
||
pages=await ContactPages.gather(ctx, self_key=self_key or ""),
|
||
self_name=self_name,
|
||
self_key=self_key,
|
||
reaction_set=reactions.reaction_set(ctx.preferences.reaction_emoji),
|
||
max_parts=ctx.preferences.message_parts_max,
|
||
part_gap_s=ctx.preferences.trace_cooldown_s,
|
||
)
|
||
ctx.chat.set_active(conversation.key)
|
||
|
||
def on_event(event: MeshEvent) -> None:
|
||
message = event.message
|
||
if message is not None and _belongs(message, conversation):
|
||
peer_name = names.get((message.sender or "").lower())
|
||
screen.append(
|
||
ChatMessage.from_message(
|
||
message, peer_name=peer_name, channel_id=conversation.channel_id
|
||
)
|
||
)
|
||
|
||
unsubscribe = ctx.events.subscribe(on_event, EventKind.MESSAGE)
|
||
try:
|
||
await session.run_screen(screen)
|
||
finally:
|
||
unsubscribe()
|
||
ctx.chat.set_active(None)
|
||
return len(screen._messages)
|
||
|
||
|
||
async def _with_restore(ctx: AppContext, attempt: Callable[[], Awaitable[Any]]) -> Any:
|
||
"""Run a direct send. If the device forgot the contact, offer to restore it, then retry.
|
||
|
||
This is the one rejection from which a send can recover: the firmware has no contact
|
||
for the recipient (refer to :class:`~meshterm.core.connection.ContactNotOnDeviceError`),
|
||
and one write corrects it. The user is asked first (:func:`_restore_contact`). If the
|
||
user declines, the error is raised again. Thus the chat shows the refusal exactly as it
|
||
shows each other failed send.
|
||
|
||
Args:
|
||
ctx: The shared application context.
|
||
attempt: The send to run. It is called a second time, with no change, when the
|
||
contact is back.
|
||
|
||
Returns:
|
||
The return value of ``attempt``.
|
||
|
||
Raises:
|
||
Exception: Each error that ``attempt`` raises. A rejection for a forgotten contact
|
||
is raised again only when the user declined the offer to add it back.
|
||
"""
|
||
try:
|
||
return await attempt()
|
||
except ContactNotOnDeviceError as missing:
|
||
if not await _restore_contact(ctx, missing):
|
||
raise
|
||
return await attempt()
|
||
|
||
|
||
async def _restore_contact(ctx: AppContext, missing: ContactNotOnDeviceError) -> bool:
|
||
"""Offer to write a forgotten contact back to the device. ``True`` if it now has it.
|
||
|
||
A direct message is addressed through the contact entry of the *device*. Thus MeshTerm
|
||
cannot send a message to a contact that the firmware removed, also when MeshTerm still
|
||
lists it. (The list on a screen is the union of the table of the device and the
|
||
contacts that MeshTerm keeps for it: refer to :mod:`meshterm.core.contact_store`.) The
|
||
contact that we already have holds all the data that the entry must have. Thus one
|
||
write is the repair, and the send that got the rejection tries again after it.
|
||
|
||
The function offers the write and does not do it silently. The write changes what the
|
||
device stores, and a full contact table refuses it (the user must see that, and it must
|
||
not be hidden).
|
||
|
||
Args:
|
||
ctx: The shared application context.
|
||
missing: The rejection, with the contact that the device could not find.
|
||
|
||
Returns:
|
||
``True`` when the contact is on the device (retry the send). ``False`` if the user
|
||
declined. Then the caller raises the error again, so that the chat shows the
|
||
refusal.
|
||
|
||
Raises:
|
||
DeviceCommandError: If the device refused the write (most often, a full table).
|
||
"""
|
||
contact = missing.contact
|
||
add = await ctx.ui.dialog(
|
||
f"{contact.name} isn't in this device's contacts, so the radio can't address it. "
|
||
"Add it back and send?",
|
||
[("Cancel", False), ("Add & send", True)],
|
||
title="Contact not on device",
|
||
default=1,
|
||
)
|
||
if not add:
|
||
return False
|
||
device = await ctx.device()
|
||
await device.add_contact(contact)
|
||
# The table of the device changed under the session cache. The next read gets it again
|
||
# (with the route that the firmware learns for the contact from now on).
|
||
ctx.devstate.invalidate_contacts()
|
||
return True
|
||
|
||
|
||
def _contact_names(contacts: list[Contact]) -> dict[str, str]:
|
||
"""Build a map from key prefix to name, for the labels of inbound direct messages."""
|
||
names: dict[str, str] = {}
|
||
for contact in contacts:
|
||
for key in (contact.key_prefix, contact.public_key[:12]):
|
||
if key:
|
||
names[key.lower()] = contact.name
|
||
return names
|
||
|
||
|
||
def _belongs(message: Message, conversation: Conversation) -> bool:
|
||
"""Whether an inbound message belongs to the open conversation.
|
||
|
||
Args:
|
||
message: The received message.
|
||
conversation: The conversation that the screen shows now.
|
||
|
||
Returns:
|
||
``True`` if the message must be appended to this transcript.
|
||
"""
|
||
if conversation.is_channel:
|
||
return message.is_channel and message.channel == conversation.channel_idx
|
||
if message.is_channel:
|
||
return False
|
||
sender = (message.sender or "").lower()
|
||
peer = (conversation.peer or "").lower()
|
||
if not sender or not peer:
|
||
return False
|
||
return sender.startswith(peer) or peer.startswith(sender)
|
||
|
||
|
||
# --- message paths (the ^P view) -----------------------------------------------------
|
||
|
||
|
||
def _sent_scope_line(ctx: AppContext, message: ChatMessage, *, relayed: bool) -> Text | None:
|
||
"""The scope of a message that we sent, for the paths dialog, and why it can be lost.
|
||
|
||
The line comes from the scope stored on the message at send time
|
||
(:attr:`ChatMessage.scope`), never from the current scope of the channel, which may
|
||
have changed since then. A scoped message can have no stored relayed copy *and* no
|
||
repeater that was ever heard here to relay its region
|
||
(:meth:`~meshterm.core.region_store.RegionStore.carriers`). Then the line gives the
|
||
most probable reason in plain words: only repeaters that relay its region relay a
|
||
scoped flood, and no known repeater in range does.
|
||
|
||
Args:
|
||
ctx: The shared application context (for the region store).
|
||
message: The message whose paths are open.
|
||
relayed: Whether a copy of it was heard when it came back.
|
||
|
||
Returns:
|
||
The line, or ``None`` for a message with no stored scope (inbound, direct, or
|
||
sent before MeshTerm stored scopes).
|
||
"""
|
||
from ..core.regions import UNSCOPED, Scope
|
||
from .widgets import scope_text
|
||
|
||
if not message.outbound or not message.scope:
|
||
return None
|
||
line = Text("sent ", style="muted")
|
||
if message.scope == WILDCARD:
|
||
line.append_text(scope_text(UNSCOPED))
|
||
return line
|
||
line.append("under ", style="muted")
|
||
line.append_text(scope_text(Scope("scoped", message.scope)))
|
||
store = getattr(ctx, "region_store", None)
|
||
carriers = len(store.carriers(message.scope)) if store is not None else 0
|
||
if not relayed and not carriers:
|
||
line.append(" — no repeater known here carries it", style="warn")
|
||
elif carriers:
|
||
line.append(
|
||
f" · {carriers} known repeater{'s' if carriers != 1 else ''} carr"
|
||
f"{'y' if carriers != 1 else 'ies'} it",
|
||
style="muted",
|
||
)
|
||
return line
|
||
|
||
|
||
async def _make_paths_presenter(
|
||
ctx: AppContext,
|
||
conversation: Conversation,
|
||
device, # noqa: ANN001 - the core Device. Its type is set at the source
|
||
) -> Callable[[ChatMessage], Awaitable[None]]:
|
||
"""Build the async presenter for the ^P delivery-paths view of the chat.
|
||
|
||
It collects what the presenter must have, one time each time the chat opens:
|
||
|
||
* a resolver of hop names over the contacts and all the names that the recorder ever
|
||
heard (the rule of the app is that a node that has a name never shows as a bare
|
||
hash),
|
||
* the name of our node, for the white ``you``,
|
||
* the width of the routing prefix, for the lit hashes, and
|
||
* for a channel, its secret. The function reads it from the device when the
|
||
conversation did not have one (a conversation from the picker knows its identity,
|
||
but not always its key).
|
||
|
||
Args:
|
||
ctx: The shared application context.
|
||
conversation: The conversation that the chat screen opens.
|
||
device: The connected device (the caller already awaited it).
|
||
|
||
Returns:
|
||
An async callable that shows the paths of one message in a floating window.
|
||
"""
|
||
from ..services import trace_runner
|
||
from ..services.message_paths import (
|
||
channel_arrivals,
|
||
collapse,
|
||
direct_arrivals,
|
||
distinct_paths,
|
||
message_scope,
|
||
)
|
||
from .message_paths_screen import MessagePathsScreen
|
||
|
||
session = ctx.ui.session
|
||
contacts = await ctx.devstate.contacts()
|
||
stored_names, named_at = ctx.repo.node_names_with_times()
|
||
resolve = trace_runner.make_node_resolver(contacts, stored_names)
|
||
type_of = trace_runner.make_node_type_resolver(contacts)
|
||
key_of = trace_runner.make_name_key_resolver(contacts, stored_names, heard=named_at)
|
||
prefix_bytes = await ctx.devstate.routing_prefix_bytes()
|
||
# Our key names one end of each direct packet in this conversation. The key of the
|
||
# contact names the other end (refer to :func:`direct_arrivals`). Our node with no name
|
||
# only gets no white highlight.
|
||
self_name, self_key = await our_node(ctx)
|
||
|
||
secret = conversation.secret
|
||
if conversation.is_channel and not secret and conversation.channel_idx is not None:
|
||
try:
|
||
payload = await device.get_channel(conversation.channel_idx)
|
||
raw = (payload or {}).get("channel_secret")
|
||
secret = bytes(raw) if raw else None
|
||
except Exception: # noqa: BLE001 - no key, no decrypt. The view says so correctly
|
||
secret = None
|
||
|
||
async def present(message: ChatMessage) -> None:
|
||
if conversation.is_channel and secret:
|
||
arrivals = channel_arrivals(
|
||
ctx.repo, message, channel_name=conversation.label, secret=secret
|
||
)
|
||
matched = True
|
||
summary = (
|
||
f"heard {len(arrivals)} time{'s' if len(arrivals) != 1 else ''}"
|
||
f" · {distinct_paths(arrivals)} distinct "
|
||
f"path{'s' if distinct_paths(arrivals) != 1 else ''}"
|
||
if arrivals
|
||
else "no copies in the packet log"
|
||
)
|
||
elif conversation.is_channel:
|
||
await session.scroll(
|
||
Text(
|
||
"This channel's key isn't at hand, so overheard frames can't be "
|
||
"matched to the message.",
|
||
style="muted",
|
||
),
|
||
title="Message paths",
|
||
)
|
||
return
|
||
else:
|
||
contact = conversation.contact
|
||
peer_key = conversation.peer or (
|
||
(contact.public_key or contact.key_prefix) if contact else None
|
||
)
|
||
arrivals, exact = direct_arrivals(
|
||
ctx.repo, message, self_key=self_key, peer_key=peer_key
|
||
)
|
||
# An exact match says the same as the channel view: these packets are this
|
||
# message. The shared MAC of the packets gives the match. Thus it is correct,
|
||
# also when MeshTerm cannot read one word of them.
|
||
matched = exact
|
||
arrivals = collapse(arrivals)
|
||
heard = sum(a.copies for a in arrivals)
|
||
if exact:
|
||
summary = (
|
||
f"heard {heard} time{'s' if heard != 1 else ''}"
|
||
f" · {len(arrivals)} distinct path{'s' if len(arrivals) != 1 else ''}"
|
||
if arrivals
|
||
else "no copies in the packet log"
|
||
)
|
||
else:
|
||
summary = (
|
||
"direct frames are encrypted — matched by address and time"
|
||
if peer_key and self_key
|
||
else "direct frames are encrypted — matched by time alone"
|
||
)
|
||
# The two ends of the path: the node that the message started from, and the node
|
||
# that it was addressed to. Our sends go from our node to the peer. An inbound
|
||
# channel message names its sender on the air and is addressed to all nodes, so it
|
||
# ends on our node. The origin of an inbound message in a direct chat is the peer of
|
||
# the conversation.
|
||
destination: str | None = self_name
|
||
if message.outbound:
|
||
source = self_name
|
||
# A channel broadcast is addressed to no specific node, and the copies that we
|
||
# log are its rebroadcasts that come back. Thus it really ends on our node. A
|
||
# direct send does not: it went to one node, and its route ends there.
|
||
if not conversation.is_channel:
|
||
destination = conversation.label
|
||
elif conversation.is_channel:
|
||
source, _body = split_channel_sender(message.text)
|
||
else:
|
||
source = conversation.label
|
||
await session.run_screen(
|
||
MessagePathsScreen(
|
||
message,
|
||
arrivals,
|
||
matched=matched,
|
||
resolve=resolve,
|
||
prefix_bytes=prefix_bytes,
|
||
self_name=self_name,
|
||
summary=summary,
|
||
source=source or None,
|
||
destination=destination or None,
|
||
type_of=type_of,
|
||
key_of=key_of,
|
||
scope=message_scope(arrivals, ctx.region_store.scope_of),
|
||
sent_scope=_sent_scope_line(ctx, message, relayed=bool(arrivals)),
|
||
)
|
||
)
|
||
|
||
return present
|