From: Christian Heller Date: Sun, 15 Jun 2025 17:18:38 +0000 (+0200) Subject: Replace Event.type_ with class hierarchy. X-Git-Url: https://plomlompom.com/repos/%22https:/validator.w3.org/static/templates?a=commitdiff_plain;h=7dd4da608038b6ae14c155b6f40baf761a6e48a1;p=ircplom Replace Event.type_ with class hierarchy. --- diff --git a/ircplom.py b/ircplom.py index cec2ccb..95dc2fb 100755 --- a/ircplom.py +++ b/ircplom.py @@ -1,33 +1,28 @@ #!/usr/bin/env python3 'Attempt at an IRC client.' -from ircplom.events import EventQueue, EventType -from ircplom.irc_conn import IrcConnection +from ircplom.events import EventQueue, ExceptionEvent, QuitEvent +from ircplom.irc_conn import ConnEvent, InitConnectEvent, IrcConnection from ircplom.tui import Terminal def run() -> None: 'Main execution code / loop.' - q_to_main = EventQueue() + q_to_main: EventQueue = EventQueue() connections: list[IrcConnection] = [] try: with Terminal().context(q_to_main) as term: while True: event = q_to_main.get() term.tui.put(event) - if event.type_ == EventType.QUIT: + if isinstance(event, QuitEvent): break - if event.type_ == EventType.EXCEPTION: + if isinstance(event, ExceptionEvent): raise event.payload - if event.type_ == EventType.INIT_CONNECT: + if isinstance(event, InitConnectEvent): connections += [IrcConnection(q_to_main, len(connections), *event.payload)] - elif event.type_ in { - EventType.CONNECTED, - EventType.DISCONNECTED, - EventType.INIT_RECONNECT, - EventType.SEND, - }: - connections[event.payload[0]].handle(event) + elif isinstance(event, ConnEvent): + connections[event.conn_idx].handle(event) finally: for conn in connections: conn.close() diff --git a/ircplom/events.py b/ircplom/events.py index 364f551..bacdc07 100644 --- a/ircplom/events.py +++ b/ircplom/events.py @@ -1,58 +1,64 @@ 'Event system with event loop.' -from enum import Enum, auto -from queue import SimpleQueue, Empty as QueueEmpty +from queue import SimpleQueue as EventQueue, Empty as QueueEmpty from threading import Thread -from typing import Any, Iterator, Literal, NamedTuple, Optional, Self - - -class EventType(Enum): - 'Differentiate Events for different treatment.' - CONNECTED = auto() - CONN_WINDOW = auto() - DISCONNECTED = auto() - EXCEPTION = auto() - INIT_CONNECT = auto() - INIT_RECONNECT = auto() - LOG = auto() - LOG_CONN = auto() - NICK_SET = auto() - PING = auto() - QUIT = auto() - SEND = auto() - SET_SCREEN = auto() - TUI_CMD = auto() - - -class Event(NamedTuple): +from typing import Iterator, Literal, Optional, Self + + +class Event: 'Communication unit between threads.' - type_: EventType - payload: Any = None -class EventQueue(SimpleQueue): - 'SimpleQueue wrapper optimized for handling Events.' +class PayloadMixin: + 'To extend Event with .payload= passed as first argument.' + + def __init__(self, payload, **kwargs) -> None: + super().__init__(**kwargs) + self.payload = payload + - def eput(self, type_: EventType, payload: Any = None) -> None: - 'Construct Event(type_, payload) and .put it onto queue.' - self.put(Event(type_, payload)) +class ExceptionEvent(Event, PayloadMixin): + 'To signal Exception to be handled by receiver.' + payload: Exception -class Loop: - 'Wraps thread looping over .eput input queue, potential bonus iterator.' +class QuitEvent(Event): + 'To signal any receiver to exit.' + + +class BroadcastMixin: + 'To provide .broadcast via newly assigned ._q_to_main.' + + def __init__(self, q_to_main: EventQueue) -> None: + self._q_to_main = q_to_main + + def broadcast[E: Event](self, + event_class: type[E], + *args, **kwargs + ) -> None: + "Put event to main loop; if event_class PayloadMixin'd, payload=args." + if PayloadMixin in event_class.__mro__: + kwargs['payload'] = args if len(args) > 1 else args[0] + args = tuple() + self._q_to_main.put(event_class(*args, **kwargs)) + + +class Loop(BroadcastMixin): + 'Wraps thread looping over input queue, potential bonus iterator.' def __init__(self, q_to_main: EventQueue, - bonus_iterator: Optional[Iterator] = None + bonus_iterator: Optional[Iterator] = None, + **kwargs ) -> None: - self._q_to_main = q_to_main + super().__init__(q_to_main=q_to_main, **kwargs) self._bonus_iterator = bonus_iterator - self._q_input = EventQueue() + self._q_input: EventQueue = EventQueue() self._thread = Thread(target=self._loop, daemon=False) self._thread.start() def stop(self) -> None: - 'Emit "QUIT" signal to break threaded loop, then wait for break.' - self._q_input.eput(EventType.QUIT) + 'Emit QuitEvent to break threaded loop, then wait for break.' + self._q_input.put(QuitEvent()) self._thread.join() def __enter__(self) -> Self: @@ -66,13 +72,9 @@ class Loop: 'Send event into thread loop.' self._q_input.put(event) - def broadcast(self, type_: EventType, payload: Any = None) -> None: - 'Send event to main loop via queue.' - self._q_to_main.eput(type_, payload) - def process_main(self, event: Event) -> bool: 'Process event yielded from input queue.' - if event.type_ == EventType.QUIT: + if isinstance(event, QuitEvent): return False return True @@ -100,4 +102,4 @@ class Loop: if yield_bonus: self.process_bonus(yield_bonus) except Exception as e: # pylint: disable=broad-exception-caught - self._q_to_main.eput(EventType.EXCEPTION, e) + self._q_to_main.put(ExceptionEvent(e)) diff --git a/ircplom/irc_conn.py b/ircplom/irc_conn.py index 8982064..6746bc7 100644 --- a/ircplom/irc_conn.py +++ b/ircplom/irc_conn.py @@ -2,9 +2,10 @@ # built-ins from socket import socket, gaierror as socket_gaierror from threading import Thread -from typing import Any, Callable, Iterator, NamedTuple, Optional, Self +from typing import Callable, Iterator, NamedTuple, Optional, Self # ourselves -from ircplom.events import Event, EventQueue, EventType, Loop +from ircplom.events import (BroadcastMixin, Event, EventQueue, Loop, + PayloadMixin) TIMEOUT_LOOP = 0.1 @@ -28,7 +29,68 @@ class LoginNames(NamedTuple): real: str -class IrcConnection: +class InitConnectEvent(Event, PayloadMixin): + 'Event to trigger connection, with payload (host, LoginNames).' + payload: tuple[str, LoginNames] + + +class ConnEvent(Event): + 'Event with connection ID at .conn_idx' + + def __init__(self, conn_idx: int, **kwargs) -> None: + super().__init__(**kwargs) + self.conn_idx = conn_idx + + +class InitConnWindowEvent(ConnEvent, PayloadMixin): + 'Event to trigger TUI making ConnectionWindow, nick in prompt = payload.' + payload: str + + +class _ConnectedEvent(ConnEvent, PayloadMixin): + 'Event to signal opening of connection, with payload login names to send.' + payload: LoginNames + + +class DisconnectedEvent(ConnEvent): + 'Event to signal closing of connection' + + +class InitReconnectEvent(ConnEvent): + 'Event to trigger re-opening of connection.' + + +class LogConnEvent(ConnEvent, PayloadMixin): + 'Event to log payload into connection window.' + payload: str + + +class NickSetEvent(ConnEvent, PayloadMixin): + 'Event to signal nickname (= payload) having been set server-side.' + payload: str + + +class SendEvent(ConnEvent, PayloadMixin): + 'Event to trigger sending of payload to server.' + payload: 'IrcMessage' + + +class BroadcastConnMixin(BroadcastMixin): + 'Provides .broadcast_conn on classes that have .conn_idx defined.' + + def __init__(self, conn_idx: int, **kwargs): + super().__init__(**kwargs) + self.conn_idx = conn_idx + + def broadcast_conn[E: ConnEvent](self, + event_class: type[E], + *args, **kwargs + ) -> None: + 'Broadcast event subclassing ConnEvent, with connection ID.' + self.broadcast(event_class, conn_idx=self.conn_idx, *args, **kwargs) + + +class IrcConnection(BroadcastConnMixin): 'Abstracts socket connection, loop over it, and handling messages from it.' def __init__(self, @@ -37,33 +99,33 @@ class IrcConnection: hostname: str, login: LoginNames, ) -> None: - self._idx = idx - self._q_to_main = q_to_main + super().__init__(conn_idx=idx, q_to_main=q_to_main) self._hostname = hostname self._login = login self._socket: Optional[socket] = None self._assumed_open = False self._loop: Optional[_ConnectionLoop] = None - self._broadcast(EventType.CONN_WINDOW, self._login.nick) + self.broadcast_conn(InitConnWindowEvent, self._login.nick) self._start_connecting() def _start_connecting(self) -> None: def connect(self) -> None: self._socket = socket() - self._broadcast(EventType.LOG_CONN, - f'Connecting to {self._hostname} …') + self.broadcast_conn(LogConnEvent, + f'Connecting to {self._hostname} …') self._socket.settimeout(_TIMEOUT_CONNECT) try: self._socket.connect((self._hostname, _PORT)) except (TimeoutError, socket_gaierror) as e: - self._broadcast(EventType.LOG_CONN, f'ALERT: {e}') + self.broadcast_conn(LogConnEvent, f'ALERT: {e}') return self._socket.settimeout(TIMEOUT_LOOP) self._assumed_open = True - self._loop = _ConnectionLoop(self._idx, self._q_to_main, - self._read_lines()) - self._broadcast(EventType.CONNECTED, self._login) + self._loop = _ConnectionLoop(conn_idx=self.conn_idx, + q_to_main=self._q_to_main, + bonus_iterator=self._read_lines()) + self.broadcast_conn(_ConnectedEvent, self._login) Thread(target=connect, daemon=True, args=(self,)).start() @@ -77,10 +139,6 @@ class IrcConnection: self._socket.close() self._socket = None - def _broadcast(self, type_: EventType, payload: Any = None) -> None: - 'Send event to main loop via queue, with connection index as 1st arg.' - self._q_to_main.eput(type_, (self._idx, payload)) - def _read_lines(self) -> Iterator[Optional[str]]: 'Receive line-separator-delimited messages from socket.' assert self._socket is not None @@ -115,28 +173,28 @@ class IrcConnection: def _write_line(self, line: str) -> None: 'Send line-separator-delimited message over socket.' if not (self._socket and self._assumed_open): - self._broadcast(EventType.LOG_CONN, - 'ALERT: cannot send, assuming connection closed') + self.broadcast_conn(LogConnEvent, + 'ALERT: cannot send, assuming connection ' + 'closed') return self._socket.sendall(line.encode('utf-8') + _IRCSPEC_LINE_SEPARATOR) - def handle(self, event: Event) -> None: + def handle(self, event: ConnEvent) -> None: 'Process connection-directed Event into further steps.' - if event.type_ == EventType.INIT_RECONNECT: + if isinstance(event, InitReconnectEvent): if self._assumed_open: - self._broadcast(EventType.LOG_CONN, - 'ALERT: Reconnect called, but still seem ' - 'connected, so nothing to do.') + self.broadcast_conn(LogConnEvent, + 'ALERT: Reconnect called, but still seem ' + 'connected, so nothing to do.') else: self._start_connecting() - elif event.type_ == EventType.CONNECTED: + elif isinstance(event, _ConnectedEvent): assert self._loop is not None self._loop.put(event) - elif event.type_ == EventType.DISCONNECTED: + elif isinstance(event, DisconnectedEvent): self.close() - elif event.type_ == EventType.SEND: - msg: IrcMessage = event.payload[1] - self._write_line(msg.raw) + elif isinstance(event, SendEvent): + self._write_line(event.payload.raw) class IrcMessage: @@ -239,40 +297,33 @@ class IrcMessage: return self._raw -class _ConnectionLoop(Loop): +class _ConnectionLoop(Loop, BroadcastConnMixin): 'Loop receiving and translating socket messages towards main loop.' - def __init__(self, connection_idx: int, *args, **kwargs) -> None: - self._conn_idx = connection_idx - super().__init__(*args, **kwargs) - - def _broadcast_conn(self, type_: EventType, *args) -> None: - self.broadcast(type_, (self._conn_idx, *args)) - def _send(self, verb: str, parameters: tuple[str, ...]) -> None: msg = IrcMessage(verb, parameters) - self._broadcast_conn(EventType.LOG_CONN, f'->: {msg.raw}') - self._broadcast_conn(EventType.SEND, msg) + self.broadcast_conn(LogConnEvent, f'->: {msg.raw}') + self.broadcast_conn(SendEvent, msg) def process_main(self, event: Event) -> bool: if not super().process_main(event): return False - if event.type_ == EventType.CONNECTED: - login = event.payload[1] + if isinstance(event, _ConnectedEvent): # self._send('CAP', ('LS', '302')) - self._send('USER', (login.user, '0', '*', login.real)) - self._send('NICK', (login.nick,)) + self._send('USER', (event.payload.user, '0', '*', + event.payload.real)) + self._send('NICK', (event.payload.nick,)) # self._send('CAP', ('LIST',)) # self._send('CAP', ('END',)) return True def process_bonus(self, yielded: str) -> None: msg = IrcMessage.from_raw(yielded) - self._broadcast_conn(EventType.LOG_CONN, f'<-: {msg.raw}') + self.broadcast_conn(LogConnEvent, f'<-: {msg.raw}') if msg.verb == 'PING': self._send('PONG', (msg.parameters[0],)) elif msg.verb == 'ERROR'\ and msg.parameters[0].startswith('Closing link:'): - self._broadcast_conn(EventType.DISCONNECTED) + self.broadcast_conn(DisconnectedEvent) elif msg.verb == '001': - self._broadcast_conn(EventType.NICK_SET, msg.parameters[0]) + self.broadcast_conn(NickSetEvent, msg.parameters[0]) diff --git a/ircplom/tui.py b/ircplom/tui.py index 360d5a7..f36ff51 100644 --- a/ircplom/tui.py +++ b/ircplom/tui.py @@ -6,12 +6,16 @@ from contextlib import contextmanager from getpass import getuser as getusername from inspect import _empty as inspect_empty, signature, stack from signal import SIGWINCH, signal -from typing import Any, Callable, Generator, Iterator, NamedTuple, Optional +from typing import Callable, Generator, Iterator, NamedTuple, Optional # requirements.txt from blessed import Terminal as BlessedTerminal # ourselves -from ircplom.events import Event, EventType, EventQueue, Loop -from ircplom.irc_conn import IrcMessage, LoginNames, TIMEOUT_LOOP +from ircplom.events import (BroadcastMixin, Event, EventQueue, Loop, + PayloadMixin, QuitEvent) +from ircplom.irc_conn import ( + BroadcastConnMixin, ConnEvent, DisconnectedEvent, InitConnectEvent, + InitConnWindowEvent, InitReconnectEvent, IrcMessage, LoginNames, + LogConnEvent, NickSetEvent, SendEvent, TIMEOUT_LOOP) _B64_PREFIX = 'b64:' @@ -41,6 +45,20 @@ _CMD_SHORTCUTS = { } +class _LogEvent(Event, PayloadMixin): + 'Event to trigger writing to current Window\'s LogWidget.' + payload: str + + +class _SetScreenEvent(Event): + 'Event to trigger re-configuration of screen sizes.' + + +class _TuiCmdEvent(Event, PayloadMixin): + 'Event to trigger call of .cmd__ method in TUI tree.' + payload: str + + class _YX(NamedTuple): '2-dimensional coordinate.' y: int @@ -270,7 +288,8 @@ class _Window(_Widget): _y_status: int prompt: _PromptWidget - def __init__(self, idx: int, term: 'Terminal') -> None: + def __init__(self, idx: int, term: 'Terminal', **kwargs) -> None: + super().__init__(**kwargs) self.idx = idx self._term = term self.log = _LogWidget(self._term.wrap, self._term.write) @@ -299,30 +318,20 @@ class _Window(_Widget): self.draw() -class _ConnectionWindow(_Window): +class _ConnectionWindow(_Window, BroadcastConnMixin): 'Window with attributes and methods for dealing with an IrcConnection.' prompt: _ConnectionPromptWidget - def __init__(self, - broadcast: Callable[[EventType, Any], None], - conn_idx: int, - *args, **kwargs - ) -> None: - self._broadcast = broadcast - self._conn_idx = conn_idx - super().__init__(*args, **kwargs) - def cmd__disconnect(self, quit_msg: str = 'ircplom says bye') -> None: 'Send QUIT command to server.' - self._broadcast(EventType.SEND, - (self._conn_idx, IrcMessage('QUIT', (quit_msg, )))) + self.broadcast_conn(SendEvent, IrcMessage('QUIT', (quit_msg, ))) def cmd__reconnect(self) -> None: 'Attempt reconnection.' - self._broadcast(EventType.INIT_RECONNECT, (self._conn_idx,)) + self.broadcast_conn(InitReconnectEvent) -class _KeyboardLoop(Loop): +class _KeyboardLoop(Loop, BroadcastMixin): 'Loop receiving and translating keyboard events towards main loop.' def process_bonus(self, yielded: str) -> None: @@ -338,28 +347,26 @@ class _KeyboardLoop(Loop): to_paste += ' ' else: to_paste += '#' - self.broadcast(EventType.TUI_CMD, - ('window.prompt.append', to_paste)) + self.broadcast(_TuiCmdEvent, ('window.prompt.append', to_paste)) elif yielded in _KEYBINDINGS: - self.broadcast(EventType.TUI_CMD, _KEYBINDINGS[yielded]) + self.broadcast(_TuiCmdEvent, _KEYBINDINGS[yielded]) elif len(yielded) == 1: - self.broadcast(EventType.TUI_CMD, - ('window.prompt.append', yielded)) + self.broadcast(_TuiCmdEvent, ('window.prompt.append', yielded)) else: - self.broadcast(EventType.LOG, + self.broadcast(_LogEvent, f'ALERT: unknown keyboard input: {yielded}') -class _TuiLoop(Loop): +class _TuiLoop(Loop, BroadcastMixin): 'Loop for drawing/updating TUI.' - def __init__(self, term: 'Terminal', *args, **kwargs) -> None: + def __init__(self, term: 'Terminal', **kwargs) -> None: self._term = term self._windows = [_Window(0, self._term)] self._window_idx = 0 self._conn_windows: list[_ConnectionWindow] = [] - super().__init__(*args, **kwargs) - self.put(Event(EventType.SET_SCREEN)) + super().__init__(**kwargs) + self.put(_SetScreenEvent()) def _cmd_name_to_cmd(self, cmd_name: str) -> Optional[Callable]: cmd_name = _CMD_SHORTCUTS.get(cmd_name, cmd_name) @@ -380,50 +387,42 @@ class _TuiLoop(Loop): def process_main(self, event: Event) -> bool: if not super().process_main(event): return False - if event.type_ == EventType.SET_SCREEN: + if isinstance(event, _SetScreenEvent): self._term.calc_geometry() for window in self._windows: window.set_geometry() self.window.draw() - elif event.type_ == EventType.CONN_WINDOW: - conn_win = _ConnectionWindow(broadcast=self.broadcast, - conn_idx=event.payload[0], + elif isinstance(event, _LogEvent): + self.window.log.append(event.payload) + self.window.log.draw() + elif isinstance(event, _TuiCmdEvent): + cmd = self._cmd_name_to_cmd(event.payload[0]) + assert cmd is not None + cmd(*event.payload[1:]) + elif isinstance(event, InitConnWindowEvent): + conn_win = _ConnectionWindow(q_to_main=self._q_to_main, + conn_idx=event.conn_idx, idx=len(self._windows), term=self._term) conn_win.prompt.update_prompt(nick_confirmed=False, - nick=event.payload[1]) + nick=event.payload) self._windows += [conn_win] self._conn_windows += [conn_win] self._switch_window(conn_win.idx) - elif event.type_ == EventType.LOG: - self.window.log.append(event.payload) - self.window.log.draw() - elif event.type_ == EventType.LOG_CONN: - conn_win = self._conn_windows[event.payload[0]] - conn_win.log.append(event.payload[1]) + elif isinstance(event, ConnEvent): + conn_win = self._conn_windows[event.conn_idx] + if isinstance(event, LogConnEvent): + conn_win.log.append(event.payload) + elif isinstance(event, NickSetEvent): + conn_win.prompt.update_prompt(nick_confirmed=True, + nick=event.payload) + elif isinstance(event, DisconnectedEvent): + conn_win.prompt.update_prompt(nick_confirmed=False) if conn_win == self.window: - self.window.log.draw() - elif event.type_ == EventType.NICK_SET: - conn_win = self._conn_windows[event.payload[0]] - conn_win.prompt.update_prompt(nick_confirmed=True, - nick=event.payload[1]) - if conn_win == self.window: - self.window.prompt.draw() - elif event.type_ == EventType.DISCONNECTED: - conn_win = self._conn_windows[event.payload[0]] - conn_win.prompt.update_prompt(nick_confirmed=False) - if conn_win == self.window: - self.window.prompt.draw() - elif event.type_ == EventType.TUI_CMD: - cmd = self._cmd_name_to_cmd(event.payload[0]) - assert cmd is not None - cmd(*event.payload[1:]) - # elif event.type_ == EventType.DEBUG: - # from traceback import format_exception - # for line in '\n'.join(format_exception(event.payload) - # ).split('\n'): - # self.window.log.append(f'DEBUG {line}') - # self.window.log.draw() + if isinstance(event, LogConnEvent): + self.window.log.draw() + if isinstance(event, (NickSetEvent, DisconnectedEvent)): + self.window.prompt.draw() else: return True self._term.flush() @@ -440,9 +439,9 @@ class _TuiLoop(Loop): def cmd__connect(self, hostname: str, nickname: str, realname: str ) -> None: - 'Send INIT_CONNECT command to main loop.' + 'Broadcast InitConnectEvent.' login = LoginNames(user=getusername(), nick=nickname, real=realname) - self.broadcast(EventType.INIT_CONNECT, (hostname, login)) + self.broadcast(InitConnectEvent, (hostname, login)) def cmd__prompt_enter(self) -> None: 'Get prompt content from .window.prompt.enter, parse to & run command.' @@ -474,11 +473,11 @@ class _TuiLoop(Loop): else: alert = 'not prefixed by /' if alert: - self.broadcast(EventType.LOG, f'invalid prompt command: {alert}') + self.broadcast(_LogEvent, f'invalid prompt command: {alert}') def cmd__quit(self) -> None: 'Send QUIT to all threads.' - self.broadcast(EventType.QUIT) + self.broadcast(QuitEvent) def cmd__window(self, towards: str) -> Optional[str]: 'Switch window selection.' @@ -510,14 +509,14 @@ class Terminal: @contextmanager def context(self, q_to_main: EventQueue) -> Generator: 'Combine multiple contexts into one.' - signal(SIGWINCH, lambda *_: q_to_main.eput(EventType.SET_SCREEN)) + signal(SIGWINCH, lambda *_: q_to_main.put(_SetScreenEvent())) self._blessed = BlessedTerminal() with (self._blessed.raw(), self._blessed.fullscreen(), self._blessed.hidden_cursor(), _KeyboardLoop(q_to_main, self.get_keypresses())): self._cursor_yx = _YX(0, 0) - with _TuiLoop(self, q_to_main) as self.tui: + with _TuiLoop(self, q_to_main=q_to_main) as self.tui: yield self @property