diff --git a/scripts/e10-client.py b/scripts/e10-client.py index 0075058..d1a2495 100644 --- a/scripts/e10-client.py +++ b/scripts/e10-client.py @@ -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": diff --git a/scripts/simulate-ednc.py b/scripts/simulate-ednc.py index 5d77751..ea2fa60 100644 --- a/scripts/simulate-ednc.py +++ b/scripts/simulate-ednc.py @@ -302,8 +302,10 @@ def main(): help="operator input, in prompt order. repeatable.") parser.add_argument("--no-clear", action="store_true", help="skip the clear-fields pass") - parser.add_argument("--e10", metavar="DEVICE", - help="also drive a real e10 on this serial device") + parser.add_argument("--e10", metavar="DEVICE", nargs="?", const="", + help="also drive a real e10. Give a port, or pass the" + " flag bare to use $E10_PORT / the platform" + " default.") parser.add_argument("--pattern", help="override the pattern name from the ^ line." " Useful when the e10 file is named differently" @@ -383,8 +385,9 @@ def drive_e10(seq, args): e10mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(e10mod) - print("\n DRIVING THE REAL CONTROLLER on %s" % args.e10) - with e10mod.E10(args.e10, timeout=5.0) as dev: + port = args.e10 or e10mod.default_port() + print("\n DRIVING THE REAL CONTROLLER on %s" % port) + with e10mod.E10(port, timeout=5.0) as dev: for label, raw in seq: type_char, data = decode_frame(raw) action, payload = translate(type_char, data)