github.com/lowobservable.oec
1
0
Fork 0
mirror of https://github.com/lowobservable/oec.git synced 2026-03-02 18:04:21 +00:00
Code Issues Projects Releases Wiki Activity

Separate responsibilities of controller and session

Browse Source
This commit is contained in:
Andrew Kay
2019-06-20 21:10:45 -05:00
parent 5cff18e924
commit f233cc41ba
3 changed files with 99 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
oec/controller.py
View File

@@ -4,16 +4,14 @@ oec.controller
"""
import time
import os
from select import select
import logging
from ptyprocess import PtyProcess
from coax import poll, poll_ack, read_terminal_id, read_extended_id, \
KeystrokePollResponse, ReceiveTimeout, ReceiveError, \
ProtocolError
from .terminal import Terminal
from .emulator import VT100Emulator
from .session import SessionDisconnectedError
from .vt100 import VT100Session
class Controller:
"""The controller."""
@@ -27,20 +25,16 @@ class Controller:
self.host_command = host_command
self.terminal = None
self.host_process = None
self.emulator = None
self.session = None
def run(self):
"""Run the controller."""
while self.running:
if self.host_process:
if self.session:
try:
if self.host_process in select([self.host_process], [], [], 0)[0]:
data = self.host_process.read()
self._handle_host_process_output(data)
except EOFError:
self._handle_host_process_terminated()
self.session.handle_host()
except SessionDisconnectedError:
self._handle_session_disconnected()
try:
poll_response = poll(self.interface, timeout=1)
@@ -93,11 +87,10 @@ class Controller:
# Show the attached indicator on the status line.
self.terminal.display.status_line.write_string(0, 'S')
# Start the process.
self.host_process = self._start_host_process()
# Start the session.
self.session = VT100Session(self.terminal, self.host_command)
# Initialize the emulator.
self.emulator = VT100Emulator(self.terminal, self.host_process)
self.session.start()
def _read_terminal_ids(self):
terminal_id = None
@@ -131,17 +124,11 @@ class Controller:
def _handle_terminal_detached(self):
self.logger.info('Terminal detached')
if self.host_process:
self.logger.debug('Terminating host process')
if not self.host_process.terminate(force=True):
self.logger.error('Unable to terminate host process')
else:
self.logger.debug('Host process terminated')
if self.session:
self.session.terminate()
self.terminal = None
self.host_process = None
self.emulator = None
self.session = None
def _handle_poll_response(self, poll_response):
if isinstance(poll_response, KeystrokePollResponse):
@@ -170,32 +157,10 @@ class Controller:
if not key:
return
if self.emulator:
self.emulator.handle_key(key, modifiers, scan_code)
if self.session:
self.session.handle_key(key, modifiers, scan_code)
def _start_host_process(self):
environment = os.environ.copy()
def _handle_session_disconnected(self):
self.logger.info('Session disconnected')
environment['TERM'] = 'vt100'
environment['LC_ALL'] = 'C'
process = PtyProcess.spawn(self.host_command, env=environment,
dimensions=self.terminal.display.dimensions)
return process
def _handle_host_process_output(self, data):
if self.logger.isEnabledFor(logging.DEBUG):
self.logger.debug(f'Output from host process: {data}')
if self.emulator:
self.emulator.handle_host_output(data)
def _handle_host_process_terminated(self):
self.logger.info('Host process terminated')
if self.host_process.isalive():
self.logger.error('Host process is reporting as alive')
self.host_process = None
self.emulator = None
self.session = None

15
oec/session.py Normal file
View File

@@ -0,0 +1,15 @@
class Session:
def start(self):
raise NotImplementedError
def terminate(self):
raise NotImplementedError
def handle_host(self):
raise NotImplementedError
def handle_key(self, key, keyboard_modifiers, scan_code):
raise NotImplementedError
class SessionDisconnectedError(Exception):
pass

90
oec/emulator.py → oec/vt100.py
View File

@@ -1,11 +1,15 @@
"""
oec.emulator
~~~~~~~~~~~~
oec.vt100
~~~~~~~~~
"""
import os
from select import select
import logging
from ptyprocess import PtyProcess
import pyte
from .session import Session, SessionDisconnectedError
from .display import encode_ascii_character
from .keyboard import Key, get_ascii_character_for_key
@@ -65,14 +69,15 @@ VT100_KEY_MAP_ALT = {
Key.NEWLINE: b'\n'
}
class VT100Emulator:
"""VT100 emulator."""
class VT100Session(Session):
"""VT100 session."""
def __init__(self, terminal, host):
def __init__(self, terminal, host_command):
self.logger = logging.getLogger(__name__)
self.terminal = terminal
self.host = host
self.host_command = host_command
self.host_process = None
# Initialize the VT100 screen.
(rows, columns) = self.terminal.display.dimensions
@@ -83,6 +88,10 @@ class VT100Emulator:
self.vt100_stream = pyte.ByteStream(self.vt100_screen)
def start(self):
# Start the host process.
self._start_host_process()
# Clear the screen.
self.terminal.display.clear_screen()
@@ -92,28 +101,30 @@ class VT100Emulator:
# Load the address counter.
self.terminal.display.load_address_counter(index=0)
def terminate(self):
if self.host_process:
self._terminate_host_process()
def handle_host(self):
try:
if self.host_process in select([self.host_process], [], [], 0)[0]:
data = self.host_process.read()
self._handle_host_output(data)
return True
return False
except EOFError:
self.host_process = None
raise SessionDisconnectedError
def handle_key(self, key, keyboard_modifiers, scan_code):
"""Handle a terminal keystroke."""
bytes_ = self._map_key(key, keyboard_modifiers)
if bytes_ is not None:
self.host.write(bytes_)
def handle_host_output(self, data):
"""Handle output from the host process."""
self.vt100_stream.feed(data)
self.update()
def update(self):
"""Update the terminal with dirty changes from the VT100 screen - clears
dirty lines after updating terminal.
"""
self._apply(self.vt100_screen)
self.vt100_screen.dirty.clear()
self._flush()
self.host_process.write(bytes_)
def _map_key(self, key, keyboard_modifiers):
if keyboard_modifiers.is_alt():
@@ -142,6 +153,37 @@ class VT100Emulator:
return None
def _start_host_process(self):
environment = os.environ.copy()
environment['TERM'] = 'vt100'
environment['LC_ALL'] = 'C'
self.host_process = PtyProcess.spawn(self.host_command, env=environment,
dimensions=self.terminal.display.dimensions)
def _terminate_host_process(self):
self.logger.debug('Terminating host process')
if not self.host_process.terminate(force=True):
self.logger.error('Unable to terminate host process')
else:
self.logger.debug('Host process terminated')
self.host_process = None
def _handle_host_output(self, data):
if self.logger.isEnabledFor(logging.DEBUG):
self.logger.debug(f'Host process output: {data}')
self.vt100_stream.feed(data)
self._apply(self.vt100_screen)
self.vt100_screen.dirty.clear()
self._flush()
def _apply(self, screen):
for row in screen.dirty:
row_buffer = screen.buffer[row]