From 046a819ee817c6735f6fb0e41edcb36c6b143105 Mon Sep 17 00:00:00 2001 From: Christian Heller Date: Mon, 4 Aug 2025 11:41:28 +0200 Subject: [PATCH] Major restructuring. --- ircplom.py | 7 +- ircplom/client.py | 277 +++++++++++++++++++++++++++++++ ircplom/client_tui.py | 135 +++++++++++++++ ircplom/irc_conn.py | 283 ++------------------------------ ircplom/{tui.py => tui_base.py} | 150 ++--------------- 5 files changed, 445 insertions(+), 407 deletions(-) create mode 100644 ircplom/client.py create mode 100644 ircplom/client_tui.py rename ircplom/{tui.py => tui_base.py} (81%) diff --git a/ircplom.py b/ircplom.py index 16b6375..b10fb65 100755 --- a/ircplom.py +++ b/ircplom.py @@ -2,8 +2,9 @@ 'Attempt at an IRC client.' from queue import SimpleQueue from ircplom.events import ExceptionEvent, QuitEvent -from ircplom.irc_conn import ClientsDb, ClientEvent, NewClientEvent -from ircplom.tui import Terminal, Tui, TuiEvent +from ircplom.client import ClientsDb, ClientEvent, NewClientEvent +from ircplom.tui_base import Terminal, TuiEvent +from ircplom.client_tui import ClientTui def main_loop() -> None: @@ -12,7 +13,7 @@ def main_loop() -> None: clients_db: ClientsDb = {} try: with Terminal(q_out=q_events).setup() as term: - tui = Tui(q_out=q_events, term=term) + tui = ClientTui(q_out=q_events, term=term) while True: event = q_events.get() if isinstance(event, QuitEvent): diff --git a/ircplom/client.py b/ircplom/client.py new file mode 100644 index 0000000..9ddea32 --- /dev/null +++ b/ircplom/client.py @@ -0,0 +1,277 @@ +'High-level IRC protocol / server connection management.' +# built-ins +from abc import ABC, abstractmethod +from dataclasses import dataclass +from getpass import getuser +from threading import Thread +from typing import Optional +from uuid import UUID, uuid4 +# ourselves +from ircplom.events import (AffectiveEvent, ExceptionEvent, PayloadMixin, + QueueMixin) +from ircplom.irc_conn import (BaseIrcConnection, IrcConnAbortException, + IrcMessage) + +ClientsDb = dict[UUID, 'Client'] +CHAT_GLOB = '*' + + +@dataclass +class ClientIdMixin: + 'Collects a Client\'s ID at .client_id.' + client_id: UUID + + +@dataclass +class ClientEvent(AffectiveEvent, ClientIdMixin): + 'To affect Client identified by ClientIdMixin.' + + +class _IrcConnection(BaseIrcConnection): + + def __init__(self, client_id: UUID, **kwargs) -> None: + # TODO: find out why I can't just ClientIdMixin here + self.client_id = client_id + super().__init__(**kwargs) + + def _make_recv_event(self, msg: IrcMessage) -> ClientEvent: + + @dataclass + class _RecvEvent(ClientEvent, PayloadMixin): + payload: IrcMessage + + def affect(self, target: 'Client') -> None: + target.handle_msg(self.payload) + + return _RecvEvent(client_id=self.client_id, payload=msg) + + +@dataclass +class ClientQueueMixin(QueueMixin): + 'To QueueMixin adds _cput to extend ._put with client_id= setting.' + client_id_name = 'id_' + + def _cput(self, event_class, **kwargs) -> None: + self._put(event_class(client_id=getattr(self, self.client_id_name), + **kwargs)) + + +@dataclass +class _ServerCapability: + 'Store data collectable via CAPS LS/LIST/NEW.' + enabled: bool + data: str + + def str_for_log(self, name: str) -> str: + 'Optimized for Client.log per-line listing.' + listing = '+' if self.enabled else '-' + listing += f' {name}' + if self.data: + listing += f' ({self.data})' + return listing + + +@dataclass +class IrcConnSetup: + 'All we need to know to set up a new Client connection.' + hostname: str + nickname: str + realname: str + + +class Client(ABC, ClientQueueMixin): + 'Abstracts socket connection, loop over it, and handling messages from it.' + nick_confirmed: bool = False + conn: Optional[_IrcConnection] = None + + def __init__(self, conn_setup: IrcConnSetup, **kwargs) -> None: + super().__init__(**kwargs) + self.conn_setup = conn_setup + self._cap_neg_states: dict[str, bool] = {} + self.caps: dict[str, _ServerCapability] = {} + self.id_ = uuid4() + self.update_login(nick_confirmed=False, + nickname=self.conn_setup.nickname) + self.start_connecting() + + def start_connecting(self) -> None: + 'Start thread to set up _IrcConnection at .conn.' + + def connect(self) -> None: + + @dataclass + class _ConnectedEvent(ClientEvent): + def affect(self, target: 'Client') -> None: + target.on_connect() + + try: + self.conn = _IrcConnection(hostname=self.conn_setup.hostname, + q_out=self.q_out, + client_id=self.id_) + self._cput(_ConnectedEvent) + except IrcConnAbortException as e: + self.log(f'# ALERT: {e}') + except Exception as e: # pylint: disable=broad-exception-caught + self._put(ExceptionEvent(e)) + + Thread(target=connect, daemon=True, args=(self,)).start() + + def on_connect(self) -> None: + 'Steps to perform right after connection.' + self.log(msg='# connected to server', chat=CHAT_GLOB) + self.try_send_cap('LS', ('302',)) + self.send(IrcMessage(verb='USER', + params=(getuser(), '0', '*', + self.conn_setup.realname))) + self.send(IrcMessage(verb='NICK', params=(self.conn_setup.nickname,))) + + def cap_neg_done(self, negotiation_step: str) -> bool: + 'Whether negotiation_step is registered as finished.' + return self._cap_neg_states.get(negotiation_step, False) + + def cap_neg(self, negotiation_step: str) -> bool: + 'Whether negotiation_step is registered at all (started or finished).' + return negotiation_step in self._cap_neg_states + + def cap_neg_set(self, negotiation_step: str, done: bool = False) -> None: + 'Declare negotiation_step started, or (if done) finished.' + self._cap_neg_states[negotiation_step] = done + + def try_send_cap(self, *params, key_fused: bool = False) -> None: + 'Run CAP command with params, handle cap neg. state.' + neg_state_key = ':'.join(params) if key_fused else params[0] + if self.cap_neg(neg_state_key): + return + self.send(IrcMessage(verb='CAP', params=params)) + self.cap_neg_set(neg_state_key) + + def collect_caps(self, params: tuple[str, ...]) -> None: + 'Record available and enabled server capabilities.' + verb = params[0] + items = params[-1].strip().split() + is_final_line = params[1] != '*' + if self.cap_neg_done(verb): + if verb == 'LS': + self.caps.clear() + else: + for cap in self.caps.values(): + cap.enabled = False + self.cap_neg_set(verb) + for item in items: + if verb == 'LS': + splitted = item.split('=', maxsplit=1) + self.caps[splitted[0]] = _ServerCapability( + enabled=False, data=''.join(splitted[1:])) + else: + self.caps[item].enabled = True + if is_final_line: + self.cap_neg_set(verb, done=True) + + @abstractmethod + def log(self, msg: str, chat: str = '') -> None: + '''Write msg into log of chat, whatever shape that may have. + + Messages to chat=CHAT_GLOB are meant to appear in all widgets mapped to + the client, those to chat="" only in the initial connection window. + ''' + + def send(self, msg: IrcMessage, chat: str = '') -> None: + 'Send line-separator-delimited message over socket.' + if not self.conn: + self.log('# ALERT: cannot send, connection seems closed') + return + self.conn.send(msg) + self.log(msg=f'> {msg.raw}', chat=chat) + self.log(msg=f'=>| {msg.raw}', chat=':raw') + + def update_login(self, nick_confirmed: bool, nickname: str = '') -> None: + '''Manage conn_setup..nickname, .nick_confirmed. + + (Useful for subclass extension.) + ''' + first_run = not hasattr(self.conn_setup, 'nickname') + prefix = '# nickname' + if first_run or (nickname and nickname != self.conn_setup.nickname): + verb = ('set' if first_run + else f'changed from "{self.conn_setup.nickname}"') + self.conn_setup.nickname = nickname + self.log(msg=f'{prefix} {verb} to "{nickname}"', chat=CHAT_GLOB) + if first_run or nick_confirmed != self.nick_confirmed: + self.nick_confirmed = nick_confirmed + if not first_run: + self.log(f'{prefix} {"" if nick_confirmed else "un"}confirmed') + + def close(self) -> None: + 'Close both recv Loop and socket.' + self.log(msg='# disconnecting from server', chat=CHAT_GLOB) + if self.conn: + self.conn.close() + self.conn = None + self.update_login(nick_confirmed=False) + + def handle_msg(self, msg: IrcMessage) -> None: + 'Process incoming msg towards appropriate client steps.' + self.log(f'<-| {msg.raw}', ':raw') + match msg.verb: + case 'PING': + self.send(IrcMessage(verb='PONG', params=(msg.params[0],))) + case 'ERROR': + self.close() + case '001' | 'NICK': + self.update_login(nickname=msg.params[0], nick_confirmed=True) + case 'PRIVMSG': + self.log(msg=str(msg.params), chat=msg.source) + case 'CAP': + match msg.params[1]: + case 'LS' | 'LIST': + self.collect_caps(msg.params[1:]) + case 'ACK' | 'NAK': + cap_names = msg.params[-1].split() + for cap_name in cap_names: + self.cap_neg_set(f'REQ:{cap_name}', done=True) + self.caps[cap_name].enabled = (msg.params[1] + == 'ACK') + if self.cap_neg_done('LIST'): + self.try_send_cap('END') + if not self.cap_neg('printing'): + self.log('# server capabilities (enabled: "+"):') + for cap_name, cap in self.caps.items(): + self.log('# ' + cap.str_for_log(cap_name)) + self.cap_neg_set('printing', done=True) + elif self.cap_neg_done('LS'): + for cap_name in ('server-time', 'account-tag', 'sasl'): + if (cap_name in self.caps + and (not self.caps[cap_name].enabled)): + self.try_send_cap('REQ', cap_name, key_fused=True) + self.try_send_cap('LIST') + + +@dataclass +class NewClientEvent(AffectiveEvent, PayloadMixin): + 'Put Client .payload into ClientsDb target.' + payload: 'Client' + + def affect(self, target: ClientsDb) -> None: + target[self.payload.id_] = self.payload + + +@dataclass +class InitReconnectEvent(ClientEvent): + 'To trigger re-opening of connection.' + + def affect(self, target: 'Client') -> None: + if target.conn: + target.log('# ALERT: reconnection called, but still seem ' + 'connected, so nothing to do.') + else: + target.start_connecting() + + +@dataclass +class SendEvent(ClientEvent, PayloadMixin): + 'To trigger sending of payload to server.' + payload: IrcMessage + chat: str = '' + + def affect(self, target: 'Client') -> None: + target.send(msg=self.payload, chat=self.chat) diff --git a/ircplom/client_tui.py b/ircplom/client_tui.py new file mode 100644 index 0000000..7c482ab --- /dev/null +++ b/ircplom/client_tui.py @@ -0,0 +1,135 @@ +'TUI adaptions to Client.' +# built-ins +from dataclasses import dataclass +from uuid import UUID +# ourselves +from ircplom.events import PayloadMixin +from ircplom.tui_base import (BaseTui, Window, TuiEvent, CMD_SHORTCUTS, + PROMPT_TEMPLATE) +from ircplom.irc_conn import IrcMessage +from ircplom.client import (CHAT_GLOB, IrcConnSetup, Client, + ClientIdMixin, ClientQueueMixin, + InitReconnectEvent, NewClientEvent, SendEvent) + + +CMD_SHORTCUTS['disconnect'] = 'window.disconnect' +CMD_SHORTCUTS['nick'] = 'window.nick' +CMD_SHORTCUTS['privmsg'] = 'window.privmsg' +CMD_SHORTCUTS['reconnect'] = 'window.reconnect' + + +class _ClientWindow(Window, ClientQueueMixin): + client_id_name = 'client_id' + + def __init__(self, client_id: UUID, chat: str = '', **kwargs) -> None: + self.client_id = client_id + self.chat = chat + super().__init__(**kwargs) + + def cmd__disconnect(self, quit_msg: str = 'ircplom says bye') -> None: + 'Send QUIT command to server.' + self._cput(SendEvent, + payload=IrcMessage(verb='QUIT', params=(quit_msg,))) + + def cmd__reconnect(self) -> None: + 'Attempt reconnection.' + self._cput(InitReconnectEvent) + + def cmd__nick(self, new_nick: str) -> None: + 'Attempt nickname change.' + self._cput(SendEvent, + payload=IrcMessage(verb='NICK', params=(new_nick,))) + + def cmd__privmsg(self, target: str, msg: str) -> None: + 'Send chat message msg to target.' + self._cput(SendEvent, chat=target, + payload=IrcMessage(verb='PRIVMSG', params=(target, msg))) + + +class ClientTui(BaseTui): + 'TUI expanded towards Client features.' + + def _new_client_window(self, client_id: UUID, chat: str = '' + ) -> _ClientWindow: + new_idx = len(self.windows) + win = _ClientWindow(idx=new_idx, term=self.term, q_out=self.q_out, + client_id=client_id, chat=chat) + self.windows += [win] + self._switch_window(new_idx) + return win + + def client_wins(self, client_id: UUID) -> list[_ClientWindow]: + 'All _ClientWindows matching client_id; if none, create one.' + wins = [win for win in self.windows + if isinstance(win, _ClientWindow) + and win.client_id == client_id] # pylint: disable=no-member + if not wins: + wins = [self._new_client_window(client_id=client_id)] + return wins + + def client_win(self, client_id: UUID, chat: str = '') -> _ClientWindow: + '''That _ClientWindow matching client_id and chat; create if none. + + In case of creation, copy prompt prefix from client's first window. + ''' + client_wins = self.client_wins(client_id) + candidates = [win for win in client_wins if win.chat == chat] + if candidates: + return candidates[0] + win = self._new_client_window(client_id=client_id, chat=chat) + if client_wins: + win.prompt.prefix = client_wins[0].prompt.prefix + return win + + def cmd__connect(self, hostname: str, nickname: str, realname: str + ) -> None: + 'Create Client and pass it via NewClientEvent.' + self._put(NewClientEvent( + _ClientKnowingTui( + q_out=self.q_out, + conn_setup=IrcConnSetup(hostname=hostname, nickname=nickname, + realname=realname)))) + + +@dataclass +class _ClientWindowEvent(TuiEvent, ClientIdMixin): + chat: str = '' + + +@dataclass +class _ClientLogEvent(_ClientWindowEvent, PayloadMixin): + payload: str + + def affect(self, target: ClientTui) -> None: + if self.chat == CHAT_GLOB: + for win in target.client_wins(self.client_id): + win.log.append(self.payload) + else: + target.client_win(self.client_id, self.chat + ).log.append(self.payload) + super().affect(target) + + +@dataclass +class _ClientPromptEvent(_ClientWindowEvent, PayloadMixin): + payload: tuple[str, str] + + def affect(self, target: ClientTui) -> None: + new_prefix = ((' ' if self.payload[0] else '?') + + f'{self.payload[1]}{PROMPT_TEMPLATE}') + for win in target.client_wins(self.client_id): + prompt = win.prompt + prompt.prefix = new_prefix + prompt.tainted = True + super().affect(target) + + +class _ClientKnowingTui(Client): + + def log(self, msg: str, chat: str = '') -> None: + self._cput(_ClientLogEvent, chat=chat, payload=msg) + + def update_login(self, nick_confirmed: bool, nickname: str = '') -> None: + super().update_login(nick_confirmed, nickname) + self._cput(_ClientPromptEvent, payload=(self.nick_confirmed, + self.conn_setup.nickname)) diff --git a/ircplom/irc_conn.py b/ircplom/irc_conn.py index ab9426d..f7f5455 100644 --- a/ircplom/irc_conn.py +++ b/ircplom/irc_conn.py @@ -1,21 +1,12 @@ -'IRC server connection management.' +'Low-level IRC protocol / server connection management.' # built-ins from abc import ABC, abstractmethod -from dataclasses import dataclass -from getpass import getuser from socket import socket, gaierror as socket_gaierror -from threading import Thread from typing import Callable, Iterator, NamedTuple, Optional, Self -from uuid import uuid4, UUID # ourselves -from ircplom.events import ( - AffectiveEvent, ExceptionEvent, Loop, PayloadMixin, QueueMixin) +from ircplom.events import Event, Loop, QueueMixin -ClientsDb = dict[UUID, 'Client'] - -CHAT_GLOB = '*' - _TIMEOUT_RECV_LOOP = 0.1 _TIMEOUT_CONNECT = 5 _CONN_RECV_BUFSIZE = 1024 @@ -129,22 +120,21 @@ class IrcMessage: return self._raw -class _IrcConnAbortException(Exception): - pass +class IrcConnAbortException(Exception): + 'Thrown by BaseIrcConnection on expectable connection failures.' -class _IrcConnection(QueueMixin): +class BaseIrcConnection(QueueMixin, ABC): 'Collects low-level server-client connection management.' - def __init__(self, hostname: str, client_id: UUID, **kwargs) -> None: + def __init__(self, hostname: str, **kwargs) -> None: super().__init__(**kwargs) - self.client_id = client_id self._socket = socket() self._socket.settimeout(_TIMEOUT_CONNECT) try: self._socket.connect((hostname, _PORT)) except (TimeoutError, socket_gaierror) as e: - raise _IrcConnAbortException(e) from e + raise IrcConnAbortException(e) from e self._socket.settimeout(_TIMEOUT_RECV_LOOP) self._recv_loop = Loop(iterator=self._read_lines(), q_out=self.q_out) @@ -157,7 +147,11 @@ class _IrcConnection(QueueMixin): 'Send line-separator-delimited message over socket.' self._socket.sendall(msg.raw.encode('utf-8') + _IRCSPEC_LINE_SEPARATOR) - def _read_lines(self) -> Iterator[Optional['_RecvEvent']]: + @abstractmethod + def _make_recv_event(self, msg: IrcMessage) -> Event: + pass + + def _read_lines(self) -> Iterator[Optional[Event]]: assert self._socket is not None bytes_total = b'' buffer_linesep = b'' @@ -186,255 +180,6 @@ class _IrcConnection(QueueMixin): buffer_linesep += c_byted if buffer_linesep == _IRCSPEC_LINE_SEPARATOR: buffer_linesep = b'' - yield _RecvEvent(client_id=self.client_id, - payload=IrcMessage.from_raw( - bytes_total.decode('utf-8'))) + yield self._make_recv_event( + IrcMessage.from_raw(bytes_total.decode('utf-8'))) bytes_total = b'' - - -@dataclass -class ClientIdMixin: - 'Collects a Client\'s ID at .client_id.' - client_id: UUID - - -@dataclass -class NewClientEvent(AffectiveEvent, PayloadMixin): - 'Put Client .payload into ClientsDb target.' - payload: 'Client' - - def affect(self, target: ClientsDb) -> None: - target[self.payload.id_] = self.payload - - -@dataclass -class ClientEvent(AffectiveEvent, ClientIdMixin): - 'To affect Client identified by ClientIdMixin.' - - -@dataclass -class _ConnectedEvent(ClientEvent): - - def affect(self, target: 'Client') -> None: - target.log(msg='# connected to server', chat=CHAT_GLOB) - target.try_send_cap('LS', ('302',)) - target.send(IrcMessage(verb='USER', - params=(getuser(), '0', '*', - target.conn_setup.realname))) - target.send(IrcMessage(verb='NICK', - params=(target.conn_setup.nickname,))) - - -@dataclass -class InitReconnectEvent(ClientEvent): - 'To trigger re-opening of connection.' - - def affect(self, target: 'Client') -> None: - if target.conn: - target.log('# ALERT: reconnection called, but still seem ' - 'connected, so nothing to do.') - else: - target.start_connecting() - - -@dataclass -class SendEvent(ClientEvent, PayloadMixin): - 'To trigger sending of payload to server.' - payload: IrcMessage - chat: str = '' - - def affect(self, target: 'Client') -> None: - target.send(msg=self.payload, chat=self.chat) - - -@dataclass -class _RecvEvent(ClientEvent, PayloadMixin): - payload: IrcMessage - - def affect(self, target: 'Client') -> None: - target.handle_msg(self.payload) - - -@dataclass -class ClientQueueMixin(QueueMixin): - 'To QueueMixin adds _cput to extend ._put with client_id= setting.' - client_id_name = 'id_' - - def _cput(self, event_class, **kwargs) -> None: - self._put(event_class(client_id=getattr(self, self.client_id_name), - **kwargs)) - - -@dataclass -class ServerCapability: - 'Store data collectable via CAPS LS/LIST/NEW.' - enabled: bool - data: str - - def str_for_log(self, name: str) -> str: - 'Optimized for Client.log per-line listing.' - listing = '+' if self.enabled else '-' - listing += f' {name}' - if self.data: - listing += f' ({self.data})' - return listing - - -@dataclass -class IrcConnSetup: - 'All we need to know to set up a new Client connection.' - hostname: str - nickname: str - realname: str - - -class Client(ABC, ClientQueueMixin): - 'Abstracts socket connection, loop over it, and handling messages from it.' - nick_confirmed: bool = False - conn: Optional[_IrcConnection] = None - - def __init__(self, conn_setup: IrcConnSetup, **kwargs) -> None: - super().__init__(**kwargs) - self.conn_setup = conn_setup - self._cap_neg_states: dict[str, bool] = {} - self.caps: dict[str, ServerCapability] = {} - self.id_ = uuid4() - self.update_login(nick_confirmed=False, - nickname=self.conn_setup.nickname) - self.start_connecting() - - def start_connecting(self) -> None: - 'Start thread to set up IrcConnection at .conn.' - - def connect(self) -> None: - try: - self.conn = _IrcConnection(hostname=self.conn_setup.hostname, - q_out=self.q_out, - client_id=self.id_) - self._cput(_ConnectedEvent) - except _IrcConnAbortException as e: - self.log(f'# ALERT: {e}') - except Exception as e: # pylint: disable=broad-exception-caught - self._put(ExceptionEvent(e)) - - Thread(target=connect, daemon=True, args=(self,)).start() - - def cap_neg_done(self, negotiation_step: str) -> bool: - 'Whether negotiation_step is registered as finished.' - return self._cap_neg_states.get(negotiation_step, False) - - def cap_neg(self, negotiation_step: str) -> bool: - 'Whether negotiation_step is registered at all (started or finished).' - return negotiation_step in self._cap_neg_states - - def cap_neg_set(self, negotiation_step: str, done: bool = False) -> None: - 'Declare negotiation_step started, or (if done) finished.' - self._cap_neg_states[negotiation_step] = done - - def try_send_cap(self, *params, key_fused: bool = False) -> None: - 'Run CAP command with params, handle cap neg. state.' - neg_state_key = ':'.join(params) if key_fused else params[0] - if self.cap_neg(neg_state_key): - return - self.send(IrcMessage(verb='CAP', params=params)) - self.cap_neg_set(neg_state_key) - - def collect_caps(self, params: tuple[str, ...]) -> None: - 'Record available and enabled server capabilities.' - verb = params[0] - items = params[-1].strip().split() - is_final_line = params[1] != '*' - if self.cap_neg_done(verb): - if verb == 'LS': - self.caps.clear() - else: - for cap in self.caps.values(): - cap.enabled = False - self.cap_neg_set(verb) - for item in items: - if verb == 'LS': - splitted = item.split('=', maxsplit=1) - self.caps[splitted[0]] = ServerCapability( - enabled=False, data=''.join(splitted[1:])) - else: - self.caps[item].enabled = True - if is_final_line: - self.cap_neg_set(verb, done=True) - - @abstractmethod - def log(self, msg: str, chat: str = '') -> None: - '''Write msg into log of chat, whatever shape that may have. - - Messages to chat=CHAT_GLOB are meant to appear in all widgets mapped to - the client, those to chat="" only in the initial connection window. - ''' - - def send(self, msg: IrcMessage, chat: str = '') -> None: - 'Send line-separator-delimited message over socket.' - if not self.conn: - self.log('# ALERT: cannot send, connection seems closed') - return - self.conn.send(msg) - self.log(msg=f'> {msg.raw}', chat=chat) - self.log(msg=f'=>| {msg.raw}', chat=':raw') - - def update_login(self, nick_confirmed: bool, nickname: str = '') -> None: - '''Manage conn_setup..nickname, .nick_confirmed. - - (Useful for subclass extension.) - ''' - first_run = not hasattr(self.conn_setup, 'nickname') - prefix = '# nickname' - if first_run or (nickname and nickname != self.conn_setup.nickname): - verb = ('set' if first_run - else f'changed from "{self.conn_setup.nickname}"') - self.conn_setup.nickname = nickname - self.log(msg=f'{prefix} {verb} to "{nickname}"', chat=CHAT_GLOB) - if first_run or nick_confirmed != self.nick_confirmed: - self.nick_confirmed = nick_confirmed - if not first_run: - self.log(f'{prefix} {"" if nick_confirmed else "un"}confirmed') - - def close(self) -> None: - 'Close both recv Loop and socket.' - self.log(msg='# disconnecting from server', chat=CHAT_GLOB) - if self.conn: - self.conn.close() - self.conn = None - self.update_login(nick_confirmed=False) - - def handle_msg(self, msg: IrcMessage) -> None: - 'Process incoming msg towards appropriate client steps.' - self.log(f'<-| {msg.raw}', ':raw') - match msg.verb: - case 'PING': - self.send(IrcMessage(verb='PONG', params=(msg.params[0],))) - case 'ERROR': - self.close() - case '001' | 'NICK': - self.update_login(nickname=msg.params[0], nick_confirmed=True) - case 'PRIVMSG': - self.log(msg=str(msg.params), chat=msg.source) - case 'CAP': - match msg.params[1]: - case 'LS' | 'LIST': - self.collect_caps(msg.params[1:]) - case 'ACK' | 'NAK': - cap_names = msg.params[-1].split() - for cap_name in cap_names: - self.cap_neg_set(f'REQ:{cap_name}', done=True) - self.caps[cap_name].enabled = (msg.params[1] - == 'ACK') - if self.cap_neg_done('LIST'): - self.try_send_cap('END') - if not self.cap_neg('printing'): - self.log('# server capabilities (enabled: "+"):') - for cap_name, cap in self.caps.items(): - self.log('# ' + cap.str_for_log(cap_name)) - self.cap_neg_set('printing', done=True) - elif self.cap_neg_done('LS'): - for cap_name in ('server-time', 'account-tag', 'sasl'): - if (cap_name in self.caps - and (not self.caps[cap_name].enabled)): - self.try_send_cap('REQ', cap_name, key_fused=True) - self.try_send_cap('LIST') diff --git a/ircplom/tui.py b/ircplom/tui_base.py similarity index 81% rename from ircplom/tui.py rename to ircplom/tui_base.py index 39b51f3..e19e90f 100644 --- a/ircplom/tui.py +++ b/ircplom/tui_base.py @@ -1,4 +1,4 @@ -'Terminal and TUI management.' +'Base Terminal and TUI management.' # built-ins from abc import ABC, abstractmethod from base64 import b64decode @@ -7,15 +7,12 @@ from dataclasses import dataclass from inspect import _empty as inspect_empty, signature, stack from signal import SIGWINCH, signal from typing import Callable, Generator, Iterator, NamedTuple, Optional -from uuid import UUID # requirements.txt from blessed import Terminal as BlessedTerminal # ourselves from ircplom.events import ( AffectiveEvent, Loop, PayloadMixin, QueueMixin, QuitEvent) -from ircplom.irc_conn import ( - CHAT_GLOB, IrcConnSetup, IrcMessage, Client, ClientIdMixin, - ClientQueueMixin, InitReconnectEvent, NewClientEvent, SendEvent) +# from ircplom.irc_conn import IrcMessage _MIN_HEIGHT = 4 _MIN_WIDTH = 32 @@ -25,7 +22,7 @@ _B64_PREFIX = 'b64:' _OSC52_PREFIX = b']52;c;' _PASTE_DELIMITER = '\007' -_PROMPT_TEMPLATE = '> ' +PROMPT_TEMPLATE = '> ' _PROMPT_ELL_IN = '<…' _PROMPT_ELL_OUT = '…>' @@ -43,19 +40,14 @@ _KEYBINDINGS = { 'esc:91:49:59:51:67': ('window', 'right'), 'KEY_F1': ('window.paste',), } -_CMD_SHORTCUTS = { - 'disconnect': 'window.disconnect', - 'nick': 'window.nick', - 'privmsg': 'window.privmsg', - 'reconnect': 'window.reconnect' -} +CMD_SHORTCUTS: dict[str, str] = {} @dataclass class TuiEvent(AffectiveEvent): 'To affect TUI, and trigger flushed .draw_tainted on it.' - def affect(self, target: 'Tui') -> None: + def affect(self, target) -> None: target.window.draw_tainted() target.term.flush() @@ -63,7 +55,7 @@ class TuiEvent(AffectiveEvent): @dataclass class _SetScreenEvent(TuiEvent): - def affect(self, target: 'Tui') -> None: + def affect(self, target: 'BaseTui') -> None: target.term.calc_geometry() for window in target.windows: window.set_geometry() @@ -205,7 +197,7 @@ class _PromptWidget(_ScrollableWidget): def __init__(self, **kwargs) -> None: super().__init__(**kwargs) - self.prefix = _PROMPT_TEMPLATE + self.prefix = PROMPT_TEMPLATE self._reset_buffer('') @property @@ -309,7 +301,8 @@ class _PromptWidget(_ScrollableWidget): return to_return -class _Window(_Widget): +class Window(_Widget): + 'Widget filling entire screen with sub-widgets like .prompt, .log.' _y_status: int prompt: _PromptWidget @@ -374,7 +367,7 @@ class _Window(_Widget): class _KeyboardEvent(TuiEvent, PayloadMixin): payload: str - def affect(self, target: 'Tui') -> None: + def affect(self, target: 'BaseTui') -> None: if self.payload[0] == _CHAR_RESIZE: _SetScreenEvent().affect(target) return @@ -403,19 +396,19 @@ class _KeyboardEvent(TuiEvent, PayloadMixin): super().affect(target) -class Tui(QueueMixin): +class BaseTui(QueueMixin): 'Base for graphical user interface elements.' def __init__(self, term: 'Terminal', **kwargs) -> None: super().__init__(**kwargs) self.term = term self._window_idx = 0 - self.windows = [_Window(idx=self._window_idx, term=self.term)] + self.windows = [Window(idx=self._window_idx, term=self.term)] self._put(_SetScreenEvent()) def cmd_name_to_cmd(self, cmd_name: str) -> Optional[Callable]: 'Map cmd_name to executable TUI element method.' - cmd_name = _CMD_SHORTCUTS.get(cmd_name, cmd_name) + cmd_name = CMD_SHORTCUTS.get(cmd_name, cmd_name) cmd_parent = self while True: cmd_name_toks = cmd_name.split('.', maxsplit=1) @@ -431,42 +424,10 @@ class Tui(QueueMixin): return getattr(cmd_parent, cmd_name) @property - def window(self) -> _Window: - 'Currently selected _Window.' + def window(self) -> Window: + 'Currently selected Window.' return self.windows[self._window_idx] - def _new_client_window(self, client_id: UUID, chat: str = '' - ) -> '_ClientWindow': - new_idx = len(self.windows) - win = _ClientWindow(idx=new_idx, term=self.term, q_out=self.q_out, - client_id=client_id, chat=chat) - self.windows += [win] - self._switch_window(new_idx) - return win - - def client_wins(self, client_id: UUID) -> list['_ClientWindow']: - 'All _ClientWindows matching client_id; if none, create one.' - wins = [win for win in self.windows - if isinstance(win, _ClientWindow) - and win.client_id == client_id] # pylint: disable=no-member - if not wins: - wins = [self._new_client_window(client_id=client_id)] - return wins - - def client_win(self, client_id: UUID, chat: str = '') -> '_ClientWindow': - '''That _ClientWindow matching client_id and chat; create if none. - - In case of creation, copy prompt prefix from client's first window. - ''' - client_wins = self.client_wins(client_id) - candidates = [win for win in client_wins if win.chat == chat] - if candidates: - return candidates[0] - win = self._new_client_window(client_id=client_id, chat=chat) - if client_wins: - win.prompt.prefix = client_wins[0].prompt.prefix - return win - def log(self, msg: str) -> None: 'Post msg to active window\'s log.' self.window.log.append(msg) @@ -475,15 +436,6 @@ class Tui(QueueMixin): self._window_idx = idx self.window.draw() - def cmd__connect(self, hostname: str, nickname: str, realname: str - ) -> None: - 'Create Client and pass it via NewClientEvent.' - self._put(NewClientEvent( - _ClientKnowingTui( - q_out=self.q_out, - conn_setup=IrcConnSetup(hostname=hostname, nickname=nickname, - realname=realname)))) - def cmd__prompt_enter(self) -> None: 'Get prompt content from .window.prompt.enter, parse to & run command.' to_parse = self.window.prompt.enter() @@ -651,75 +603,3 @@ class Terminal(QueueMixin): else: to_yield = 'esc:' + ':'.join([str(int(b)) for b in chars]) yield _KeyboardEvent(to_yield) if to_yield else None - - -class _ClientWindow(_Window, ClientQueueMixin): - client_id_name = 'client_id' - - def __init__(self, client_id: UUID, chat: str = '', **kwargs) -> None: - self.client_id = client_id - self.chat = chat - super().__init__(**kwargs) - - def cmd__disconnect(self, quit_msg: str = 'ircplom says bye') -> None: - 'Send QUIT command to server.' - self._cput(SendEvent, - payload=IrcMessage(verb='QUIT', params=(quit_msg,))) - - def cmd__reconnect(self) -> None: - 'Attempt reconnection.' - self._cput(InitReconnectEvent) - - def cmd__nick(self, new_nick: str) -> None: - 'Attempt nickname change.' - self._cput(SendEvent, - payload=IrcMessage(verb='NICK', params=(new_nick,))) - - def cmd__privmsg(self, target: str, msg: str) -> None: - 'Send chat message msg to target.' - self._cput(SendEvent, chat=target, - payload=IrcMessage(verb='PRIVMSG', params=(target, msg))) - - -@dataclass -class _ClientWindowEvent(TuiEvent, ClientIdMixin): - chat: str = '' - - -@dataclass -class _ClientLogEvent(_ClientWindowEvent, PayloadMixin): - payload: str - - def affect(self, target: Tui) -> None: - if self.chat == CHAT_GLOB: - for win in target.client_wins(self.client_id): - win.log.append(self.payload) - else: - target.client_win(self.client_id, self.chat - ).log.append(self.payload) - super().affect(target) - - -@dataclass -class _ClientPromptEvent(_ClientWindowEvent, PayloadMixin): - payload: tuple[str, str] - - def affect(self, target: Tui) -> None: - new_prefix = ((' ' if self.payload[0] else '?') - + f'{self.payload[1]}{_PROMPT_TEMPLATE}') - for win in target.client_wins(self.client_id): - prompt = win.prompt - prompt.prefix = new_prefix - prompt.tainted = True - super().affect(target) - - -class _ClientKnowingTui(Client): - - def log(self, msg: str, chat: str = '') -> None: - self._cput(_ClientLogEvent, chat=chat, payload=msg) - - def update_login(self, nick_confirmed: bool, nickname: str = '') -> None: - super().update_login(nick_confirmed, nickname) - self._cput(_ClientPromptEvent, payload=(self.nick_confirmed, - self.conn_setup.nickname)) -- 2.30.2