ge-aerospace/pxe-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(e10): run on Windows as well as Linux

Browse Source 
Serial access moves behind open_port(), with a termios backend on POSIX and a
Win32 backend reached through ctypes. Deliberately no pyserial: these tools
have to run on locked-down, air-gapped shopfloor machines where installing a
package is a project of its own.

Verified on a real controller from both a Linux laptop and a Windows 11 PC.

Three things that made this harder than it should have been, all now handled:

  - CreateFileW returns a HANDLE, which is 64-bit. Without an explicit restype
    ctypes truncates it to c_int. The open then appears to succeed and every
    read and write silently operates on a bogus handle, which looks exactly
    like a dead cable. Every kernel32 call now declares argtypes and restype.

  - Defaulting to a fixed COM number was actively harmful. A wrong-but-real
    port opens without error and answers nothing, so defaulting to COM3 while
    the adapter was on COM4 produced the same silent symptom. Windows now
    auto-picks only when exactly one port looks like a USB virtual COM port,
    and otherwise refuses and lists what it found.

  - COM10 and above need the \\.\ prefix, which is harmless on COM1-9, so it
    is always applied.

Adds two diagnostics, because "no response" has too many possible causes to
guess at:
  ports    every serial port that actually exists. On Windows this reads
           SERIALCOMM from the registry, so a port absent here does not exist
           now, whatever Device Manager showed last week.
  doctor   walks the layers - discovery, open, write, read - and reports which
           one fails, including a warning when the configured port is an
           onboard one while a USB adapter is present.

Port selection is -d, then $E10_PORT, then the safe auto-pick. Friendlier
errors on both platforms for a missing port or a permissions problem.
This commit is contained in:
cproudlock
2026-08-27 13:40:15 -04:00
parent eb4d067f12
commit 5e249e8ad0
2 changed files with 477 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

514
scripts/e10-client.py
View File

