From: Christian Heller Date: Mon, 9 Jun 2025 14:00:57 +0000 (+0200) Subject: Replace Event.type_ strings with Enums. X-Git-Url: https://plomlompom.com/repos/%22https:/validator.w3.org/static/blog?a=commitdiff_plain;h=7e0126e6991c8c846765ef34220c80dd8c56ac7a;p=ircplom Replace Event.type_ strings with Enums. --- diff --git a/ircplom.py b/ircplom.py index a61a470..5472072 100755 --- a/ircplom.py +++ b/ircplom.py @@ -4,6 +4,7 @@ from abc import ABC, abstractmethod from base64 import b64decode from contextlib import contextmanager +from enum import Enum, auto from inspect import _empty as inspect_empty, signature, stack from queue import SimpleQueue, Empty as QueueEmpty from signal import SIGWINCH, signal @@ -46,16 +47,34 @@ IRCSPEC_TAG_ESCAPES = ((r'\:', ';'), (r'\\', '\\')) +class EventType(Enum): + 'Differentiate Events for different treatment.' + ALERT = auto() + CONN_ALERT = auto() + CONNECTED = auto() + CONN_WINDOW = auto() + EXCEPTION = auto() + INIT_CONNECT = auto() + INIT_RECONNECT = auto() + KEYBINDING = auto() + PING = auto() + PROMPT_ADD = auto() + QUIT = auto() + RECV = auto() + SEND = auto() + SET_SCREEN = auto() + + class Event(NamedTuple): 'Communication unit between threads.' - type_: str + type_: EventType payload: Any = None class EventQueue(SimpleQueue): 'SimpleQueue wrapper optimized for handling Events.' - def eput(self, type_: str, payload: Any = None) -> None: + def eput(self, type_: EventType, payload: Any = None) -> None: 'Construct Event(type_, payload) and .put it onto queue.' self.put(Event(type_, payload)) @@ -82,7 +101,7 @@ class Terminal: @contextmanager def context(self, q_to_main: EventQueue) -> Generator: 'Combine multiple contexts into one.' - signal(SIGWINCH, lambda *_: q_to_main.eput('SET_SCREEN')) + signal(SIGWINCH, lambda *_: q_to_main.eput(EventType.SET_SCREEN)) self._blessed = BlessedTerminal() with (self._blessed.raw(), self._blessed.fullscreen(), @@ -201,7 +220,7 @@ class IrcConnection: self._socket: Optional[socket] = None self._assumed_open = False self._recv_loop: Optional[SocketRecvLoop] = None - self._broadcast('CONNECTION_WINDOW', self._idx) + self._broadcast(EventType.CONN_WINDOW, self._idx) self._start_connecting() def _start_connecting(self) -> None: @@ -234,7 +253,7 @@ class IrcConnection: self._socket.close() self._socket = None - def _broadcast(self, type_: str, payload: Any = None) -> 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)) @@ -272,31 +291,34 @@ 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('CONN_ALERT', + self._broadcast(EventType.CONN_ALERT, 'cannot send, assuming connection closed') return self._socket.sendall(line.encode('utf-8') + IRCSPEC_LINE_SEPARATOR) def handle(self, event: Event) -> None: 'Process connection-directed Event into further steps.' - if event.type_ == 'CONNECTED': - self._broadcast('SEND', IrcMessage('USER', [self._login[0], '0', - '*', self._login[2]])) - self._broadcast('SEND', IrcMessage('NICK', [self._login[1]])) + if event.type_ == EventType.CONNECTED: + self._broadcast(EventType.SEND, + IrcMessage('USER', [self._login[0], '0', '*', + self._login[2]])) + self._broadcast(EventType.SEND, + IrcMessage('NICK', [self._login[1]])) return - if event.type_ == 'INIT_RECONNECTION': + if event.type_ == EventType.INIT_RECONNECT: if self._assumed_open: - self._broadcast('CONN_ALERT', 'Reconnect called, but still ' - 'seem connected, so nothing to do.') + self._broadcast(EventType.CONN_ALERT, + 'Reconnect called, but still seem connected, ' + 'so nothing to do.') else: self._start_connecting() return msg: IrcMessage = event.payload[1] - if event.type_ == 'SEND': + if event.type_ == EventType.SEND: self._write_line(msg.raw) - elif event.type_ == 'RECV': - if msg.verb == 'PING': - self._broadcast('SEND', + elif event.type_ == EventType.RECV: + if msg.verb == EventType.PING: + self._broadcast(EventType.SEND, IrcMessage('PONG', [msg.parameters[0]])) elif msg.verb == 'ERROR'\ and msg.parameters[0].startswith('Closing link:'): @@ -421,7 +443,7 @@ class Loop: def stop(self) -> None: 'Emit "QUIT" signal to break threaded loop, then wait for break.' - self._q_input.eput('QUIT') + self._q_input.eput(EventType.QUIT) self._thread.join() def __enter__(self) -> Self: @@ -435,13 +457,13 @@ class Loop: 'Send event into thread loop.' self._q_input.put(event) - def broadcast(self, type_: str, payload: Any = None) -> None: + 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_ == 'QUIT': + if event.type_ == EventType.QUIT: return False return True @@ -469,7 +491,7 @@ class Loop: if yield_bonus: self.process_bonus(yield_bonus) except Exception as e: # pylint: disable=broad-exception-caught - self._q_to_main.eput('EXCEPTION', e) + self._q_to_main.eput(EventType.EXCEPTION, e) class Widget(ABC): @@ -671,7 +693,8 @@ class Window(Widget): def cmd__paste(self) -> None: 'Write OSC 52 ? sequence to get encoded clipboard paste into stdin.' - self._term.write(f'\033{OSC52_PREFIX}?\007', self._y_status) + self._term.write(f'\033{OSC52_PREFIX}?\007', + self._y_status) self.draw() @@ -679,7 +702,7 @@ class ConnectionWindow(Window): 'Window with attributes and methods for dealing with an IrcConnection.' def __init__(self, - broadcast: Callable[[str, Any], None], + broadcast: Callable[[EventType, Any], None], conn_idx: int, *args, **kwargs ) -> None: @@ -689,12 +712,12 @@ class ConnectionWindow(Window): def cmd__disconnect(self, quit_msg: str = 'ircplom says bye') -> None: 'Send QUIT command to server.' - self._broadcast('SEND', + self._broadcast(EventType.SEND, (self._conn_idx, IrcMessage('QUIT', [quit_msg]))) def cmd__reconnect(self) -> None: 'Attempt reconnection.' - self._broadcast('INIT_RECONNECTION', (self._conn_idx,)) + self._broadcast(EventType.INIT_RECONNECT, (self._conn_idx,)) class TuiLoop(Loop): @@ -706,7 +729,7 @@ class TuiLoop(Loop): self._window_idx = 0 self._conn_windows: list[Window] = [] super().__init__(*args, **kwargs) - self.put(Event('SET_SCREEN')) + self.put(Event(EventType.SET_SCREEN)) def _cmd_name_to_cmd(self, cmd_name: str) -> Optional[Callable]: cmd_name = CMD_SHORTCUTS.get(cmd_name, cmd_name) @@ -727,37 +750,38 @@ class TuiLoop(Loop): def process_main(self, event: Event) -> bool: if not super().process_main(event): return False - if event.type_ == 'SET_SCREEN': + if event.type_ == EventType.SET_SCREEN: self._term.calc_geometry() for window in self._windows: window.set_geometry() self.window.draw() - elif event.type_ == 'CONNECTION_WINDOW': + elif event.type_ == EventType.CONN_WINDOW: conn_win = ConnectionWindow(self.broadcast, event.payload[0], len(self._windows), self._term) self._windows += [conn_win] self._conn_windows += [conn_win] self._switch_window(conn_win.idx) - elif event.type_ == 'ALERT': - self.window.log.append(f'{event.type_} {event.payload}') + elif event.type_ == EventType.ALERT: + self.window.log.append(f'ALERT {event.payload}') self.window.log.draw() - elif event.type_ in {'RECV', 'SEND', 'CONN_ALERT'}: + elif event.type_ in {EventType.RECV, EventType.SEND, + EventType.CONN_ALERT}: conn_win = self._conn_windows[event.payload[0]] - if event.type_ == 'CONN_ALERT': + if event.type_ == EventType.CONN_ALERT: msg = f'ALERT {event.payload[1]}' else: - msg = (('<-' if event.type_ == 'RECV' else '->') + msg = (('<-' if event.type_ == EventType.RECV else '->') + event.payload[1].raw) conn_win.log.append(msg) if conn_win == self.window: self.window.log.draw() - elif event.type_ == 'KEYBINDING': + elif event.type_ == EventType.KEYBINDING: cmd = self._cmd_name_to_cmd(event.payload[0]) assert cmd is not None cmd(*event.payload[1:]) - elif event.type_ == 'PROMPT_ADD': + elif event.type_ == EventType.PROMPT_ADD: self.window.prompt.append(event.payload) - # elif event.type_ == 'DEBUG': + # elif event.type_ == EventType.DEBUG: # from traceback import format_exception # for line in '\n'.join(format_exception(event.payload) # ).split('\n'): @@ -783,8 +807,8 @@ class TuiLoop(Loop): nickname: str, realname: str ) -> None: - 'Send INIT_CONNECTION command to main loop.' - self.broadcast('INIT_CONNECTION', + 'Send INIT_CONNECT command to main loop.' + self.broadcast(EventType.INIT_CONNECT, (hostname, (username, nickname, realname))) def cmd__prompt_enter(self) -> None: @@ -817,11 +841,11 @@ class TuiLoop(Loop): else: alert = 'not prefixed by /' if alert: - self.broadcast('ALERT', f'invalid prompt command: {alert}') + self.broadcast(EventType.ALERT, f'invalid prompt command: {alert}') def cmd__quit(self) -> None: 'Send QUIT to all threads.' - self.broadcast('QUIT') + self.broadcast(EventType.QUIT) def cmd__window(self, towards: str) -> Optional[str]: 'Switch window selection.' @@ -851,7 +875,8 @@ class SocketRecvLoop(Loop): super().__init__(*args, **kwargs) def process_bonus(self, yielded: str) -> None: - self.broadcast('RECV', (self._conn_idx, IrcMessage.from_raw(yielded))) + self.broadcast(EventType.RECV, (self._conn_idx, + IrcMessage.from_raw(yielded))) class KeyboardLoop(Loop): @@ -870,13 +895,14 @@ class KeyboardLoop(Loop): to_paste += ' ' else: to_paste += '#' - self.broadcast('PROMPT_ADD', to_paste) + self.broadcast(EventType.PROMPT_ADD, to_paste) elif yielded in KEYBINDINGS: - self.broadcast('KEYBINDING', KEYBINDINGS[yielded]) + self.broadcast(EventType.KEYBINDING, KEYBINDINGS[yielded]) elif len(yielded) == 1: - self.broadcast('PROMPT_ADD', yielded) + self.broadcast(EventType.PROMPT_ADD, yielded) else: - self.broadcast('ALERT', f'unknown keyboard input: {yielded}') + self.broadcast(EventType.ALERT, + 'unknown keyboard input: {yielded}') def run() -> None: @@ -888,15 +914,16 @@ def run() -> None: while True: event = q_to_main.get() term.tui.put(event) - if event.type_ == 'QUIT': + if event.type_ == EventType.QUIT: break - if event.type_ == 'EXCEPTION': + if event.type_ == EventType.EXCEPTION: raise event.payload - if event.type_ == 'INIT_CONNECTION': + if event.type_ == EventType.INIT_CONNECT: connections += [IrcConnection(q_to_main, len(connections), *event.payload)] elif event.type_ in { - 'CONNECTED', 'INIT_RECONNECTION', 'RECV', 'SEND'}: + EventType.CONNECTED, EventType.INIT_RECONNECT, + EventType.RECV, EventType.SEND}: connections[event.payload[0]].handle(event) finally: for conn in connections: