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
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]
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]
def __delitem__(self, key: str) -> None:
del self._dict[key]
- self._on_update(f'-{key}')
+ self._on_update(key)
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]
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:
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'
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):
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.'
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])
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 '[ ]'
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))
projects / ircplom / commitdiff