@@ -90,15 +90,31 @@ LINK LAYER
Both ends are DTE (each calls pin 2 RX and pin 3 TX), so the cable must be
a crossover: 3 to 2, 2 to 3, 5 to 5.
Stdlib only, termios - no pyserial on the bench box.
RUNS ON LINUX AND WINDOWS, stdlib only - no pyserial. Serial access sits
behind open_port(): termios on POSIX, the Win32 API through ctypes on Windows.
That choice is deliberate rather than purist: these tools have to run on
locked-down, air-gapped shopfloor machines where installing a package is a
project of its own.
Pick the port with -d, or set E10_PORT once:
Windows set E10_PORT=COM4 Linux export E10_PORT=/dev/ttyUSB0
"""
import argparse
import os
import sys
import termios
import time
# Serial access differs per platform and neither module exists on the other.
# termios is POSIX; the Windows path goes through the Win32 API via ctypes so
# there is nothing to pip-install on an air-gapped shopfloor PC.
if os.name == "nt":
import ctypes
from ctypes import wintypes
termios = None
else:
import termios
# framing
STX = 0x02
ETX = 0x03
@@ -233,6 +249,200 @@ class E10Fault(E10Error):
% (status.hex(" "), "; ".join(self.reasons)))
DEFAULT_PORT_LINUX = "/dev/ttyUSB0"
def default_port():
"""Which serial port to talk to, without editing any source.
Order of precedence:
1. the --device / -d argument
2. the E10_PORT environment variable
3. a per-platform default
Set E10_PORT once and every tool here picks it up. On Windows that is a
COM name as Device Manager shows it; on Linux a device path.
Windows set E10_PORT=COM4
Linux export E10_PORT=/dev/ttyUSB0
"""
from_env = os.environ.get("E10_PORT")
if from_env:
return from_env
# NEVER blind-default to a fixed COM number on Windows. A wrong-but-real
# port opens without error and answers nothing, which is indistinguishable
# from a dead cable - it cost a debugging round trip when the default was
# COM3 (an onboard \Device\Serial0) while the adapter was COM4.
# Prefer a port that is clearly a USB virtual COM port; if that is
# ambiguous, say so rather than guessing.
if os.name == "nt":
ports = list_ports()
virtual = [name for name, detail in ports
if "VCP" in detail.upper() or "USB" in detail.upper()]
if len(virtual) == 1:
return virtual[0]
if not ports:
raise E10Error(
"no serial ports on this machine. Plug the USB-serial adapter"
" in, then run: e10-client.py ports")
raise E10Error(
"cannot tell which port to use - set E10_PORT or pass -d.\n"
" available: %s\n"
" the adapter is normally the \\Device\\VCP* one.\n"
" example: set E10_PORT=%s"
% (", ".join("%s (%s)" % (n, d or "?") for n, d in ports),
(virtual or [ports[0][0]])[0]))
return DEFAULT_PORT_LINUX
def list_ports():
"""Every serial port this machine actually has, with a description.
Windows: read HKLM\\HARDWARE\\DEVICEMAP\\SERIALCOMM, which lists only
real, present serial devices - so a port missing from here does not exist
right now, whatever Device Manager showed last week.
Linux: glob the usual device nodes.
"""
found = []
if os.name == "nt":
import winreg
try:
key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE,
r"HARDWARE\DEVICEMAP\SERIALCOMM")
except OSError:
return found
index = 0
while True:
try:
name, value, _ = winreg.EnumValue(key, index)
except OSError:
break
found.append((value, name))
index += 1
winreg.CloseKey(key)
else:
import glob
for path in sorted(glob.glob("/dev/ttyUSB*") +
glob.glob("/dev/ttyACM*")):
found.append((path, _linux_detail(path)))
# ttyS0..ttyS31 exist as 8250 placeholders on almost every Linux
# machine and are nearly all phantom - every discriminator I tried
# (a /sys device node, a bound driver) is true for all 32. Listing
# them buries the one port that matters under 32 lines of noise, and
# a USB adapter is never one of them, so they are summarised instead.
legacy = sorted(glob.glob("/dev/ttyS*"))
if legacy:
found.append(("(%d x /dev/ttyS*)" % len(legacy),
"legacy onboard ports, rarely the adapter"))
return found
def _linux_detail(path):
"""Best-effort description of a USB serial adapter."""
base = os.path.basename(path)
try:
link = os.path.realpath("/sys/class/tty/%s/device" % base)
for _ in range(4):
link = os.path.dirname(link)
product = os.path.join(link, "product")
if os.path.exists(product):
with open(product) as handle:
return handle.read().strip()
except OSError:
pass
return ""
def doctor(device, timeout=3.0):
"""Work out why a port is silent, one layer at a time."""
print("PORT DISCOVERY")
ports = list_ports()
if not ports:
print(" no serial ports found on this machine at all")
for name, detail in ports:
mark = " <-- configured" if name.upper() == (device or "").upper() else ""
print(" %-10s %s%s" % (name, detail, mark))
print()
print("CONFIGURED PORT: %s" % device)
if os.name == "nt" and ports:
names = [n.upper() for n, _ in ports]
if device.upper() not in names:
print(" *** NOT IN THE LIST ABOVE ***")
print(" This port does not exist. Set E10_PORT to one that does:")
print(" set E10_PORT=%s" % ports[0][0])
return
if os.name == "nt" and ports:
detail = dict((n.upper(), d) for n, d in ports).get(
(device or "").upper(), "")
virtual = [n for n, d in ports
if "VCP" in d.upper() or "USB" in d.upper()]
if virtual and device.upper() not in [v.upper() for v in virtual]:
print(" WARNING: %s is %s, which is not a USB adapter."
% (device, detail or "an onboard port"))
print(" A port like this opens fine and answers nothing.")
print(" The adapter is probably: %s" % ", ".join(virtual))
print()
print("OPENING")
try:
port = open_port(device)
except E10Error as err:
print(" FAILED: %s" % err)
return
print(" opened ok")
if os.name == "nt":
print(" handle: 0x%X (a plausible handle is small and non-zero;"
" 0xFFFFFFFFFFFFFFFF means the open really failed)"
% port.handle)
print()
print("WRITING")
probe = b"GETDATETIME\n"
try:
port.write(probe)
print(" wrote %d bytes: %r" % (len(probe), probe))
except E10Error as err:
print(" FAILED: %s" % err)
port.close()
return
print()
print("READING for %.1fs" % timeout)
got = bytearray()
end = time.monotonic() + timeout
while time.monotonic() < end:
chunk = port.read(256)
if chunk:
got += chunk
else:
time.sleep(0.01)
port.close()
if got:
print(" %d bytes: %s" % (len(got), bytes(got).hex(" ")))
print(" as text : %r" % bytes(got))
print()
print(" THE LINK WORKS.")
else:
print(" nothing came back.")
print()
print(" The port opened and the write succeeded, so the software and")
print(" the driver are fine. That leaves, in order:")
print(" 1. WRONG PORT - it opened, but is it the adapter? A")
print(" Bluetooth or virtual COM port opens happily and never")
print(" answers. Unplug the adapter, re-run, and see which")
print(" entry disappears from the list above.")
print(" 2. CABLE - both ends are DTE, so it must be a CROSSOVER.")
print(" Meter it: pin 2 at one end to pin 3 at the other.")
print(" 3. WRONG SOCKET on the controller - use HOST, not SERIAL.")
print(" 4. Controller powered off, or not finished booting.")
def encode_name(name):
"""Pack a file name into the 12-byte name field the controller expects.
@@ -256,6 +466,234 @@ def decode_name(raw):
return raw[:NAME_FIELD_LEN].split(b"\x00", 1)[0].decode("ascii", "replace")
class _PosixPort:
"""Serial port through termios. Raw, 8-N-1, no flow control."""
def __init__(self, device, baud=9600):
try:
self.fd = os.open(device,
os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
except FileNotFoundError:
raise E10Error(
"no such port: %s. Is the USB-serial adapter plugged in?"
" Check with: ls /dev/ttyUSB*" % device)
except PermissionError:
raise E10Error(
"permission denied on %s. Add yourself to the dialout group"
" (sudo usermod -aG dialout $USER, then log out and in), or"
" run under: sg dialout -c '...'" % device)
speed = getattr(termios, "B%d" % baud)
attrs = termios.tcgetattr(self.fd)
cc = list(attrs[6])
cc[termios.VMIN] = 0
cc[termios.VTIME] = 0
# raw: no canonical mode, echo, signals or translation. CLOCAL matters
# because the e10 HOST port has no handshake pins, so waiting on
# carrier detect would hang forever.
termios.tcsetattr(self.fd, termios.TCSANOW,
[0, 0,
termios.CS8 | termios.CREAD | termios.CLOCAL,
0, speed, speed, cc])
self.flush()
def read(self, count):
try:
return os.read(self.fd, count)
except BlockingIOError:
return b""
except OSError:
return b""
def write(self, data):
offset = 0
while offset < len(data):
offset += os.write(self.fd, data[offset:])
def flush(self):
try:
termios.tcflush(self.fd, termios.TCIOFLUSH)
except OSError:
pass
def close(self):
os.close(self.fd)
class _WindowsPort:
r"""Serial port through the Win32 API, reached with ctypes.
Deliberately no pyserial: these tools have to run on locked-down,
air-gapped shopfloor machines where installing a package is a project of
its own. Everything here is in the standard library.
Port naming: COM10 and above REQUIRE the \\.\ prefix, and it is harmless
on COM1-COM9, so it is always applied. Without it COM10 fails to open with
a file-not-found error that looks nothing like the real cause.
"""
GENERIC_READ = 0x80000000
GENERIC_WRITE = 0x40000000
OPEN_EXISTING = 3
INVALID_HANDLE = (1 << 64) - 1 # 0xFFFF...FF, what CreateFileW returns
PURGE_RXCLEAR = 0x0008
PURGE_TXCLEAR = 0x0004
MAXDWORD = 0xFFFFFFFF
def __init__(self, device, baud=9600):
self.k32 = ctypes.WinDLL("kernel32", use_last_error=True)
self._declare()
name = device if device.startswith("\\\\") else "\\\\.\\" + device
self.handle = self.k32.CreateFileW(
ctypes.c_wchar_p(name),
self.GENERIC_READ | self.GENERIC_WRITE,
0, # no sharing - a serial port is exclusive
None,
self.OPEN_EXISTING,
0, # synchronous
None)
if self.handle is None or self.handle == self.INVALID_HANDLE:
code = ctypes.get_last_error()
hint = ""
if code == 2:
hint = (" - no such port. Check Device Manager under"
" 'Ports (COM & LPT)'.")
elif code == 5:
hint = (" - access denied. Another program has the port open;"
" close any terminal or vendor software using it.")
raise E10Error("cannot open %s: Windows error %d%s"
% (device, code, hint))
self._configure(baud)
self.flush()
def _declare(self):
"""Declare argument and return types for every kernel32 call.
THIS IS NOT OPTIONAL ON 64-BIT WINDOWS. A HANDLE is a pointer, so it is
64 bits, but an undeclared ctypes function returns c_int - 32 bits. The
handle from CreateFileW is then silently TRUNCATED. It does not match
INVALID_HANDLE_VALUE, so the open appears to succeed, and every
subsequent read and write operates on a handle that is not the port and
returns nothing at all.
The symptom is a tool that runs, reports no error, and never receives a
byte - which is indistinguishable from a dead cable, and cost a round
trip to diagnose.
"""
k = self.k32
k.CreateFileW.restype = wintypes.HANDLE
k.CreateFileW.argtypes = [wintypes.LPCWSTR, wintypes.DWORD,
wintypes.DWORD, wintypes.LPVOID,
wintypes.DWORD, wintypes.DWORD,
wintypes.HANDLE]
k.CloseHandle.restype = wintypes.BOOL
k.CloseHandle.argtypes = [wintypes.HANDLE]
k.GetCommState.restype = wintypes.BOOL
k.GetCommState.argtypes = [wintypes.HANDLE, wintypes.LPVOID]
k.SetCommState.restype = wintypes.BOOL
k.SetCommState.argtypes = [wintypes.HANDLE, wintypes.LPVOID]
k.SetCommTimeouts.restype = wintypes.BOOL
k.SetCommTimeouts.argtypes = [wintypes.HANDLE, wintypes.LPVOID]
k.PurgeComm.restype = wintypes.BOOL
k.PurgeComm.argtypes = [wintypes.HANDLE, wintypes.DWORD]
k.ReadFile.restype = wintypes.BOOL
k.ReadFile.argtypes = [wintypes.HANDLE, wintypes.LPVOID,
wintypes.DWORD, wintypes.LPDWORD,
wintypes.LPVOID]
k.WriteFile.restype = wintypes.BOOL
k.WriteFile.argtypes = [wintypes.HANDLE, wintypes.LPCVOID,
wintypes.DWORD, wintypes.LPDWORD,
wintypes.LPVOID]
k.ClearCommError.restype = wintypes.BOOL
k.ClearCommError.argtypes = [wintypes.HANDLE, wintypes.LPDWORD,
wintypes.LPVOID]
def _configure(self, baud):
class DCB(ctypes.Structure):
_fields_ = [("DCBlength", wintypes.DWORD),
("BaudRate", wintypes.DWORD),
("bitfield", wintypes.DWORD),
("wReserved", wintypes.WORD),
("XonLim", wintypes.WORD),
("XoffLim", wintypes.WORD),
("ByteSize", ctypes.c_ubyte),
("Parity", ctypes.c_ubyte),
("StopBits", ctypes.c_ubyte),
("XonChar", ctypes.c_char),
("XoffChar", ctypes.c_char),
("ErrorChar", ctypes.c_char),
("EofChar", ctypes.c_char),
("EvtChar", ctypes.c_char),
("wReserved1", wintypes.WORD)]
class COMMTIMEOUTS(ctypes.Structure):
_fields_ = [("ReadIntervalTimeout", wintypes.DWORD),
("ReadTotalTimeoutMultiplier", wintypes.DWORD),
("ReadTotalTimeoutConstant", wintypes.DWORD),
("WriteTotalTimeoutMultiplier", wintypes.DWORD),
("WriteTotalTimeoutConstant", wintypes.DWORD)]
dcb = DCB()
dcb.DCBlength = ctypes.sizeof(DCB)
if not self.k32.GetCommState(self.handle, ctypes.byref(dcb)):
raise E10Error("GetCommState failed: Windows error %d"
% ctypes.get_last_error())
dcb.BaudRate = baud
dcb.ByteSize = 8
dcb.Parity = 0 # NOPARITY
dcb.StopBits = 0 # ONESTOPBIT
# bit 0 is fBinary. Every other flag stays clear, which disables
# XON/XOFF, CTS/DSR handshaking, and leaves DTR and RTS disabled -
# correct here, because the HOST port has no handshake pins at all.
dcb.bitfield = 0x0001
if not self.k32.SetCommState(self.handle, ctypes.byref(dcb)):
raise E10Error("SetCommState failed: Windows error %d"
% ctypes.get_last_error())
# Return whatever has arrived, immediately, and never block - the
# caller runs its own polling loop with its own deadline.
timeouts = COMMTIMEOUTS(self.MAXDWORD, 0, 0, 0, 1000)
if not self.k32.SetCommTimeouts(self.handle, ctypes.byref(timeouts)):
raise E10Error("SetCommTimeouts failed: Windows error %d"
% ctypes.get_last_error())
def read(self, count):
buf = ctypes.create_string_buffer(count)
got = wintypes.DWORD(0)
if not self.k32.ReadFile(self.handle, buf, count,
ctypes.byref(got), None):
return b""
return buf.raw[:got.value]
def write(self, data):
written = wintypes.DWORD(0)
offset = 0
while offset < len(data):
chunk = ctypes.create_string_buffer(bytes(data[offset:]))
size = len(data) - offset
if not self.k32.WriteFile(self.handle, chunk, size,
ctypes.byref(written), None):
raise E10Error("WriteFile failed: Windows error %d"
% ctypes.get_last_error())
if not written.value:
raise E10Error("WriteFile wrote nothing")
offset += written.value
def flush(self):
self.k32.PurgeComm(self.handle,
self.PURGE_RXCLEAR | self.PURGE_TXCLEAR)
def close(self):
self.k32.CloseHandle(self.handle)
def open_port(device, baud=9600):
"""Open a serial port on whichever platform we are running on."""
return (_WindowsPort(device, baud) if os.name == "nt"
else _PosixPort(device, baud))
class E10:
"""One serial connection to an e10 controller.
@@ -263,55 +701,35 @@ class E10:
binary protocol; the text_* helpers exist for bench work only.
"""
def __init__(self, device="/dev/ttyUSB0", timeout=2.0, checksum=False,
def __init__(self, device=None, timeout=2.0, checksum=False,
debug=False):
device = device or default_port()
self.device = device
self.timeout = timeout
self.checksum = checksum
self.debug = debug
self.fd = None
self.port = None
self._var_sizes = {}
self._loaded = None
# ---------------------------------------------------------------- port
def open(self):
self.fd = os.open(self.device, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
attrs = termios.tcgetattr(self.fd)
iflag, oflag, cflag, lflag, ispeed, ospeed, cc = attrs
# raw. no canonical mode, no echo, no signal chars, no translation.
iflag = 0
oflag = 0
lflag = 0
# 8N1, receiver on, ignore modem lines. CLOCAL matters: HOST port has
# no handshake pins, so waiting on carrier would hang forever.
cflag = termios.CS8 | termios.CREAD | termios.CLOCAL
ispeed = ospeed = termios.B9600
cc = list(cc)
cc[termios.VMIN] = 0
cc[termios.VTIME] = 0
termios.tcsetattr(self.fd, termios.TCSANOW,
[iflag, oflag, cflag, lflag, ispeed, ospeed, cc])
termios.tcflush(self.fd, termios.TCIOFLUSH)
self.port = open_port(self.device)
# Let anything still in flight from a previous session arrive, then
# throw it away. Without this the first transaction reads someone
# else's tail.
# else's tail - which happened repeatedly during development.
settle = time.monotonic() + 0.25
while time.monotonic() < settle:
try:
if not os.read(self.fd, 256):
time.sleep(0.01)
except BlockingIOError:
if not self.port.read(256):
time.sleep(0.01)
termios.tcflush(self.fd, termios.TCIOFLUSH)
self.port.flush()
return self
def close(self):
if self.fd is not None:
os.close(self.fd)
self.fd = None
if self.port is not None:
self.port.close()
self.port = None
self._var_sizes = {}
self._loaded = None
@@ -331,25 +749,17 @@ class E10:
Line noise accumulates between transactions and would otherwise be
read back as the front of the next reply.
"""
try:
termios.tcflush(self.fd, termios.TCIFLUSH)
except OSError:
pass
self.port.flush()
def _write(self, data):
self._flush_input()
self._log("TX", data)
offset = 0
while offset < len(data):
offset += os.write(self.fd, data[offset:])
self.port.write(data)
def _read_byte(self, deadline):
"""One byte, or None once the deadline passes."""
while True:
try:
chunk = os.read(self.fd, 1)
except BlockingIOError:
chunk = b""
chunk = self.port.read(1)
if chunk:
return chunk[0]
if time.monotonic() >= deadline:
@@ -690,7 +1100,11 @@ class E10:
def main():
parser = argparse.ArgumentParser(
description="Bench client for a SIC Marking e10 controller.")
parser.add_argument("-d", "--device", default="/dev/ttyUSB0")
parser.add_argument("-d", "--device", default=None,
help="serial port, e.g. COM4 or"
" /dev/ttyUSB0. Defaults to $E10_PORT; on Windows"
" it will auto-pick a single USB virtual COM port"
" rather than guess.")
parser.add_argument("-t", "--timeout", type=float, default=2.0)
parser.add_argument("--checksum", action="store_true",
help="enable the XOR checksum form of the framing")
@@ -699,6 +1113,9 @@ def main():
sub = parser.add_subparsers(dest="cmd", required=True)
sub.add_parser("version", help="GETVERSION over text. first contact test.")
sub.add_parser("ports", help="list the serial ports this machine has")
sub.add_parser("doctor",
help="diagnose a silent port, one layer at a time")
p_text = sub.add_parser("text", help="send one raw text protocol line")
p_text.add_argument("line")
@@ -736,6 +1153,15 @@ def main():
args = parser.parse_args()
# these two run without opening a session
if args.cmd == "ports":
for name, detail in list_ports():
print(" %-10s %s" % (name, detail))
return
if args.cmd == "doctor":
doctor(args.device or default_port(), timeout=args.timeout)
return
with E10(args.device, timeout=args.timeout, checksum=args.checksum,
debug=args.debug) as e10:
if args.cmd == "version":