pxe-server/scripts/fake-telesis.py

#!/usr/bin/env python3
"""Pretend to be a Telesis marking controller, so MarkTelesis will talk to us.

WHY THIS EXISTS
    We need to know exactly what GE's MarkTelesis.exe puts on the wire before a
    translator can impersonate a Telesis controller in front of it. A passive
    tap is useless here: MarkTelesis waits for a reply to every command and
    gives up after the first timeout, so a listener that never answers captures
    one frame and nothing else.

    So this ANSWERS. It decodes each frame, logs it, and sends back the ACK
    shape MarkTelesis is parsing for. That keeps the conversation going long
    enough to record a whole marking job.

    It is also the first half of the real translator - the same decode, minus
    the part where the other side becomes a SIC e10.

THE PROTOCOL, from disassembly of MarkTelesis.exe (2026-08-27)
    request  (host -> controller):
        EOT SOH <TYPE> STX <data> ETX <BCC:3 ascii digits> CR
    reply    (controller -> host):
        SOH <TYPE> ACK STX <data> ETX <BCC:3 ascii digits> CR

    BCC = (TYPE + sum of data bytes) & 0xFF, as three ASCII decimal digits.

    NOTE THE MISSING STATION ID. Backup470 framed requests as
    EOT SOH <TYPE> <2-digit station> STX ... - MarkTelesis does NOT send those
    two digits. Same protocol family, different dialect. The frame builder at
    0x401cf0 pushes exactly EOT, SOH, TYPE, STX, data, ETX, BCC, CR and nothing
    else.

COMMAND TYPES, all confirmed at their call sites
    'P'      pattern load, data is the pattern name
    '1'..'9' numbered line send - the 2Line/3Line/9Line dialect. The TYPE byte
             IS the line number: the loop at 0x4090b0 does `add bl, 0x31`.
    'V'      set field, the TMC400 dialect. Matches Message Type"="V" in
             WJPRT.reg.
    'O'      online, empty data
    'R'      empty data, believed to be the mark/run trigger. INFERRED, not yet
             traced to a UI action - one of the things this capture settles.
    'D'      pattern download, data is "<name>;<6-digit size>;128". Same write
             header as Backup470.

FORCING THE ERROR PATHS
    Happy-path cycles teach us almost nothing we cannot already read out of the
    binary. What we actually need is how MarkTelesis behaves when things go
    wrong, because the translator has to reproduce it:

        --nak TYPE[,TYPE...]    answer these types with NAK instead of ACK
        --drop TYPE[,TYPE...]   do not answer these at all, to see the retry
                                and timeout cadence
        --delay SECONDS         stall before every reply

    Run it once clean, then again with --nak P (pattern not found) and once
    with --drop R, and the resulting logs cover the cases that would otherwise
    only show up at a bay with an operator standing there.
"""

import argparse
import os
import sys
import termios
import time

EOT = 0x04
SOH = 0x01
STX = 0x02
ETX = 0x03
ACK = 0x06
NAK = 0x15
CR = 0x0D

TYPE_MEANING = {
    "P": "pattern load",
    "V": "set field (TMC400 dialect)",
    "O": "online",
    "R": "run / mark (inferred)",
    "D": "pattern download",
    "I": "status request",
}


def describe(type_char):
    if type_char in TYPE_MEANING:
        return TYPE_MEANING[type_char]
    if type_char.isdigit() and type_char != "0":
        return "line %s text (2Line/3Line/9Line dialect)" % type_char
    return "unknown"


def bcc(type_char, data):
    """(TYPE + sum(data)) & 0xFF, as three ASCII decimal digits."""
    total = ord(type_char)
    for byte in data:
        total += byte
    return b"%03d" % (total & 0xFF)


