Terminal I/O

x84.terminal

Terminal handler for x/84

class x84.terminal.Terminal(kind, stream, rows, columns)[source]

Bases: blessed.terminal.Terminal

A thin wrapper over blessed.Terminal.

Class initializer.

cbreak(*args, **kwds)[source]

Dummy method yields nothing for blessed compatibility.

getch() → unicode[source]

Read and decode next byte from keyboard stream. May return u’’ if decoding is not yet complete, or completed unicode character. Should always return bytes when self.kbhit() returns True.

Implementors of input streams other than os.read() on the stdin fd should derive and override this method.

inkey(timeout=None, [esc_delay, [_intr_continue]]) -> Keypress()[source]

Receive next keystroke from keyboard (stdin), blocking until a keypress is received or timeout elapsed, if specified.

When used without the context manager cbreak, stdin remains line-buffered, and this function will block until return is pressed, even though only one unicode character is returned at a time..

The value returned is an instance of Keystroke, with properties is_sequence, and, when True, non-None values for attributes code and name. The value of code may be compared against attributes of this terminal beginning with KEY, such as KEY_ESCAPE.

To distinguish between KEY_ESCAPE, and sequences beginning with escape, the esc_delay specifies the amount of time after receiving the escape character (chr(27)) to seek for the completion of other application keys before returning KEY_ESCAPE.

Normally, when this function is interrupted by a signal, such as the installment of SIGWINCH, this function will ignore this interruption and continue to poll for input up to the timeout specified. If you’d rather this function return u'' early, specify a value of False for _intr_continue.

is_a_tty[source]

Dummy property always returns True.

kbhit([timeout=None]) → bool[source]

Returns True if a keypress has been detected on keyboard.

When timeout is 0, this call is non-blocking, Otherwise blocking until keypress is detected (default). When timeout is a positive number, returns after timeout seconds have elapsed.

If input is not a terminal, False is always returned.

raw(*args, **kwds)[source]

Dummy method yields nothing for blessed compatibility.

session[source]

Session associated with this terminal.

set_keyboard_decoder(encoding)[source]

Set or change incremental decoder for keyboard input.

class x84.terminal.TerminalProcess(client, sid, master_pipes)[source]

Bases: object

Class record for tracking “terminals”.

Probably of most interest, is that a TerminalProcess is an abstract association with a multiprocessing.Process sub-process, and its i/o queues (master_pipes).

This is not a really tty, or even a pseudo-tty (pty)! No termios, fnctl, or any terminal driver i/o is performed, it is all virtual.

An instance of this class is stored using register_tty() and removed by unregister_tty(), and discovered using get_terminals().

Class constructor.

x84.terminal.determine_encoding(env)[source]

Determine and return preferred encoding given session env.

x84.terminal.find_tty(client)[source]

Given a client, return a matching tty, or None if not registered.

x84.terminal.flush_queue(queue)[source]

Flush all data awaiting on the ipc queue.

Seeks any remaining events in queue, used before closing to prevent zombie processes with IPC waiting to be picked up.

x84.terminal.get_terminals()[source]

Returns a list of all terminals as tuples (session-id, ttys).

x84.terminal.init_term(writer, env)[source]

Determine the final TERM and encoding and return a Terminal.

curses is initialized using the value of ‘TERM’ of dictionary env, as well as a starting window size of ‘LINES’ and ‘COLUMNS’. If the terminal-type is of ‘ansi’ or ‘ansi-bbs’, then the cp437 encoding is assumed; otherwise ‘utf8’.

A blessed-abstracted curses terminal is returned.

x84.terminal.kill_session(client, reason='killed')[source]

Given a client, shutdown its socket and signal subprocess exit.

x84.terminal.on_naws(client)[source]

Callback for telnet NAWS negotiation.

On a Telnet NAWS sub-negotiation, check if client is yet registered in registry, and if so, send a ‘refresh’ event down the event queue.

This is ultimately handled by x84.bbs.session.Session.buffer_event().

x84.terminal.register_tty(tty)[source]

Register a TerminalProcess instance.

x84.terminal.spawn_client_session(client, matrix_kwargs=None)[source]

Spawn sub-process for connecting client.

Optional

x84.terminal.start_process(sid, env, CFG, child_pipes, kind, addrport, matrix_args=None, matrix_kwargs=None)[source]

A multiprocessing.Process target.

Parameters:
  • sid (str) – string describing session source (IP address & port).
  • env (dict) – dictionary of client environment variables (must contain at least 'TERM').
  • CFG (ConfigParser.ConfigParser) – bbs configuration
  • child_pipes (tuple) – tuple of (writer, reader) for engine IPC.
  • kind (str) – what kind of connection as string, 'telnet', 'ssh', etc.
  • addrport (tuple) – (client-ip, client-port) as string and integer.
  • matrix_args (tuple) – optional positional arguments to pass to matrix script.
  • matrix_kwargs (dict) – optional keyward arguments to pass to matrix script.
x84.terminal.translate_ttype(ttype)[source]

Return preferred terminal type given the session-negotiation ttype.

This provides a kind of coercion; we know some terminals, such as SyncTerm report a terminal type of ‘ansi’ – however, the author publishes a termcap database for ‘ansi-bbs’ which he instructs should be used! So an [system] configuration item of termcap-ansi may be set to 'ansi-bbs' to coerce such terminals for Syncterm-centric telnet servers – though I would not recommend it.

Furthermore, if the ttype is (literally) ‘unknown’, then a system-wide default terminal type may be returned, also by [system] configuration option termcap-unknown.

x84.terminal.unregister_tty(tty)[source]

Unregister a TerminalProcess instance.