#!/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()
'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:
'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
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))
# 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
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,
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()
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
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:
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])
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:'
}
+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
_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)
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:
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)
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()
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.'
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.'
@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
projects / ircplom / commitdiff