from threading import Thread
from typing import Callable, Optional
# ourselves
-from ircplom.events import AffectiveEvent, ExceptionEvent, Logger, QueueMixin
+from ircplom.events import AffectiveEvent, ExceptionEvent, QueueMixin
from ircplom.irc_conn import (BaseIrcConnection, IrcConnAbortException,
IrcMessage, PORT_SSL)
self.conn_setup.port = PORT_SSL
self.client_id = self.conn_setup.hostname
super().__init__(client_id=self.client_id, **kwargs)
- self._log = Logger(self._do_log).add
self.update_login(nick_confirmed=False,
nickname=self.conn_setup.nickname)
self._start_connecting()
self.send(IrcMessage(verb='NICK', params=(self.conn_setup.nickname,)))
@abstractmethod
- def _do_log(self, msg: str, stream: str = STREAM_SERVER, **kwargs) -> None:
- 'Write msg into log of stream, whatever shape that may have.'
+ def _log(self, msg: str, stream: str = STREAM_SERVER, **kwargs) -> None:
+ pass
def send(self,
msg: IrcMessage,
def __init__(self, client_id: str, new_window: Callable, tui_log: Callable
) -> None:
- self.log = lambda msg, **kw:tui_log(msg=msg, client_id=client_id, **kw
- ) or True
+ self.log = lambda msg, **kw: tui_log(msg=msg, client_id=client_id, **kw
+ ) or True
self.windows: list[_ClientWindow] = []
self._tui_new_window = new_window
for stream in (STREAM_SERVER, STREAM_RAW):
'Forward todo to appropriate _ClientWindowsManager.'
if client_id not in self._client_mngrs:
self._client_mngrs[client_id] = _ClientWindowsManager(
- client_id=client_id, tui_log=self._targeted_log,
+ client_id=client_id, tui_log=self._log,
new_window=lambda cls, **kw: self._new_window(
cls, _q_out=self._q_out, client_id=client_id, **kw))
if getattr(self._client_mngrs[client_id], todo)(**kwargs):
return
self._start_connecting()
- def _do_log(self, msg: str, stream: str = STREAM_SERVER, **kwargs) -> None:
+ def _log(self, msg: str, stream: str = STREAM_SERVER, **kwargs) -> None:
to_log = []
if msg:
to_log += [msg]
from threading import Thread
from typing import Any, Callable, Iterator, Literal, Self
-_LOG_PREFIX_DEFAULT = '# '
-_LOG_PREFIX_ALERT = f'{_LOG_PREFIX_DEFAULT}ALERT: '
-
class Event: # pylint: disable=too-few-public-methods
'Communication unit between threads.'
self._put(it_yield)
except Exception as e: # pylint: disable=broad-exception-caught
self._put(ExceptionEvent(e))
-
-
-class Logger:
- 'Wrapper for logging tasks.'
-
- def __init__(self, to_call: Callable[[str], None]) -> None:
- self._to_call = to_call
-
- def add(self,
- msg: str,
- prefix: str = _LOG_PREFIX_DEFAULT,
- alert=False,
- **kwargs
- ) -> None:
- 'Add msg to log.'
- if alert:
- prefix = _LOG_PREFIX_ALERT + prefix
- time = str(datetime.now())[11:19]
- self._to_call(f'{time} {prefix}{msg}', **kwargs)
from abc import ABC, abstractmethod
from base64 import b64decode
from contextlib import contextmanager
+from datetime import datetime
from inspect import _empty as inspect_empty, signature, stack
from signal import SIGWINCH, signal
from typing import (Callable, Generator, Iterator, NamedTuple, Optional,
# requirements.txt
from blessed import Terminal as BlessedTerminal
# ourselves
-from ircplom.events import AffectiveEvent, Logger, Loop, QueueMixin, QuitEvent
+from ircplom.events import AffectiveEvent, Loop, QueueMixin, QuitEvent
+
+_LOG_PREFIX_DEFAULT = '# '
+_LOG_PREFIX_ALERT = f'{_LOG_PREFIX_DEFAULT}ALERT: '
_MIN_HEIGHT = 4
_MIN_WIDTH = 32
self._window_idx = 0
self._windows: list[Window] = []
self._new_window()
- self._log = Logger(self._targeted_log).add
self._set_screen()
signal(SIGWINCH, lambda *_: self._set_screen())
# separated to serve as hook for subclass window selection
return [self.window]
- def _targeted_log(self, msg: str, **kwargs) -> None:
+ def _log(self, msg: str, **kwargs) -> None:
+ prefix = kwargs.get('prefix', _LOG_PREFIX_DEFAULT)
+ if kwargs.get('alert', False):
+ prefix = _LOG_PREFIX_ALERT + prefix
+ msg = f'{str(datetime.now())[11:19]} {prefix}{msg}'
for win in self._log_target_wins(**kwargs):
win.history.append(msg)