From fc0852f6e45914fdd9b5c5f386eca94fcdb3b40a Mon Sep 17 00:00:00 2001 From: Christian Heller Date: Sun, 17 Aug 2025 13:17:51 +0200 Subject: [PATCH] Use proper data structure for update passing, rather than to-parse strings. --- ircplom/client.py | 34 ++++++---- ircplom/client_tui.py | 145 ++++++++++++++++++++++-------------------- 2 files changed, 97 insertions(+), 82 deletions(-) diff --git a/ircplom/client.py b/ircplom/client.py index 777e1ad..b376ad7 100644 --- a/ircplom/client.py +++ b/ircplom/client.py @@ -14,7 +14,6 @@ from ircplom.irc_conn import (BaseIrcConnection, IrcConnAbortException, IrcMessage, PORT_SSL) ClientsDb = dict[str, 'Client'] -CLEAR_WORD = 'CLEAR' _NAMES_DESIRED_SERVER_CAPS = ('server-time', 'account-tag', 'sasl') # NB: in below numerics accounting, tuples define inclusive ranges @@ -218,12 +217,9 @@ class Db: for c in self.__class__.__mro__: if hasattr(c, '__annotations__'): self._types = c.__annotations__ | self._types - self._set_empty_defaults() - super().__init__(**kwargs) - - def _set_empty_defaults(self) -> None: for name, type_ in self._types.items(): setattr(self, name, type_()) + super().__init__(**kwargs) def _typecheck(self, key: str, value) -> None: type_ = self._types[key] @@ -321,13 +317,17 @@ class _UpdatingDict: self._dict: dict[str, str] = {} def set_on_update(self, name: str, on_update: Callable) -> None: - 'Set on_update caller for "d {name} {key}."' - self._on_update = lambda k: on_update(f'd {name} {k}') + 'Caller of on_update with path= set to name.' + self._on_update = lambda k: on_update(name, k) def clear(self) -> None: - 'Zero dict and send CLEAR_WORD update.' + 'Zero dict and send clearance update.' self._dict.clear() - self._on_update(CLEAR_WORD) + self._on_update('') + + def has(self, key: str) -> bool: + 'Test if entry of name in dictionary.' + return key in self._dict def __getitem__(self, key: str) -> str: return self._dict[key] @@ -338,7 +338,7 @@ class _UpdatingDict: def __delitem__(self, key: str) -> None: del self._dict[key] - self._on_update(f'-{key}') + self._on_update(key) class _ClientDb(_Db, IrcConnSetup): @@ -350,16 +350,24 @@ class _ClientDb(_Db, IrcConnSetup): _completable_motd: _CompletableStringsList _channels: dict[str, _ChannelDb] + def needs_arg(self, key: str) -> bool: + 'Reply if attribute of key may reasonably be addressed without an arg.' + return not isinstance(getattr(self, key), (bool, int, str, tuple)) + def del_chan(self, name: str) -> None: 'Remove DB for channel of name.' del self._channels[name] - self._on_update(f'{name} {CLEAR_WORD}') + self._on_update(name) + + def has_chan(self, name: str) -> bool: + 'Test if entry of name in channels dictionary.' + return name in self._channels def chan(self, name: str) -> _ChannelDb: 'Produce DB for channel of name – pre-existing, or newly created.' if name not in self._channels: self._channels[name] = _ChannelDb( - on_update=lambda k: self._on_update(f'c {name} {k}')) + on_update=lambda k: self._on_update(name, k)) return self._channels[name] @@ -399,7 +407,7 @@ class Client(ABC, ClientQueueMixin): Thread(target=connect, daemon=True, args=(self,)).start() @abstractmethod - def _on_update(self, key: str) -> None: + def _on_update(self, path: str, arg: str = '') -> None: pass def _on_connect(self) -> None: diff --git a/ircplom/client_tui.py b/ircplom/client_tui.py index 1e55a60..0d86c95 100644 --- a/ircplom/client_tui.py +++ b/ircplom/client_tui.py @@ -8,7 +8,7 @@ from ircplom.tui_base import (BaseTui, PromptWidget, TuiEvent, Window, CMD_SHORTCUTS) from ircplom.irc_conn import IrcMessage from ircplom.client import (Client, ClientQueueMixin, Db, IrcConnSetup, - LogScope, NewClientEvent, CLEAR_WORD) + LogScope, NewClientEvent) CMD_SHORTCUTS['disconnect'] = 'window.disconnect' CMD_SHORTCUTS['join'] = 'window.join' @@ -121,24 +121,47 @@ class _ChannelWindow(_ChatWindow): self._send_msg('PART', (self.chatname,)) +@dataclass +class _Update: + path: str + arg: str = '' + value: Optional[_DbType] = None + + @property + def is_chan(self) -> bool: + 'Return if .path points to a _ChannelDb.' + return self.path[0] == '#' + + class _Db(Db): - def set_and_check_for_change(self, key: str, value: _DbType) -> bool: - 'To attribute of key set value, reply if that changed anything.' - self._typecheck(key, value) - old_value = getattr(self, key) - setattr(self, key, value) - return value != old_value + def _set_and_check_for_dict(self, update: _Update) -> bool: + d = getattr(self, update.path) + if update.value is None: + if update.arg == '': + d.clear() + else: + del d[update.arg] + return True + old_value = d.get(update.arg, None) + d[update.arg] = update.value + return update.value != old_value + + def set_and_check_for_change(self, update: _Update) -> bool: + 'Apply update, return if that actually made a difference.' + self._typecheck(update.path, update.value) + old_value = getattr(self, update.path) + setattr(self, update.path, update.value) + return update.value != old_value class _ChannelDb(_Db): users: tuple[str, ...] - def set_and_check_for_change(self, key: str, value: _DbType) -> bool: - if key == CLEAR_WORD: - self._set_empty_defaults() - return True - return super().set_and_check_for_change(key, value) + def set_and_check_for_change(self, update: _Update) -> bool: + if isinstance(getattr(self, update.path), dict): + return self._set_and_check_for_dict(update) + return super().set_and_check_for_change(update) class _TuiClientDb(_Db, IrcConnSetup): @@ -151,20 +174,18 @@ class _TuiClientDb(_Db, IrcConnSetup): user_modes: str _channels: dict[str, _ChannelDb] - def set_and_check_for_change(self, key: str, value: _DbType) -> bool: - if ' ' in key: - _, dict_name, arg = key.split(' ') - d = getattr(self, dict_name) - if arg == CLEAR_WORD: - d.clear() - elif arg[0] == '-': - del d[arg[1:]] - else: - old_value = d.get(arg, None) - d[arg] = value - return value != old_value - return True - return super().set_and_check_for_change(key, value) + def set_and_check_for_change(self, update: _Update) -> bool: + if update.is_chan: + chan_name = update.path + if update.value is None and not update.arg: + del self._channels[chan_name] + return True + update.path = update.arg + update.arg = '' + return self.chan(chan_name).set_and_check_for_change(update) + if isinstance(getattr(self, update.path), dict): + return self._set_and_check_for_dict(update) + return super().set_and_check_for_change(update) def chan(self, name: str) -> _ChannelDb: 'Produce DB for channel of name – pre-existing, or newly created.' @@ -220,41 +241,25 @@ class _ClientWindowsManager: prefix = f'{first_char}{sender_label}' self._tui_log(msg, scope=scope, prefix=prefix, **kwargs) - def update_db(self, key: str, value: _DbType) -> bool: - 'Ensure key at value, follow representation update triggers.' - db: _TuiClientDb | _ChannelDb = self._db - scope = LogScope.SERVER - log_kwargs: dict[str, str] = {} - verb = 'changed to:' - what = key - if ' ' in key: - type_char, parent_name, subkey = key.split() - if type_char == 'c': - db = self._db.chan(parent_name) - scope = LogScope.CHAT - log_kwargs |= {'channel': parent_name} - if subkey == CLEAR_WORD: - verb = 'cleared' - what = parent_name - elif type_char == 'c': - key = what = subkey - else: - if subkey[0] == '-': - verb = 'unset' - subkey = subkey[1:] - what = f'{parent_name}:{subkey}' - elif key == 'connection_state': - scope = LogScope.ALL - if not db.set_and_check_for_change(key, value): + def update_db(self, update: _Update) -> bool: + 'Apply update to ._db, and if changing anything, log and trigger.' + scope = (LogScope.CHAT if update.is_chan + else (LogScope.ALL if update.path == 'connection_state' + else LogScope.SERVER)) + verb = 'cleared' if update.value is None else 'changed to:' + what = f'{update.path}:{update.arg}' if update.arg else update.path + log_kwargs = {'channel': update.path} if update.is_chan else {} + if not self._db.set_and_check_for_change(update): return False announcement = f'{what} {verb}' - if isinstance(value, tuple) or announcement[-1] != ':': + if isinstance(update.value, tuple) or announcement[-1] != ':': self.log(announcement, scope=scope, **log_kwargs) - if isinstance(value, tuple): - for item in value: + if isinstance(update.value, tuple): + for item in update.value: self.log(f' {item}', scope=scope, **log_kwargs) elif announcement[-1] == ':': - self.log(f'{announcement} [{value}]', scope=scope, **log_kwargs) + self.log(f'{announcement} [{update.value}]', + scope=scope, **log_kwargs) for win in [w for w in self.windows if isinstance(w, _ChatWindow)]: win.set_prompt_prefix() return bool([w for w in self.windows if w.tainted]) @@ -347,9 +352,10 @@ class _ClientKnowingTui(Client): with open(f'{self.client_id}.log', 'a', encoding='utf8') as f: f.write(('>' if kwargs['out'] else '<') + f' {msg}\n') - def _on_update(self, key: str) -> None: - value: _DbType - if key == 'caps': + def _on_update(self, path: str, arg: str = '') -> None: + value: Optional[_DbType] = None + is_chan = path[0] == '#' + if path == 'caps': lines: list[str] = [] for cap_name, cap_entry in self._caps.as_caps.items(): line = '[*]' if cap_entry.enabled else '[ ]' @@ -358,14 +364,15 @@ class _ClientKnowingTui(Client): line += f' ({cap_entry.data})' lines += [line] value = tuple(lines) - elif ' ' in key: - type_char, parent_name, arg = key.split() - if arg == CLEAR_WORD or arg[0] == '-': - value = '' - elif type_char == 'c': - value = getattr(self._db.chan(parent_name), arg) + elif arg: + if is_chan and self._db.has_chan(path): + is_chan = True + if (chan := self._db.chan(path)) and hasattr(chan, arg): + value = getattr(chan, arg) else: - value = getattr(self._db, parent_name)[arg] - else: - value = getattr(self._db, key) - self._client_tui_trigger('update_db', key=key, value=value) + d = getattr(self._db, path) + if d.has(arg): + value = d[arg] + elif (not is_chan) and not self._db.needs_arg(path): + value = getattr(self._db, path) + self._client_tui_trigger('update_db', update=_Update(path, arg, value)) -- 2.30.2