from getpass import getuser
from dataclasses import asdict as dc_asdict
from inspect import signature
-from typing import Callable, Optional
+from typing import Callable, Optional, Sequence
# ourselves
from ircplom.tui_base import (BaseTui, PromptWidget, TuiEvent, Window,
CMD_SHORTCUTS)
nick_confirmed: bool = False
nickname: str = '?'
- def __init__(self, new_window: Callable) -> None:
+ def __init__(self, client_id: str, new_window: Callable, tui_log: Callable
+ ) -> None:
+ self.log = lambda **kw: tui_log(client_id=client_id, **kw) or True
self.windows: list[_ClientWindow] = []
self._tui_new_window = new_window
for stream in (STREAM_SERVER, STREAM_RAW):
return win
return self._new_window(stream)
- def log(self, msg: str, stream: str) -> None:
- 'Log msg to stram.'
- if stream == STREAM_ALL:
- for win in self.windows:
- win.history.append(msg)
- else:
- self.window(stream).history.append(msg)
-
def update(self, **kwargs) -> bool:
'Apply settings in kwargs, follow represntation update triggres.'
for k, v in kwargs.items():
super().__init__(**kwargs)
self._client_mngrs: dict[str, _ClientWindowsManager] = {}
+ def _log_target_wins(self, **kwargs) -> Sequence[Window]:
+ stream = kwargs.get('stream', _STREAM_SAME)
+ if stream != _STREAM_SAME and 'client_id' in kwargs:
+ m = self._client_mngrs[kwargs['client_id']]
+ return m.windows if stream == STREAM_ALL else [m.window(stream)]
+ return super()._log_target_wins(**kwargs)
+
def for_client_do(self, client_id: str, todo: str, **kwargs) -> None:
- 'Forward todo to appropriate ClientTuiMgr.'
+ 'Forward todo to appropriate _ClientWindowManager.'
if client_id not in self._client_mngrs:
self._client_mngrs[client_id] = _ClientWindowsManager(
- new_window=lambda cls, **kwargs: self._new_window(
- cls, _q_out=self._q_out, client_id=client_id, **kwargs))
+ client_id=client_id, tui_log=self._targeted_log,
+ new_window=lambda cls, **kw: self._new_window(
+ cls, _q_out=self._q_out, client_id=client_id, **kw))
client_data = self._client_mngrs[client_id]
method = getattr(client_data, todo)
if method(**kwargs) or signature(method).return_annotation is None:
for k, v in dc_asdict(kwargs['conn_setup']).items():
to_log += [f' {k}: [{v}]']
for item in to_log:
- if stream == _STREAM_SAME:
- self._put(TuiEvent.affector('_log').kw(msg=item))
- self._put(TuiEvent.affector('redraw_affected'))
- else:
- self._client_tui_trigger('log', stream=stream, msg=item)
+ self._client_tui_trigger('log', stream=stream, msg=item)
def update_login(self, nick_confirmed: bool, nickname: str = '') -> None:
super().update_login(nick_confirmed, nickname)
from contextlib import contextmanager
from inspect import _empty as inspect_empty, signature, stack
from signal import SIGWINCH, signal
-from typing import Callable, Generator, Iterator, NamedTuple, Optional
+from typing import (Callable, Generator, Iterator, NamedTuple, Optional,
+ Sequence)
# requirements.txt
from blessed import Terminal as BlessedTerminal
# ourselves
self._window_idx = 0
self._windows: list[Window] = []
self._new_window()
- self._log = Logger(
- lambda msg: # pylint: disable=unnecessary-lambda # to keep up-to…
- self.window.history.append(msg) # …-date _what_…
- ).add # …window's .history
+ self._log = Logger(self._targeted_log).add
self._set_screen()
signal(SIGWINCH, lambda *_: self._set_screen())
+ def _log_target_wins(self, **_) -> Sequence[Window]:
+ # separated to serve as hook for subclass window selection
+ return [self.window]
+
+ def _targeted_log(self, msg: str, **kwargs) -> None:
+ for win in self._log_target_wins(**kwargs):
+ win.history.append(msg)
+
def _new_window(self, win_class=Window, **kwargs) -> Window:
new_idx = len(self._windows)
win = win_class(idx=new_idx, term=self._term, **kwargs)