# MIT License
#
# Copyright (c) 2023 Neil Webber
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
# simulation of a KL-11 console interface
#
# There are two modes of operation: socket or stdin.
#
# socket: In socket mode, this starts a simple TCP server and accepts
# connections on port 1170 (cute). Characters are proxied back
# and forth from this socket to the emulated console
#
# stdin: In stdin mode, python's stdin is usurped and put into raw mode.
# The python console is the emulated console. This mode
# requires termios ("unix only" but may exist on other systems)
#
# In this mode the normal ^C/SIGINT does nothing because in raw
# mode there are no special characters; everything is passed to
# the emulated environment. The only way to exit/stop the emulator is
# if the running code halts. Alternatively, the input byte sequence:
# 0xC3 0xA7
# which is Unicode U+00E7, c-cedilla, will cause the emulation to
# halt as if the physical console HALT toggle had been selected.
# This sequence was chosen because it is option-C on a mac; see
# HALT_SEQUENCE
# to override this choice. NOTE: the first N-1 HALT_SEQUENCE bytes
# will still be transmitted to the emulation, the HALT will only occur
# once the full sequence has been received.
#
import socket
import threading
import queue
from contextlib import contextmanager
# termios is only on unix-like/POSIX systems.
# If there is no termios module then use_stdin mode is not available.
try:
import termios
except ModuleNotFoundError:
termios = None
else:
# these are only needed for use_stdin, which also requires termios
import tty
import sys
import os
from pdptraps import PDPTraps
from unibus import BusCycle
class KL11:
KL11_DEFAULT = 0o17560 # offset within I/O page
BUSY = 0o400 # probably no reason to ever set this
RCDONE = 0o200 # means character available in buf
TXRDY = 0o200 # same bit on tx side is called "RDY"
IENABLE = 0o100
RDRENA = 0o001
_SHUTDOWN_SENTINEL = object()
SERVERHOST = ''
SERVERPORT = 1170
HALT_SEQUENCE = bytes((0xc3, 0xa7))
# HALT_SEQUENCE = bytes((0x05,)) # this will do ^E to exit, like SIMH
def __init__(self, ub, /, *, baseaddr=KL11_DEFAULT,
send_telnet=False, use_stdin=False):
"""Initialize the emulated console. Listens on port 1170.
Argument use_stdin (True/False, default False) determines whether
a remote socket is used or stdin (python console). This only works
on python hosts with termios (generally any POSIX-like system)
If use_stdin is False, argument send_telnet (True/False, dflt False)
controls whether RFC854 sequences to turn off echo, etc will be sent.
"""
self.addr = baseaddr
self.ub = ub
self.ub.register(ub.autobyte(self.klregs), baseaddr, 4)
self.send_telnet = send_telnet
self.start_this = None
# output characters are just queued (via tq) to the output thread
# input characters have to undergo a more careful 1-by-1
# dance to properly match interrupts to their arrival
self.tq = queue.Queue()
self.rxc = threading.Condition()
# bits broken out of virtualized KL11 Reader Status Register
self.rcdone = False
self.r_ienable = False
# reader buffer register (address: baseaddr + 2)
self.rdrbuf = 0
# transmit buffer status (address: baseaddr + 4)
self.t_ienable = False
if use_stdin:
if not termios:
raise ValueError("Cannot use_stdin without termios module")
serverloop = self._stdindeferred
else:
serverloop = self._connectionserver
# The socket server connection/listener
self._t = threading.Thread(target=serverloop, daemon=True)
self._t.start()
def _telnetsequences(self, s):
"""If telnet is being used to connect, turn off local echo etc."""
dont_auth = bytes((0xff, 0xf4, 0x25))
s.sendall(dont_auth)
suppress_goahead = bytes((0xff, 0xfb, 0x03))
s.sendall(suppress_goahead)
dont_linemode = bytes((0xff, 0xfe, 0x22))
s.sendall(dont_linemode)
dont_new_env = bytes((0xff, 0xfe, 0x27))
s.sendall(dont_new_env)
will_echo = bytes((0xff, 0xfb, 0x01))
s.sendall(will_echo)
dont_echo = bytes((0xff, 0xfe, 0x01))
s.sendall(dont_echo)
noecho = bytes((0xff, 0xfd, 0x2d))
s.sendall(noecho)
def klregs(self, addr, cycle, /, *, value=None):
if self.start_this:
t, self.start_this = self.start_this, None
t.start()
if cycle == BusCycle.RESET:
self.rcdone = False
self.r_ienable = False
self.r_tenable = False
return
match addr - self.addr:
case 0: # rcsr
if cycle == BusCycle.READ16:
value = 0
if self.r_ienable:
value |= self.IENABLE
if self.rcdone:
value |= self.RCDONE
elif cycle == BusCycle.WRITE16:
if value & self.RDRENA:
with self.rxc:
# a request to get one character, which only
# has to clear the rcdone bit here.
self.rcdone = False
self.rdrbuf = 0
self.r_ienable = (value & self.IENABLE)
self.rxc.notify()
case 2 if cycle == BusCycle.READ16: # rbuf
with self.rxc:
value = self.rdrbuf
self.rcdone = False
self.rxc.notify()
# transmit buffer status (sometimes called tcsr)
case 4:
if cycle == BusCycle.READ16:
value = self.TXRDY # always ready to send chars
if self.t_ienable:
value |= self.IENABLE
elif cycle == BusCycle.WRITE16:
prev = self.t_ienable
self.t_ienable = (value & self.IENABLE)
if self.t_ienable and not prev:
self.ub.intmgr.simple_irq(pri=4, vector=0o64)
# transmit buffer
case 6 if cycle == BusCycle.WRITE16: # tbuf
value &= 0o177
if (value != 0o177):
s = chr(value)
self.tq.put(s)
if self.t_ienable:
self.ub.intmgr.simple_irq(pri=4, vector=0o64)
case _:
raise PDPTraps.AddressError
return value
def _stdindeferred(self):
# because the stdin method steals the console I/O, which hoses
# interactive use (if a KL11 has been set up), the start of the
# stdin thread is deferred until any first KL11 access.
self.start_this = threading.Thread(
target=self._stdinserver, daemon=True)
def _stdinserver(self):
"""Server loop daemon thread for console I/O via stdin."""
@contextmanager
def _rawmode(fd):
saved = termios.tcgetattr(fd)
try:
tty.setraw(fd)
yield
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, saved)
def _outloop(q, outf):
while True:
try:
c = q.get()
if c is self._SHUTDOWN_SENTINEL:
break
except queue.Empty:
pass
else:
outf.write(c.encode())
def _inloop(inf):
in_halt_sequence = None
while len(c := inf.read(1)) != 0:
b = ord(c)
with self.rxc:
if in_halt_sequence is None:
if b == self.HALT_SEQUENCE[0]:
in_halt_sequence = 1
if len(self.HALT_SEQUENCE) == 1:
return
elif b == self.HALT_SEQUENCE[in_halt_sequence]:
in_halt_sequence += 1
if len(self.HALT_SEQUENCE) == in_halt_sequence:
return
else:
in_halt_sequence = None
self.rxc.wait_for(lambda: not self.rcdone)
self.rdrbuf = b
self.rcdone = True
if self.r_ienable:
self.ub.intmgr.simple_irq(pri=4, vector=0o60)
with _rawmode(sys.stdin.fileno()):
inf = os.fdopen(sys.stdin.fileno(), 'rb',
buffering=0, closefd=False)
outf = os.fdopen(sys.stdout.fileno(), 'wb',
buffering=0, closefd=False)
outthread = threading.Thread(target=_outloop, args=(self.tq, outf))
inthread = threading.Thread(target=_inloop, args=(inf,))
outthread.start()
inthread.start()
inthread.join()
self.tq.put(self._SHUTDOWN_SENTINEL)
outthread.join()
self.ub.intmgr.halt_toggle(f"CONSOLE HALT ({self.HALT_SEQUENCE})")
def _connectionserver(self):
"""Server loop daemon thread for console I/O via telnet/nc socket."""
serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
serversocket.bind((self.SERVERHOST, self.SERVERPORT))
serversocket.listen(1)
def _outloop(q, s):
while True:
try:
c = q.get()
if c is self._SHUTDOWN_SENTINEL:
break
except queue.Empty:
pass
else:
s.sendall(c.encode())
def _inloop(s):
while len(b := s.recv(1)) != 0:
with self.rxc:
self.rxc.wait_for(lambda: not self.rcdone)
self.rdrbuf = ord(b)
self.rcdone = True
if self.r_ienable:
self.ub.intmgr.simple_irq(pri=4, vector=0o60)
while True:
s, addr = serversocket.accept()
if self.send_telnet:
self._telnetsequences(s)
outthread = threading.Thread(target=_outloop, args=(self.tq, s))
inthread = threading.Thread(target=_inloop, args=(s,))
outthread.start()
inthread.start()
inthread.join()
self.tq.put(self._SHUTDOWN_SENTINEL)
outthread.join()
# debugging tool
def statestring(self):
s = ""
if self.r_ienable:
s += " RINT"
if self.t_ienable:
s += " TINT"
if self.rdrbuf:
s += f" LC={self.rdrbuf}"
if self.rcdone:
s += f" RCDONE"
return s