457 lines
14 KiB
Python
457 lines
14 KiB
Python
import asyncio
|
|
from typing import Any, Awaitable, Callable, List, Optional, Tuple, TypeVar
|
|
|
|
import urwid
|
|
import yaboli
|
|
|
|
from ..attributed_text_widget import ATWidget
|
|
from ..cursor_rendering import CursorTreeRenderer
|
|
from ..cursor_tree_widget import CursorTreeWidget
|
|
from ..element import Message, RenderedMessage
|
|
from ..element_supply import InMemorySupply
|
|
from ..markup import AT, AttributedText, Attributes
|
|
from .edit_widgets import EditWidget
|
|
from .euph_renderer import EuphRenderer
|
|
|
|
__all__ = ["RoomWidget"]
|
|
|
|
# I don't know of a way to type the arguments correctly with mypy. You can't
|
|
# just substitute a Callable's parameter list with a type variable (sadly), and
|
|
# other workarounds didn't seem to solve this exact problem.
|
|
def synchronous(f: Any) -> Any:
|
|
def wrapper(*args: Any, **kwargs: Any) -> None:
|
|
asyncio.create_task(f(*args, **kwargs))
|
|
return wrapper
|
|
|
|
class RoomLayout(urwid.WidgetWrap):
|
|
def __init__(self,
|
|
room_name: Any,
|
|
nick_list: Any,
|
|
tree: Any,
|
|
edit: Any,
|
|
nick_list_width: int = 24,
|
|
border_attrs: Attributes = {},
|
|
room_name_separator: str = "═",
|
|
room_name_split: str = "╤",
|
|
nick_list_separator: str = "│",
|
|
nick_list_split: str = "┤",
|
|
edit_separator: str = "─",
|
|
) -> None:
|
|
|
|
self._width = 0
|
|
self._height = 0
|
|
self._redraw = True
|
|
|
|
self._room_name = room_name
|
|
self._nick_list = nick_list
|
|
self._tree = tree
|
|
self._edit = edit
|
|
self._nick_list_width = nick_list_width
|
|
self._border_attrs = border_attrs
|
|
self._room_name_separator = room_name_separator
|
|
self._room_name_split = room_name_split
|
|
self._nick_list_separator = nick_list_separator
|
|
self._nick_list_split = nick_list_split
|
|
self._edit_separator = edit_separator
|
|
|
|
# Placeholders (TODO: Use urwid.Text)
|
|
self._room_name_divider = ATWidget(AT())
|
|
self._nick_list_divider = ATWidget(AT())
|
|
self._edit_divider = ATWidget(AT())
|
|
|
|
self._edit_pile = urwid.Pile([
|
|
self._tree,
|
|
("pack", self._edit_divider),
|
|
("pack", self._edit),
|
|
])
|
|
self._edit_pile.focus_position = 2
|
|
|
|
self._left_wrap = urwid.WidgetWrap(self._tree)
|
|
|
|
self._columns = urwid.Columns([
|
|
self._left_wrap,
|
|
(1, urwid.Filler(self._nick_list_divider)),
|
|
(self._nick_list_width, self._nick_list),
|
|
])
|
|
self._columns.focus_position = 0
|
|
|
|
self._outer_pile = urwid.Pile([
|
|
("pack", self._room_name),
|
|
("pack", self._room_name_divider),
|
|
self._columns,
|
|
])
|
|
self._outer_pile.focus_position = 2
|
|
|
|
super().__init__(self._outer_pile)
|
|
|
|
def render(self, size: Tuple[int, int], focus: bool) -> Any:
|
|
self._width, self._height = size
|
|
|
|
tree_width = self._width - self._nick_list_width - 1
|
|
|
|
self._room_name_divider.set_attributed_text(
|
|
self._render_room_name_divider(tree_width))
|
|
self._nick_list_divider.set_attributed_text(
|
|
self._render_nick_list_divider(tree_width))
|
|
self._edit_divider.set_attributed_text(
|
|
self._render_edit_divider(tree_width))
|
|
|
|
return super().render(size, focus)
|
|
|
|
def _render_room_name_divider(self, tree_width: int) -> AttributedText:
|
|
string = (
|
|
self._room_name_separator * tree_width +
|
|
self._room_name_split +
|
|
self._room_name_separator * self._nick_list_width
|
|
)
|
|
return AT(string, attributes=self._border_attrs)
|
|
|
|
def _render_nick_list_divider(self, tree_width: int) -> AttributedText:
|
|
height = self._height - self._room_name.rows((self._width,)) - 1
|
|
|
|
if self._left_wrap._w is self._tree:
|
|
lines = [self._nick_list_separator] * height
|
|
else:
|
|
edit_height = self._edit.rows((tree_width,))
|
|
tree_height = height - edit_height - 1
|
|
lines = (
|
|
[self._nick_list_separator] * tree_height +
|
|
[self._nick_list_split] +
|
|
[self._nick_list_separator] * edit_height
|
|
)
|
|
|
|
string = "\n".join(lines)
|
|
return AT(string, attributes=self._border_attrs)
|
|
|
|
def _render_edit_divider(self, tree_width: int) -> AttributedText:
|
|
string = self._edit_separator * tree_width
|
|
return AT(string, attributes=self._border_attrs)
|
|
|
|
def set_edit_visible(self, visible: bool):
|
|
if visible:
|
|
self._left_wrap._w = self._edit_pile
|
|
else:
|
|
self._left_wrap._w = self._tree
|
|
|
|
def focus_on_edit(self):
|
|
self._edit_pile.focus_position = 2
|
|
self._columns.focus_position = 0
|
|
|
|
def focus_on_tree(self):
|
|
self._edit_pile.focus_position = 0
|
|
self._columns.focus_position = 0
|
|
|
|
def focus_on_user_list(self):
|
|
self._columns.focus_position = 2
|
|
|
|
class RoomWidget(urwid.WidgetWrap):
|
|
"""
|
|
The RoomWidget connects to and displays a single yaboli room.
|
|
|
|
Its life cycle looks like this:
|
|
1. Create widget
|
|
2. Call connect() (while the event loop is running)
|
|
3. Keep widget around and occasionally display it
|
|
4. Call disconnect() (while the event loop is runnning)
|
|
5. When the room should be destroyed/forgotten about, it sends a "close"
|
|
event
|
|
"""
|
|
|
|
CONNECTING = "connecting"
|
|
CONNECTION_FAILED = "connection_failed"
|
|
VIEWING = "viewing"
|
|
EDITING = "editing"
|
|
|
|
def __init__(self,
|
|
roomname: str,
|
|
log_amount: int = 200,
|
|
) -> None:
|
|
|
|
if log_amount < 1:
|
|
raise ValueError() # TODO add better text
|
|
self._log_amount = log_amount
|
|
|
|
self._mode: str
|
|
self._requesting_logs = False
|
|
|
|
self._room = yaboli.Room(roomname)
|
|
self._room.register_event("snapshot", self.on_snapshot)
|
|
self._room.register_event("send", self.on_send)
|
|
|
|
self._supply = InMemorySupply[Message]()
|
|
self._renderer = self._create_euph_renderer()
|
|
self._tree = CursorTreeRenderer[Message](self._supply, self._renderer)
|
|
|
|
# All of the widgets
|
|
|
|
self._connecting = self._create_connecting_widget()
|
|
self._connection_failed = self._create_connection_failed_widget()
|
|
|
|
self._edit_nick = self._create_edit_nick_widget()
|
|
#self._edit_password = self._create_password_edit_widget()
|
|
#self._authenticating = self._create_authenticating_widget()
|
|
|
|
self._room_name = self._create_room_name_widget()
|
|
self._nick_list = self._create_nick_list_widget()
|
|
self._tree_widget = self._create_tree_widget()
|
|
self._edit = self._create_edit_widget()
|
|
self._layout = self._create_room_layout_widget(self._room_name,
|
|
self._nick_list, self._tree_widget, self._edit)
|
|
|
|
self._box = urwid.LineBox(self._edit_nick)
|
|
self._overlay = urwid.Overlay(
|
|
self._box,
|
|
self._layout,
|
|
align=urwid.CENTER,
|
|
width=(urwid.RELATIVE, 24),
|
|
valign=urwid.MIDDLE,
|
|
height=urwid.PACK,
|
|
)
|
|
|
|
super().__init__(self._connecting)
|
|
self.switch_connecting()
|
|
|
|
def _create_euph_renderer(self) -> EuphRenderer:
|
|
return EuphRenderer("")
|
|
|
|
def _create_connecting_widget(self) -> Any:
|
|
text = (
|
|
AT("Connecting to ") +
|
|
AT("&" + self._room.name) +
|
|
AT("...")
|
|
)
|
|
# Centered vertically and horizontally
|
|
return urwid.Filler(ATWidget(text, align=urwid.CENTER))
|
|
|
|
def _create_connection_failed_widget(self) -> Any:
|
|
text = (
|
|
AT("Could not connect to ") +
|
|
AT("&" + self._room.name) +
|
|
AT("...")
|
|
)
|
|
# Centered vertically and horizontally
|
|
return urwid.Filler(ATWidget(text, align=urwid.CENTER))
|
|
|
|
def _create_room_name_widget(self) -> Any:
|
|
return urwid.Text("&" + self._room.name, align=urwid.CENTER)
|
|
|
|
def _create_tree_widget(self) -> Any:
|
|
return CursorTreeWidget(self._tree)
|
|
|
|
def _create_edit_widget(self) -> Any:
|
|
return urwid.Edit(multiline=True)
|
|
|
|
def _create_nick_list_widget(self) -> Any:
|
|
return urwid.SolidFill("n")
|
|
|
|
def _create_room_layout_widget(self,
|
|
room_name: Any,
|
|
nick_list: Any,
|
|
tree: Any,
|
|
edit: Any,
|
|
) -> Any:
|
|
return RoomLayout(room_name, nick_list, tree, edit)
|
|
|
|
def _create_edit_nick_widget(self) -> Any:
|
|
return EditWidget("Choose a nick: ", "@")
|
|
|
|
# Start up the connection and room
|
|
|
|
@synchronous
|
|
async def connect(self) -> None:
|
|
success = await self._room.connect()
|
|
if success:
|
|
self.switch_view()
|
|
else:
|
|
self.switch_connection_failed()
|
|
urwid.emit_signal(self, "close")
|
|
|
|
@synchronous
|
|
async def disconnect(self) -> None:
|
|
self._room.disconnect()
|
|
|
|
# NEW, BETTER ORGANIZED ROOM WIDGET
|
|
|
|
# UI mode and mode switching
|
|
|
|
CONNECTING = "connecting"
|
|
CONNECTION_FAILED = "connection_failed"
|
|
SETTING_PASSWORD = "setting_password"
|
|
AUTHENTICATING = "authenticating"
|
|
SETTING_NICK = "setting_nick"
|
|
VIEWING = "viewing"
|
|
EDITING = "editing"
|
|
|
|
def switch_connecting(self) -> None:
|
|
self._w = self._connecting
|
|
self._mode = self.CONNECTING
|
|
|
|
def switch_connection_failed(self) -> None:
|
|
self._w = self._connection_failed
|
|
self._mode = self.CONNECTION_FAILED
|
|
|
|
def switch_setting_password(self) -> None:
|
|
self._w = self._overlay
|
|
self._overlay.set_top(self._edit_password)
|
|
self._mode = self.SETTING_PASSWORD
|
|
|
|
def switch_authenticating(self) -> None:
|
|
self._w = self._overlay
|
|
self._overlay.set_top(self._authenticating)
|
|
self._mode = self.AUTHENTICATING
|
|
|
|
def switch_setting_nick(self) -> None:
|
|
self._w = self._overlay
|
|
self._box.original_widget = self._edit_nick
|
|
self._edit_nick.text = self._room.session.nick
|
|
self.update_edit_nick()
|
|
self._mode = self.SETTING_NICK
|
|
|
|
def switch_view(self) -> None:
|
|
self._w = self._layout
|
|
self._layout.set_edit_visible(False)
|
|
self._layout.focus_on_tree()
|
|
self._mode = self.VIEWING
|
|
|
|
def switch_edit(self) -> None:
|
|
self._w = self._layout
|
|
self._layout.set_edit_visible(True)
|
|
self._layout.focus_on_edit()
|
|
self._mode = self.EDITING
|
|
|
|
# Updating various parts of the UI
|
|
|
|
def update_tree(self) -> None:
|
|
self._tree_widget._invalidate()
|
|
|
|
def update_nick_list(self) -> None:
|
|
# Ensure that self._room.session and self._room.users exist
|
|
if self._mode not in {self.SETTING_NICK, self.VIEWING, self.EDITING}:
|
|
return
|
|
|
|
#self._nick_list.update(self._room.session, self._room.users)
|
|
|
|
def update_edit_nick(self):
|
|
width = self._edit_nick.width
|
|
self._overlay.set_overlay_parameters(
|
|
align=urwid.CENTER,
|
|
width=width + 2, # for the LineBox
|
|
valign=urwid.MIDDLE,
|
|
height=urwid.PACK,
|
|
)
|
|
self._overlay._invalidate()
|
|
|
|
# Reacting to changes
|
|
|
|
def own_nick_change(self):
|
|
self._renderer.nick = self._room.session.nick
|
|
self._tree.invalidate_all()
|
|
self.update_tree()
|
|
self.update_nick_list()
|
|
|
|
def receive_message(self, msg: yaboli.Message):
|
|
self._supply.add(Message(
|
|
msg.message_id,
|
|
msg.parent_id,
|
|
msg.time,
|
|
msg.sender.nick,
|
|
msg.content,
|
|
))
|
|
self._tree.invalidate(msg.message_id)
|
|
self.update_tree()
|
|
|
|
# Reacting to urwid stuff
|
|
|
|
def render(self, size: Tuple[int, int], focus: bool) -> None:
|
|
canvas = super().render(size, focus)
|
|
|
|
if self._tree.hit_top and not self._requesting_logs:
|
|
self._requesting_logs = True
|
|
self.request_logs()
|
|
|
|
return canvas
|
|
|
|
def keypress(self, size: Tuple[int, int], key: str) -> Optional[str]:
|
|
if self._mode == self.VIEWING:
|
|
if key in {"enter", "meta enter"} and not self._room.session.nick:
|
|
self.switch_setting_nick()
|
|
elif key == "enter":
|
|
self._edit.edit_text = ""
|
|
self.switch_edit()
|
|
elif key == "meta enter":
|
|
self.switch_edit()
|
|
elif key == "n":
|
|
self.switch_setting_nick()
|
|
elif key == "r":
|
|
self._tree.invalidate_all()
|
|
self._tree_widget._invalidate()
|
|
else:
|
|
return super().keypress(size, key)
|
|
|
|
elif self._mode == self.EDITING:
|
|
if key == "enter":
|
|
if self._edit.edit_text:
|
|
self.send(self._edit.edit_text, self._tree.cursor_id)
|
|
self.switch_view()
|
|
elif key == "esc":
|
|
self.switch_view()
|
|
elif key == "meta enter":
|
|
self._edit.insert_text("\n")
|
|
else:
|
|
return super().keypress(size, key)
|
|
|
|
elif self._mode == self.SETTING_NICK:
|
|
if key == "enter":
|
|
if self._edit_nick.text:
|
|
self.nick(self._edit_nick.text)
|
|
self.switch_view()
|
|
elif key == "esc":
|
|
self.switch_view()
|
|
else:
|
|
key = super().keypress(size, key)
|
|
self.update_edit_nick()
|
|
return key
|
|
|
|
else:
|
|
return super().keypress(size, key)
|
|
|
|
return None
|
|
|
|
## Euph stuff
|
|
|
|
# Euph actions
|
|
|
|
@synchronous
|
|
async def request_logs(self):
|
|
oldest_id = self._supply.oldest_id()
|
|
if oldest_id is not None:
|
|
messages = await self._room.log(self._log_amount, oldest_id)
|
|
for message in messages:
|
|
self.receive_message(message)
|
|
self.update_tree()
|
|
|
|
self._requesting_logs = False
|
|
|
|
@synchronous
|
|
async def nick(self, nick: str):
|
|
new_nick = await self._room.nick(nick)
|
|
self.own_nick_change()
|
|
|
|
@synchronous
|
|
async def send(self, content: str, parent_id: Optional[str]):
|
|
message = await self._room.send(content, parent_id=parent_id)
|
|
self.receive_message(message)
|
|
self.update_tree()
|
|
|
|
# Reacting to euph events
|
|
|
|
async def on_snapshot(self, messages: List[yaboli.Message]):
|
|
for message in messages:
|
|
self.receive_message(message)
|
|
self.update_tree()
|
|
|
|
async def on_send(self, message: yaboli.Message):
|
|
self.receive_message(message)
|
|
self.update_tree()
|
|
|
|
urwid.register_signal(RoomWidget, ["close"])
|