class Line:
    """One serial endpoint, raw, no flow control."""

    def __init__(self, device, baud=termios.B9600):
        self.fd = os.open(device, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
        attrs = termios.tcgetattr(self.fd)
        cc = list(attrs[6])
        cc[termios.VMIN] = 0
        cc[termios.VTIME] = 0
        termios.tcsetattr(self.fd, termios.TCSANOW,
                          [0, 0,
                           termios.CS8 | termios.CREAD | termios.CLOCAL,
                           0, baud, baud, cc])
        termios.tcflush(self.fd, termios.TCIOFLUSH)

    def read_byte(self, timeout=None):
        deadline = None if timeout is None else time.monotonic() + timeout
        while True:
            try:
                chunk = os.read(self.fd, 1)
            except BlockingIOError:
                chunk = b""
            except OSError:
                return None
            if chunk:
                return chunk[0]
            if deadline is not None and time.monotonic() >= deadline:
                return None
            time.sleep(0.002)

    def write(self, data):
        offset = 0
        while offset < len(data):
            offset += os.write(self.fd, data[offset:])

    def close(self):
        os.close(self.fd)


def main():
    parser = argparse.ArgumentParser(
        description="Answer MarkTelesis as if we were a Telesis controller.")
    parser.add_argument("-d", "--device", default="/dev/pts/4",
                        help="serial endpoint (the win11 VM's pty)")
    parser.add_argument("-l", "--log", help="also append the transcript here")
    parser.add_argument("--nak", default="",
                        help="comma-separated TYPEs to answer with NAK")
    parser.add_argument("--drop", default="",
                        help="comma-separated TYPEs to leave unanswered")
    parser.add_argument("--delay", type=float, default=0.0,
                        help="seconds to stall before each reply")
    parser.add_argument("--status", default="0A;000",
                        help="data to return for a Type I status request."
                             " 0A;000 is a real TMC470 idle reply.")
    args = parser.parse_args()

    nak_types = set(t.strip() for t in args.nak.split(",") if t.strip())
    drop_types = set(t.strip() for t in args.drop.split(",") if t.strip())

    log_handle = open(args.log, "a") if args.log else None

    def emit(text):
        stamp = time.strftime("%H:%M:%S")
        line = "%s  %s" % (stamp, text)
        print(line, flush=True)
        if log_handle:
            log_handle.write(line + "\n")
            log_handle.flush()

    line = Line(args.device)
    emit("listening on %s as a Telesis controller" % args.device)
    if nak_types:
        emit("will NAK: %s" % ", ".join(sorted(nak_types)))
    if drop_types:
        emit("will DROP (no reply): %s" % ", ".join(sorted(drop_types)))
    emit("waiting for MarkTelesis...")

    frames = 0
    try:
        while True:
            # requests open with EOT. anything else is noise worth seeing.
            byte = line.read_byte(timeout=None)
            if byte is None:
                continue
            if byte != EOT:
                emit("  stray byte before EOT: 0x%02X" % byte)
                continue

            soh = line.read_byte(timeout=2)
            if soh != SOH:
                emit("  expected SOH after EOT, got %s"
                     % ("nothing" if soh is None else "0x%02X" % soh))
                continue

            type_byte = line.read_byte(timeout=2)
            if type_byte is None:
                emit("  timed out waiting for TYPE")
                continue
            type_char = chr(type_byte)

            stx = line.read_byte(timeout=2)
            if stx != STX:
                emit("  expected STX, got %s"
                     % ("nothing" if stx is None else "0x%02X" % stx))
                continue

            data = bytearray()
            while True:
                nxt = line.read_byte(timeout=2)
                if nxt is None:
                    emit("  timed out reading data")
                    break
                if nxt == ETX:
                    break
                data.append(nxt)

            checksum = bytearray()
            for _ in range(3):
                nxt = line.read_byte(timeout=2)
                if nxt is None:
                    break
                checksum.append(nxt)
            line.read_byte(timeout=1)  # trailing CR

            frames += 1
            expected = bcc(type_char, data)
            ok = "ok" if bytes(checksum) == expected else (
                "MISMATCH, we computed %s" % expected.decode())
            emit("RX #%d  TYPE '%s' (%s)  data=%r  bcc=%s [%s]"
                 % (frames, type_char, describe(type_char), bytes(data),
                    bytes(checksum).decode("ascii", "replace"), ok))

            if type_char in drop_types:
                emit("     -> dropping, no reply (forcing a timeout)")
                continue

            if args.delay:
                time.sleep(args.delay)

            if type_char in nak_types:
                reply = bytes([SOH, ord(type_char), NAK, STX, ETX]) \
                    + bcc(type_char, b"") + bytes([CR])
                emit("     -> NAK")
            else:
                payload = b""
                if type_char == "I":
                    payload = args.status.encode("ascii")
                reply = bytes([SOH, ord(type_char), ACK, STX]) + payload \
                    + bytes([ETX]) + bcc(type_char, payload) + bytes([CR])
                emit("     -> ACK%s" % (" data=%r" % payload if payload else ""))
            line.write(reply)

    except KeyboardInterrupt:
        emit("stopped after %d frames" % frames)
    finally:
        line.close()
        if log_handle:
            log_handle.close()


if __name__ == "__main__":
    main()