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
(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))
@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(),
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:
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))
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:'):
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:
'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
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):
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()
'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:
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):
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)
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'):
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:
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.'
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):
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:
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:
projects / ircplom / commitdiff