'TUI adaptions to Client.'
# built-ins
from enum import Enum, auto
-from getpass import getuser
from pathlib import Path
from tomllib import load as toml_load
from typing import Any, Callable, Optional, Sequence
return bool([w for w in self.windows if w.tainted])
+class ClientKnowingTui(Client):
+ 'Adapted to communicate with ClientTui.'
+
+ def _tui_trigger(self, method_name: str, **kwargs) -> None:
+ self._put(TuiEvent.affector(method_name).kw(**kwargs))
+
+ def _tui_alert_trigger(self, msg: str) -> None:
+ self._tui_trigger('log', msg=msg, prefix=_LOG_PREFIX_SERVER,
+ alert=True)
+
+ def _client_tui_trigger(self, todo: str, **kwargs) -> None:
+ self._tui_trigger('for_client_do', client_id=self.client_id,
+ todo=todo, **kwargs)
+
+ def send_w_params_tuple(self, verb: str, params: tuple[str, ...]) -> None:
+ 'Helper for ClientWindow to trigger .send, for it can only do kwargs.'
+ self.send(verb, *params)
+
+ def privmsg(self, chat_target: str, msg: str) -> None:
+ 'Catch /privmsg, only allow for channel if in channel, else complain.'
+ try:
+ if self.db.is_chan_name(chat_target)\
+ and chat_target not in self.db.channels.keys():
+ raise SendFail('not sending, since not in channel')
+ self.send('PRIVMSG', chat_target, msg)
+ except SendFail as e:
+ self._tui_alert_trigger(f'{e}')
+ else:
+ self.db.messaging('').to(chat_target).privmsg = msg # type: ignore
+
+ def reconnect(self) -> None:
+ 'Catch /reconnect, only initiate if not connected, else complain back.'
+ if self.conn:
+ self._tui_alert_trigger(
+ 'not re-connecting since already connected')
+ return
+ self.connect()
+
+ def send(self, verb: str, *args) -> IrcMessage:
+ msg = super().send(verb, *args)
+ self._log(msg.raw, out=True)
+ return msg
+
+ def handle_msg(self, msg: IrcMessage) -> None:
+ self._log(msg.raw, out=False)
+ try:
+ super().handle_msg(msg)
+ except ImplementationFail as e:
+ self._alert(str(e))
+ except TargetUserOffline as e:
+ name = f'{e}'
+ self._log(f'{name} not online', target=name, alert=True)
+
+ def _alert(self, msg: str) -> None:
+ self._log(msg, alert=True)
+
+ def _log(self, msg: str, alert=False, target='', out: Optional[bool] = None
+ ) -> None:
+ scope = _LogScope.CHAT if target else _LogScope.DEBUG
+ self._client_tui_trigger('log', scope=scope, msg=msg, alert=alert,
+ target=target, out=out)
+
+ def _on_update(self, *path) -> None:
+ for path, value in self.db.into_endnode_updates(path):
+ self._client_tui_trigger('update_db', update=_Update(path, value))
+
+
class ClientTui(BaseTui):
'TUI expanded towards Client features.'
+ _client_cls = ClientKnowingTui
_path_config: Optional[Path] = _PATH_CONFIG
def __init__(self, **kwargs) -> None:
password = server_conf.get('password', '')
username = server_conf.get('username', '')
realname = server_conf.get('realname', '')
- setup = IrcConnSetup(
- hostname, port, nickname, username, realname, password)
- self._put(NewClientEvent(self._new_client(
- setup, set(server_conf.get('channels', [])))))
+ self._new_client(channels=set(server_conf.get('channels', [])),
+ conn_setup=IrcConnSetup(
+ hostname, port, nickname, username, realname,
+ password))
self.redraw_affected()
def _log_target_wins(self, **kwargs) -> Sequence[Window]:
if getattr(self._client_mngrs[client_id], todo)(**kwargs) is not False:
self.redraw_affected()
- def _new_client(self, conn_setup: IrcConnSetup, channels: set[str]
- ) -> 'ClientKnowingTui':
- return ClientKnowingTui(
+ def _new_client(self, conn_setup: IrcConnSetup, channels: set[str]):
+ client = self._client_cls(
_q_out=self._q_out, conn_setup=conn_setup, channels=channels)
+ self._put(NewClientEvent(client))
+ return client
def cmd__connect(self,
host_port: str,
return f'invalid port number: {port_str}'
nickname, password = from_split(nickname_pw)
realname, username = from_split(realname_username)
- setup = IrcConnSetup(
- hostname, port, nickname, username, realname, password)
- self._put(NewClientEvent(self._new_client(setup, set())))
+ self._new_client(channels=set(), conn_setup=IrcConnSetup(
+ hostname, port, nickname, username, realname, password))
return None
-
-
-class ClientKnowingTui(Client):
- 'Adapted to communicate with ClientTui.'
-
- def _tui_trigger(self, method_name: str, **kwargs) -> None:
- self._put(TuiEvent.affector(method_name).kw(**kwargs))
-
- def _tui_alert_trigger(self, msg: str) -> None:
- self._tui_trigger('log', msg=msg, prefix=_LOG_PREFIX_SERVER,
- alert=True)
-
- def _client_tui_trigger(self, todo: str, **kwargs) -> None:
- self._tui_trigger('for_client_do', client_id=self.client_id,
- todo=todo, **kwargs)
-
- def send_w_params_tuple(self, verb: str, params: tuple[str, ...]) -> None:
- 'Helper for ClientWindow to trigger .send, for it can only do kwargs.'
- self.send(verb, *params)
-
- def privmsg(self, chat_target: str, msg: str) -> None:
- 'Catch /privmsg, only allow for channel if in channel, else complain.'
- try:
- if self.db.is_chan_name(chat_target)\
- and chat_target not in self.db.channels.keys():
- raise SendFail('not sending, since not in channel')
- self.send('PRIVMSG', chat_target, msg)
- except SendFail as e:
- self._tui_alert_trigger(f'{e}')
- else:
- self.db.messaging('').to(chat_target).privmsg = msg # type: ignore
-
- def reconnect(self) -> None:
- 'Catch /reconnect, only initiate if not connected, else complain back.'
- if self.conn:
- self._tui_alert_trigger(
- 'not re-connecting since already connected')
- return
- self.connect()
-
- def send(self, verb: str, *args) -> IrcMessage:
- msg = super().send(verb, *args)
- self._log(msg.raw, out=True)
- return msg
-
- def handle_msg(self, msg: IrcMessage) -> None:
- self._log(msg.raw, out=False)
- try:
- super().handle_msg(msg)
- except ImplementationFail as e:
- self._alert(str(e))
- except TargetUserOffline as e:
- name = f'{e}'
- self._log(f'{name} not online', target=name, alert=True)
-
- def _alert(self, msg: str) -> None:
- self._log(msg, alert=True)
-
- def _log(self, msg: str, alert=False, target='', out: Optional[bool] = None
- ) -> None:
- scope = _LogScope.CHAT if target else _LogScope.DEBUG
- self._client_tui_trigger('log', scope=scope, msg=msg, alert=alert,
- target=target, out=out)
-
- def _on_update(self, *path) -> None:
- for path, value in self.db.into_endnode_updates(path):
- self._client_tui_trigger('update_db', update=_Update(path, value))
projects / ircplom / commitdiff