self._still_on_init = True
self._on_update = on_update
super().__init__(**kwargs)
+ for key in [k for k in self._types if self._types[k] is _UpdatingDict]:
+ getattr(self, key).set_on_update(key, self._on_update)
self._still_on_init = False
def __setattr__(self, key: str, value) -> None:
# channel_modes: str
+class _UpdatingDict:
+ _on_update: Callable
+
+ def __init__(self) -> None:
+ 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}')
+
+ def clear(self) -> None:
+ 'Zero dict and send CLEAR_WORD update.'
+ self._dict.clear()
+ self._on_update(CLEAR_WORD)
+
+ def __getitem__(self, key: str) -> str:
+ return self._dict[key]
+
+ def __setitem__(self, key: str, val: str) -> None:
+ self._dict[key] = val
+ self._on_update(key)
+
+
class _ClientDb(_Db, IrcConnSetup):
connection_state: str
client_host: str
+ isupports: _UpdatingDict
nickname_confirmed: bool
user_modes: str
_completable_motd: _CompletableStringsList
- _completable_isupports: _CompletableStringsList
_channels: dict[str, _ChannelDb]
def del_chan(self, name: str) -> None:
'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'{name} {k}'))
+ on_update=lambda k: self._on_update(f'c {name} {k}'))
return self._channels[name]
def __init__(self, conn_setup: IrcConnSetup, **kwargs) -> None:
self.client_id = conn_setup.hostname
super().__init__(client_id=self.client_id, **kwargs)
- self._caps = _CapsManager(self.send, self._on_update)
self._db = _ClientDb(on_update=self._on_update)
+ self._caps = _CapsManager(self.send, self._on_update)
for k in conn_setup.__annotations__:
setattr(self._db, k, getattr(conn_setup, k))
if self._db.port <= 0:
if self.conn:
self.conn.close()
self.conn = None
+ self._db.isupports.clear()
self._db.nickname_confirmed = False
def on_handled_loop_exception(self, e: IrcConnAbortException) -> None:
def handle_msg(self, msg: IrcMessage) -> None:
'Log msg.raw, then process incoming msg into appropriate client steps.'
self._log(msg.raw, scope=LogScope.RAW, out=False)
- if msg.verb != self._prev_verb and self._prev_verb == '005':
- self._db.declare_complete('isupports')
self._prev_verb = msg.verb
if _NumericsToConfirmNickname.contain(msg.verb):
self._db.nickname = msg.params[0]
match msg.verb:
case '005' if len(msg.params) > 2:
for param in msg.params[1:-1]:
- self._db.append_completable('isupports', param)
+ toks = param.split('=', maxsplit=1)
+ self._db.isupports[toks[0]] = (toks[1] if len(toks) > 1
+ else '')
case '353' if len(msg.params) == 4:
for user in msg.params[3].split():
self._db.chan(msg.params[2]
caps: tuple[str]
client_host: str
connection_state: str
- isupports: tuple[str]
+ isupports: dict[str, str]
motd: tuple[str]
nickname_confirmed: bool
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()
+ return True
+ old_value = d.get(arg, None)
+ d[arg] = value
+ return value != old_value
+ return super().set_and_check_for_change(key, value)
+
def chan(self, name: str) -> _ChannelDb:
'Produce DB for channel of name – pre-existing, or newly created.'
if name not in self._channels:
db: _TuiClientDb | _ChannelDb = self._db
scope = LogScope.SERVER
log_kwargs: dict[str, str] = {}
+ clearing = False
+ what = key
if ' ' in key:
- chan_name, key = key.split()
- db = self._db.chan(chan_name)
- scope = LogScope.CHAT
- log_kwargs |= {'channel': chan_name}
+ type_char, parent_name, subkey = key.split()
+ clearing = subkey == CLEAR_WORD
+ if type_char == 'c':
+ db = self._db.chan(parent_name)
+ scope = LogScope.CHAT
+ log_kwargs |= {'channel': parent_name}
+ key = what = subkey
+ else:
+ what = f'{parent_name}:{subkey}'
elif key == 'connection_state':
scope = LogScope.ALL
if not db.set_and_check_for_change(key, value):
return False
- if key != CLEAR_WORD:
- announcement = f'changed {key} to:'
+ if not clearing:
+ announcement = f'changed {what} to:'
if isinstance(value, tuple):
self.log(announcement, scope=scope, **log_kwargs)
for item in value:
lines += [line]
value = tuple(lines)
elif ' ' in key:
- chan_name, arg = key.split()
- value = ('' if arg == CLEAR_WORD
- else getattr(self._db.chan(chan_name), arg))
+ type_char, parent_name, arg = key.split()
+ if arg == CLEAR_WORD:
+ value = ''
+ elif type_char == 'c':
+ value = getattr(self._db.chan(parent_name), 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)