5e249e8ad0
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.
1256 lines
49 KiB
Python
1256 lines
49 KiB
Python
#!/usr/bin/env python3
|
|
"""Talk to a SIC Marking e10-V6 marking controller over RS-232.
|
|
|
|
Written against NOTPROTOE10V6US-R08 (the e10-V6 communication protocol manual)
|
|
and VERIFIED against a real controller on 2026-08-27 - a P63, firmware
|
|
"E10 P63 (20140924)". Listing, reading and the machine query all work over the
|
|
HOST port; a 52-file snapshot came off it in one command.
|
|
|
|
Two things the hardware taught us that the manual gets wrong:
|
|
- File names are CASE-SENSITIVE and not always uppercase (see encode_name).
|
|
- File kinds 1 and 7 exist and are readable, though the manual lists only
|
|
0/2/4/5. Kind 1 is where the controller configuration lives, which is what
|
|
makes a complete settings snapshot possible.
|
|
|
|
Still unverified against hardware: the marking path itself (LOAD FILE,
|
|
FILE SET VAR, START MARKING). Those move a stylus, so they wait for a
|
|
deliberate session rather than a probe.
|
|
|
|
WHY BINARY AND NOT TEXT
|
|
The manual defines two protocols. The TEXT one is easier to read and is
|
|
what SIC's own support quoted at us, but it cannot carry the values this
|
|
integration needs:
|
|
|
|
1. Space is the TEXT protocol's delimiter - data is defined as "the data
|
|
of the command ( without Space char)". The GE marker software blanks
|
|
every field to a SINGLE SPACE before setting values ("TmcClrFlds: Put
|
|
a SPACE in each field."). A value that IS a space is unrepresentable.
|
|
2. Real GE part programs mark text containing runs of spaces, e.g.
|
|
"S/N <SERIALNO> 4AM79". TEXT protocol handling of embedded and
|
|
consecutive spaces is undefined by the manual.
|
|
3. TEXT frames carry NO checksum of any kind. The Telesis protocol this
|
|
replaces has a BCC on every frame in both directions.
|
|
|
|
BINARY FILE SET VAR ('7') carries the value as length-framed bytes, 1 to
|
|
127 of them, so a lone space is legal and space runs survive. And the
|
|
binary framing has an optional XOR checksum. Hence: binary for anything
|
|
that reaches a part, text for bench poking only.
|
|
|
|
FRAMING (manual section I.3.a)
|
|
without checksum: [STX][NUL][VER] <command>... [ETX]
|
|
with checksum: [STX][VER] <command>... [ETX][XOR]
|
|
|
|
VER is '5' (0x35). The XOR covers every byte of the string INCLUDING STX
|
|
and ETX. A NUL directly after STX is what disables checksum control, which
|
|
is why every worked example in the manual starts 02 00 35.
|
|
|
|
command: [code][size hi][size lo][data] (size is big-endian)
|
|
|
|
The manual also defines a break-code command form for when you do not want
|
|
to count the data. Not implemented - the length form is unambiguous and we
|
|
always know our own lengths.
|
|
|
|
RESPONSES (manual section I.3.b)
|
|
Whole-string errors arrive as one bare byte:
|
|
BS checksum error HT bad string syntax NAK timeout
|
|
Otherwise: [STX] <answer>... [ETX]
|
|
where answer is [original code][size hi][size lo][data], and data is very
|
|
often a single return code:
|
|
ACK ok HT wrong data syntax
|
|
BEL file not found LF variable not found
|
|
|
|
MARKING COMPLETION IS ASYNCHRONOUS AND TWO-STAGE (manual, START MARKING 'g')
|
|
After the ACK for the start command:
|
|
[EOT] last impact done
|
|
[ENQ] head back at home position
|
|
[NAK][status(3)] fault, and the error must then be reset
|
|
DONE MEANS ENQ, NOT EOT. Calling it done at EOT reports a finished cycle
|
|
while the head is still travelling, which at a bay means an operator
|
|
reaching for a part that is still moving.
|
|
|
|
PAUSE LINES ARE NOT OURS TO ANSWER
|
|
At a PAUSE line the controller sends 'P' and waits for either 'p' on the
|
|
serial line or the physical Start button. This client NEVER sends 'p'. A
|
|
pause exists so a human can reposition the part; answering it from software
|
|
resumes marking on a part nobody has touched. wait_for_mark() surfaces the
|
|
pause to the caller and keeps waiting.
|
|
|
|
DELETE IS DELIBERATELY ABSENT
|
|
DEL_FILE ('D') exists and is documented. It is not implemented here, for
|
|
the same reason Telesis TYPE 'R' was left out of TelesisMarker.psm1: a
|
|
backup tool that can delete is a backup tool that can be made to delete.
|
|
|
|
LINK LAYER
|
|
e10 HOST port is fixed at 9600-8-N-1 and cannot be reconfigured. Pins
|
|
4/6/7/8 are NC, so there are no handshake lines and hardware flow control
|
|
is impossible - the port is opened with flow control off and the manual's
|
|
own backpressure rule applies instead: 500-byte controller buffer, do not
|
|
push faster than it drains, 40000 bytes maximum per string.
|
|
|
|
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.
|
|
|
|
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 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
|
|
VERSION = 0x35 # '5', protocol version byte
|
|
|
|
# return codes and whole-string errors
|
|
EOT = 0x04 # last impact marked
|
|
ENQ = 0x05 # back at home position
|
|
ACK = 0x06 # success
|
|
BEL = 0x07 # file not found
|
|
BS = 0x08 # checksum error (whole string)
|
|
HT = 0x09 # bad syntax
|
|
LF_ = 0x0A # variable not found
|
|
NAK = 0x15 # timeout, or fault during marking
|
|
PAUSE_ASK = 0x50 # 'P', controller waiting at a PAUSE line
|
|
PAUSE_ANSWER = 0x70 # 'p'. deliberately never sent. see module docstring.
|
|
|
|
# command codes, manual section I.3.c
|
|
CMD_LOAD_FILE = 0x63 # 'c'
|
|
CMD_FILE_SET_VAR = 0x37 # '7'
|
|
CMD_START_MARKING = 0x67 # 'g'
|
|
CMD_RESET_ERROR = 0x45 # 'E'
|
|
CMD_FILE_LIST = 0x4C # 'L'
|
|
CMD_PC_TO_E10 = 0x47 # 'G'
|
|
CMD_E10_TO_PC = 0x53 # 'S'
|
|
CMD_GET_MACHINE = 0x81
|
|
|
|
# file kinds, manual FILE LIST
|
|
KIND_ALL = 0
|
|
|
|
# 1 and 7 are UNDOCUMENTED - the manual lists only 0/2/4/5. Found on a real
|
|
# controller: kind 1 holds ConfigCtrl (the controller configuration),
|
|
# File-Select (the digital-input file selection map) and FolderDef; kind 7 held
|
|
# SAVE_ACC_1, which looks like accessory or axis data. They pull fine over
|
|
# E10_TO_PC, which is what makes a complete settings snapshot possible at all.
|
|
KIND_CONFIG = 1
|
|
KIND_MARKING = 2
|
|
KIND_DOT_LOGO = 4
|
|
KIND_SCRIBBLE_LOGO = 5
|
|
KIND_ACCESSORY = 7
|
|
|
|
KIND_NAMES = {
|
|
KIND_CONFIG: "config",
|
|
KIND_MARKING: "marking",
|
|
KIND_DOT_LOGO: "dotlogo",
|
|
KIND_SCRIBBLE_LOGO: "scribblelogo",
|
|
KIND_ACCESSORY: "accessory",
|
|
}
|
|
|
|
# Files whose contents carry the programmer/supervisor password in cleartext.
|
|
# They ARE captured - a config snapshot is the only thing that restores a
|
|
# controller as it was - but anything that renders a snapshot to a human should
|
|
# check this first and show the name only.
|
|
SENSITIVE_NAMES = ("ConfigCtrl",)
|
|
|
|
|
|
def is_sensitive(name):
|
|
"""True if this file's contents hold a credential."""
|
|
return name in SENSITIVE_NAMES
|
|
|
|
# name field is 11 bytes plus one reserved NUL byte
|
|
NAME_LEN = 11
|
|
NAME_FIELD_LEN = 12
|
|
LIST_ENTRY_LEN = 19 # name 11 + reserved 1 + kind 1 + size 2 + reserved 4
|
|
|
|
RETURN_CODE_TEXT = {
|
|
ACK: "ok",
|
|
HT: "wrong data syntax",
|
|
BEL: "file not found",
|
|
LF_: "variable not found",
|
|
}
|
|
|
|
STRING_ERROR_TEXT = {
|
|
BS: "checksum error",
|
|
HT: "bad string syntax",
|
|
NAK: "timeout",
|
|
}
|
|
|
|
|
|
# Marking fault bits, manual fig 1. The status is 3 bytes, most significant
|
|
# first, and the bits combine - 00 30 00 is out-of-bounds on X (10 00 + 20 00).
|
|
FAULT_BITS = [
|
|
(0, 0x01, "marking font"),
|
|
(0, 0x02, "dot logo"),
|
|
(0, 0x04, "vectorial logo"),
|
|
(0, 0x08, "Ecc200"),
|
|
(0, 0x10, "text zone syntax"),
|
|
(0, 0x20, "variable"),
|
|
(0, 0x40, "I/O"),
|
|
(0, 0x80, "RS232"),
|
|
(1, 0x01, "stop button activated"),
|
|
(1, 0x02, "stylus"),
|
|
(1, 0x04, "motor"),
|
|
(1, 0x08, "sensor"),
|
|
(1, 0x10, "out of marking window bounds"),
|
|
(1, 0x20, "X axis"),
|
|
(1, 0x40, "Y axis"),
|
|
(1, 0x80, "accessory axis"),
|
|
(2, 0x01, "blocked feeder, or autosensing found no part"),
|
|
(2, 0x02, "empty feeder, part out of bounds, or binary axis"),
|
|
(2, 0x04, "marking head lost steps"),
|
|
(2, 0x08, "external motor"),
|
|
(2, 0x10, "history full"),
|
|
(2, 0x20, "duplicate detected for history"),
|
|
(2, 0x40, "stylus needs changing"),
|
|
(2, 0x80, "stylus has to be changed"),
|
|
]
|
|
|
|
|
|
def decode_fault(status):
|
|
"""Turn a 3-byte marking status into readable reasons."""
|
|
# the wire order is most significant byte first, so index 2 is Err1
|
|
ordered = [status[2], status[1], status[0]] if len(status) == 3 else []
|
|
reasons = []
|
|
for index, mask, text in FAULT_BITS:
|
|
if index < len(ordered) and ordered[index] & mask:
|
|
reasons.append(text)
|
|
return reasons or ["unknown status %s" % status.hex()]
|
|
|
|
|
|
class E10Error(Exception):
|
|
"""Any protocol-level failure talking to the controller."""
|
|
|
|
|
|
class E10Fault(E10Error):
|
|
"""Controller reported a fault. Carries the raw 3-byte machine status."""
|
|
|
|
def __init__(self, status):
|
|
self.status = status
|
|
self.reasons = decode_fault(status)
|
|
super().__init__("marking fault [%s]: %s"
|
|
% (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.
|
|
|
|
No spaces, 11 characters maximum, NUL padded.
|
|
|
|
CASE IS PRESERVED EXACTLY, and that matters. The manual says names are
|
|
"in upper-case", but a real controller lists files named ConfigCtrl,
|
|
File-Select and FolderDef - mixed case, and asking for CONFIGCTRL gets
|
|
nothing back. So the manual describes a convention, not a constraint, and
|
|
the only safe thing to send is whatever the listing gave us.
|
|
"""
|
|
if " " in name:
|
|
raise E10Error("e10 file names cannot contain spaces: %r" % name)
|
|
if len(name) > NAME_LEN:
|
|
raise E10Error("e10 file name over %d chars: %r" % (NAME_LEN, name))
|
|
return name.encode("ascii").ljust(NAME_FIELD_LEN, b"\x00")
|
|
|
|
|
|
def decode_name(raw):
|
|
"""Pull a file name back out of a 12-byte name field."""
|
|
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.
|
|
|
|
Use as a context manager. Every method that reaches a part goes through the
|
|
binary protocol; the text_* helpers exist for bench work only.
|
|
"""
|
|
|
|
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.port = None
|
|
self._var_sizes = {}
|
|
self._loaded = None
|
|
|
|
# ---------------------------------------------------------------- port
|
|
|
|
def open(self):
|
|
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 - which happened repeatedly during development.
|
|
settle = time.monotonic() + 0.25
|
|
while time.monotonic() < settle:
|
|
if not self.port.read(256):
|
|
time.sleep(0.01)
|
|
self.port.flush()
|
|
return self
|
|
|
|
def close(self):
|
|
if self.port is not None:
|
|
self.port.close()
|
|
self.port = None
|
|
self._var_sizes = {}
|
|
self._loaded = None
|
|
|
|
def __enter__(self):
|
|
return self.open()
|
|
|
|
def __exit__(self, *exc):
|
|
self.close()
|
|
|
|
def _log(self, direction, data):
|
|
if self.debug:
|
|
sys.stderr.write("%s %s\n" % (direction, data.hex(" ")))
|
|
|
|
def _flush_input(self):
|
|
"""Drop anything sitting in the receive buffer before we transmit.
|
|
|
|
Line noise accumulates between transactions and would otherwise be
|
|
read back as the front of the next reply.
|
|
"""
|
|
self.port.flush()
|
|
|
|
def _write(self, data):
|
|
self._flush_input()
|
|
self._log("TX", data)
|
|
self.port.write(data)
|
|
|
|
def _read_byte(self, deadline):
|
|
"""One byte, or None once the deadline passes."""
|
|
while True:
|
|
chunk = self.port.read(1)
|
|
if chunk:
|
|
return chunk[0]
|
|
if time.monotonic() >= deadline:
|
|
return None
|
|
time.sleep(0.002)
|
|
|
|
# ------------------------------------------------------------- framing
|
|
|
|
def _frame(self, commands):
|
|
"""Wrap encoded commands in a protocol string.
|
|
|
|
commands is already-encoded bytes, one or more [code][size][data].
|
|
"""
|
|
if self.checksum:
|
|
body = bytes([STX, VERSION]) + commands + bytes([ETX])
|
|
xor = 0
|
|
for byte in body:
|
|
xor ^= byte
|
|
return body + bytes([xor])
|
|
# NUL straight after STX is what turns checksum control off
|
|
return bytes([STX, 0x00, VERSION]) + commands + bytes([ETX])
|
|
|
|
@staticmethod
|
|
def _command(code, data=b""):
|
|
if len(data) > 0xFFFF:
|
|
raise E10Error("command data too long: %d bytes" % len(data))
|
|
return bytes([code, (len(data) >> 8) & 0xFF, len(data) & 0xFF]) + data
|
|
|
|
def _read_response(self, timeout=None):
|
|
"""Read one response string and return {code: data}.
|
|
|
|
Raises E10Error on a bare whole-string error byte.
|
|
"""
|
|
deadline = time.monotonic() + (self.timeout if timeout is None else timeout)
|
|
|
|
# HUNT FOR THE FRAME START rather than demanding it immediately.
|
|
# Two things put junk in front of a good reply on this link: framing
|
|
# errors from line noise arrive as 0x00, and the tail of an earlier
|
|
# reply (an abandoned transaction, a killed script) is still in flight.
|
|
# Neither should fail a transaction whose real answer is right behind
|
|
# it. A frame starts with STX; a whole-string rejection is one bare
|
|
# error byte. Anything else in between is noise worth logging and
|
|
# skipping.
|
|
skipped = bytearray()
|
|
while True:
|
|
first = self._read_byte(deadline)
|
|
if first is None:
|
|
if skipped:
|
|
raise E10Error(
|
|
"no valid frame; saw only %s" % bytes(skipped).hex(" "))
|
|
raise E10Error("no response from controller (timeout)")
|
|
if first == STX:
|
|
break
|
|
if first in STRING_ERROR_TEXT:
|
|
raise E10Error("controller rejected the string: %s"
|
|
% STRING_ERROR_TEXT[first])
|
|
skipped.append(first)
|
|
if len(skipped) > 256:
|
|
raise E10Error("no frame start in 256 bytes: %s"
|
|
% bytes(skipped[:32]).hex(" "))
|
|
if skipped:
|
|
self._log("RX skipped before STX", bytes(skipped))
|
|
|
|
answers = {}
|
|
while True:
|
|
code = self._read_byte(deadline)
|
|
if code is None:
|
|
raise E10Error("truncated response, no ETX")
|
|
if code == ETX:
|
|
break
|
|
hi = self._read_byte(deadline)
|
|
lo = self._read_byte(deadline)
|
|
if hi is None or lo is None:
|
|
raise E10Error("truncated response reading answer size")
|
|
size = (hi << 8) | lo
|
|
data = bytearray()
|
|
while len(data) < size:
|
|
byte = self._read_byte(deadline)
|
|
if byte is None:
|
|
raise E10Error("truncated response reading answer data")
|
|
data.append(byte)
|
|
answers[code] = bytes(data)
|
|
|
|
self._log("RX", b"".join(
|
|
bytes([c]) + d for c, d in answers.items()))
|
|
return answers
|
|
|
|
def _single(self, code, data=b"", timeout=None):
|
|
"""Send one command, return its answer bytes."""
|
|
self._write(self._frame(self._command(code, data)))
|
|
answers = self._read_response(timeout=timeout)
|
|
if code not in answers:
|
|
raise E10Error("no answer for command 0x%02X" % code)
|
|
return answers[code]
|
|
|
|
@staticmethod
|
|
def _check_return_code(answer, what):
|
|
if len(answer) != 1:
|
|
raise E10Error("%s: expected a return code, got %d bytes"
|
|
% (what, len(answer)))
|
|
if answer[0] == ACK:
|
|
return
|
|
raise E10Error("%s: %s" % (
|
|
what, RETURN_CODE_TEXT.get(answer[0], "code 0x%02X" % answer[0])))
|
|
|
|
# ------------------------------------------------------- marking path
|
|
|
|
def load_file(self, name, check_capacity=True):
|
|
"""LOAD FILE 'c'. Select a marking file on the controller."""
|
|
upper = name.upper()
|
|
if " " in upper:
|
|
raise E10Error("e10 file names cannot contain spaces: %r" % name)
|
|
if not 1 <= len(upper) <= NAME_LEN:
|
|
raise E10Error("e10 file name must be 1 to %d chars: %r"
|
|
% (NAME_LEN, name))
|
|
self._check_return_code(
|
|
self._single(CMD_LOAD_FILE, upper.encode("ascii")),
|
|
"load_file(%s)" % upper)
|
|
self._loaded = name
|
|
# pull the declared buffer sizes so set_var can refuse to overflow one
|
|
if check_capacity:
|
|
try:
|
|
self._var_sizes = self.variable_capacities(name)
|
|
except E10Error:
|
|
self._var_sizes = {}
|
|
|
|
def set_var(self, var, value):
|
|
"""FILE SET VAR '7'. Set one variable in the loaded file.
|
|
|
|
value is bytes or str. Spaces are fine - this is the whole reason the
|
|
marking path is binary rather than text. A single space is the correct
|
|
way to blank a field, matching what the Telesis side does.
|
|
"""
|
|
name = var.upper()
|
|
if " " in name:
|
|
raise E10Error("e10 variable names cannot contain spaces: %r" % var)
|
|
if not 1 <= len(name) <= 20:
|
|
raise E10Error("e10 variable name must be 1 to 20 chars: %r" % var)
|
|
payload = value.encode("ascii") if isinstance(value, str) else bytes(value)
|
|
if not 1 <= len(payload) <= 127:
|
|
raise E10Error("e10 variable value must be 1 to 127 bytes, got %d"
|
|
% len(payload))
|
|
self._check_capacity(name, payload)
|
|
data = name.encode("ascii") + b"=" + payload
|
|
self._check_return_code(self._single(CMD_FILE_SET_VAR, data),
|
|
"set_var(%s)" % name)
|
|
|
|
def _check_capacity(self, name, payload):
|
|
"""Refuse to send more than the variable's declared buffer holds.
|
|
|
|
THE CONTROLLER TRUNCATES SILENTLY AT MARK TIME. Each variable is
|
|
declared in the marking file with a value buffer size, and a longer
|
|
value is cut to fit when the pattern renders - while SETVAR answers ok.
|
|
Observed on hardware: a variable declared with a 10-byte buffer was
|
|
given "07482 07482ABC" and the machine displayed "07482 0748".
|
|
|
|
GETVAR IS NO USE FOR DETECTING THIS. It echoes the full value back
|
|
whatever the buffer size, so a read-back check passes while the mark
|
|
would still come out short. The only honest check is against the size
|
|
declared in the file, which is why load_file() reads it.
|
|
|
|
On a part marker this is the worst failure available: a truncated
|
|
serial number engraved on a real part, reported as success.
|
|
"""
|
|
limit = self._var_sizes.get(name)
|
|
if limit is None or len(payload) <= limit:
|
|
return
|
|
raise E10Error(
|
|
"set_var(%s): value is %d bytes but the variable is declared with a"
|
|
" %d-byte buffer in %s. The controller would truncate it silently"
|
|
" and mark %r."
|
|
% (name, len(payload), limit, self._loaded or "the loaded file",
|
|
payload[:limit]))
|
|
|
|
def variable_capacities(self, name, kind=KIND_MARKING):
|
|
"""Declared value buffer size per variable, read from the file itself.
|
|
|
|
Record layout: [name_len][value_size][class][00][name][value buffer].
|
|
Records are located through the offset table at 0x0E; entry count is
|
|
implied by the first offset. See make-e10-pattern.py for the full
|
|
format notes.
|
|
"""
|
|
raw = self.read_file(name, kind=kind)
|
|
if not raw.startswith(b"MARK_V5.0\x00"):
|
|
return {}
|
|
first = (raw[0x0E] << 8) | raw[0x0F]
|
|
entries = (first - 0x0E) // 2
|
|
if entries < 1:
|
|
return {}
|
|
offsets = [(raw[0x0E + 2 * i] << 8) | raw[0x0F + 2 * i]
|
|
for i in range(entries)]
|
|
sizes = {}
|
|
for index, start in enumerate(offsets):
|
|
end = offsets[index + 1] if index + 1 < len(offsets) else len(raw)
|
|
if end <= start:
|
|
end = len(raw)
|
|
rec = raw[start:end]
|
|
if (len(rec) >= 5 and rec[2] in (0x40, 0x50) and rec[3] == 0x00
|
|
and len(rec) == 4 + rec[0] + rec[1]):
|
|
sizes[rec[4:4 + rec[0]].decode("ascii", "replace")] = rec[1]
|
|
return sizes
|
|
|
|
def clear_var(self, var):
|
|
"""Blank a variable to a single space.
|
|
|
|
Mirrors the Telesis TmcClrFlds behaviour. Only reachable at all because
|
|
the value is length-framed here.
|
|
"""
|
|
self.set_var(var, " ")
|
|
|
|
def start_marking(self, simulate=False):
|
|
"""START MARKING 'g'. Returns once the controller has ACKed the start.
|
|
|
|
The cycle itself is asynchronous - call wait_for_mark() next.
|
|
"""
|
|
mode = b"\x01" if simulate else b"\x00"
|
|
self._check_return_code(self._single(CMD_START_MARKING, mode),
|
|
"start_marking")
|
|
|
|
def wait_for_mark(self, timeout=120.0, on_pause=None):
|
|
"""Wait out a marking cycle. Returns only once the head is home.
|
|
|
|
Sequence is [EOT] last impact, then [ENQ] home. Anything that reports
|
|
completion at EOT is lying about a head that is still moving.
|
|
|
|
A PAUSE line makes the controller send 'P' and wait. This NEVER answers
|
|
it - on_pause (if given) is called so the caller can tell somebody, and
|
|
the wait continues until a human presses Start.
|
|
"""
|
|
deadline = time.monotonic() + timeout
|
|
saw_eot = False
|
|
while True:
|
|
byte = self._read_byte(deadline)
|
|
if byte is None:
|
|
raise E10Error("timed out waiting for the cycle to finish"
|
|
" (saw_eot=%s)" % saw_eot)
|
|
if byte == EOT:
|
|
saw_eot = True
|
|
elif byte == ENQ:
|
|
return # home. and only now is the cycle actually over.
|
|
elif byte == NAK:
|
|
status = bytearray()
|
|
while len(status) < 3:
|
|
part = self._read_byte(deadline)
|
|
if part is None:
|
|
raise E10Error("fault reported but status truncated")
|
|
status.append(part)
|
|
raise E10Fault(bytes(status))
|
|
elif byte == PAUSE_ASK:
|
|
if on_pause:
|
|
on_pause()
|
|
# deliberately no reply. the Start button is the only way on.
|
|
else:
|
|
self._log("RX unexpected", bytes([byte]))
|
|
|
|
def reset_error(self):
|
|
"""RESET_ERREUR 'E'. Required after any fault before marking again."""
|
|
self._check_return_code(self._single(CMD_RESET_ERROR), "reset_error")
|
|
|
|
# -------------------------------------------------------- backup path
|
|
|
|
def list_files(self, name=None, kind=KIND_ALL):
|
|
"""FILE LIST 'L'. Returns [{name, kind, size}].
|
|
|
|
name None lists everything of the given kind.
|
|
"""
|
|
field = (encode_name(name) if name
|
|
else b"\x00" * NAME_FIELD_LEN)
|
|
answer = self._single(CMD_FILE_LIST, field + bytes([kind]))
|
|
if len(answer) % LIST_ENTRY_LEN:
|
|
raise E10Error("file list not a whole number of %d-byte entries"
|
|
" (%d bytes)" % (LIST_ENTRY_LEN, len(answer)))
|
|
files = []
|
|
for offset in range(0, len(answer), LIST_ENTRY_LEN):
|
|
entry = answer[offset:offset + LIST_ENTRY_LEN]
|
|
files.append({
|
|
"name": decode_name(entry),
|
|
"kind": entry[12],
|
|
"size": (entry[13] << 8) | entry[14],
|
|
})
|
|
return files
|
|
|
|
def read_file(self, name, kind=KIND_MARKING):
|
|
"""E10_TO_PC 'S'. Pull one file whole. No block dance, unlike Telesis."""
|
|
answer = self._single(CMD_E10_TO_PC, encode_name(name) + bytes([kind]),
|
|
timeout=max(self.timeout, 10.0))
|
|
if len(answer) <= NAME_FIELD_LEN + 1:
|
|
raise E10Error("read_file(%s): controller returned no contents"
|
|
" (file not found?)" % name)
|
|
return answer[NAME_FIELD_LEN + 1:]
|
|
|
|
def write_file(self, name, contents, kind=KIND_MARKING):
|
|
"""PC_TO_E10 'G'. Push one file back.
|
|
|
|
Restore path. Nothing calls this on a schedule - it exists so a
|
|
controller swap can be put back the way it was.
|
|
"""
|
|
data = encode_name(name) + bytes([kind]) + bytes(contents)
|
|
self._check_return_code(
|
|
self._single(CMD_PC_TO_E10, data, timeout=max(self.timeout, 30.0)),
|
|
"write_file(%s)" % name)
|
|
|
|
def get_machine(self):
|
|
"""GET_MACHINE (129). Raw machine configuration bytes."""
|
|
return self._single(CMD_GET_MACHINE)
|
|
|
|
# ---------------------------------------------------------- text mode
|
|
|
|
def text_command(self, line, timeout=None):
|
|
"""Send one TEXT protocol line and return the reply.
|
|
|
|
BENCH ONLY. Do not build a marking path on this - see the module
|
|
docstring for the three reasons. Useful for GETVERSION as a first
|
|
contact test, and for settling whether this firmware answers SETVAR or
|
|
SETTEXTVAR (the manual uses both, in different chapters).
|
|
"""
|
|
self._write(line.encode("ascii") + b"\n")
|
|
deadline = time.monotonic() + (self.timeout if timeout is None else timeout)
|
|
out = bytearray()
|
|
while True:
|
|
byte = self._read_byte(deadline)
|
|
if byte is None:
|
|
if out:
|
|
break
|
|
raise E10Error("no response to text command %r" % line)
|
|
if byte == 0x0A: # LF ends the reply
|
|
break
|
|
if byte != 0x0D:
|
|
out.append(byte)
|
|
reply = bytes(out).decode("ascii", "replace")
|
|
self._log("RX text", bytes(out))
|
|
return reply
|
|
|
|
|
|
# ------------------------------------------------------------------- CLI
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Bench client for a SIC Marking e10 controller.")
|
|
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")
|
|
parser.add_argument("-v", "--debug", action="store_true",
|
|
help="hex dump every frame")
|
|
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")
|
|
|
|
p_list = sub.add_parser("list", help="list files on the controller")
|
|
p_list.add_argument("--kind", type=int, default=KIND_ALL)
|
|
|
|
p_get = sub.add_parser("get", help="read a file off the controller")
|
|
p_get.add_argument("name")
|
|
p_get.add_argument("-o", "--out")
|
|
p_get.add_argument("--kind", type=int, default=KIND_MARKING)
|
|
|
|
p_put = sub.add_parser("put", help="write a file to the controller")
|
|
p_put.add_argument("name")
|
|
p_put.add_argument("path")
|
|
p_put.add_argument("--kind", type=int, default=KIND_MARKING)
|
|
|
|
p_load = sub.add_parser("load", help="select a marking file")
|
|
p_load.add_argument("name")
|
|
|
|
p_set = sub.add_parser("setvar", help="set a variable in the loaded file")
|
|
p_set.add_argument("var")
|
|
p_set.add_argument("value")
|
|
|
|
p_mark = sub.add_parser("mark", help="run the loaded file")
|
|
p_mark.add_argument("--simulate", action="store_true",
|
|
help="force 0, stylus stays up")
|
|
|
|
sub.add_parser("reset", help="clear a controller fault")
|
|
sub.add_parser("machine", help="dump machine configuration bytes")
|
|
|
|
p_backup = sub.add_parser(
|
|
"backup", help="pull every marking file and logo into a directory")
|
|
p_backup.add_argument("outdir")
|
|
|
|
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":
|
|
print(e10.text_command("GETVERSION"))
|
|
|
|
elif args.cmd == "text":
|
|
print(e10.text_command(args.line))
|
|
|
|
elif args.cmd == "list":
|
|
for entry in e10.list_files(kind=args.kind):
|
|
print("%-12s kind=%-14s %6d bytes" % (
|
|
entry["name"],
|
|
KIND_NAMES.get(entry["kind"], entry["kind"]),
|
|
entry["size"]))
|
|
|
|
elif args.cmd == "get":
|
|
blob = e10.read_file(args.name, kind=args.kind)
|
|
if args.out:
|
|
with open(args.out, "wb") as handle:
|
|
handle.write(blob)
|
|
print("%s -> %s (%d bytes)" % (args.name, args.out, len(blob)))
|
|
else:
|
|
sys.stdout.buffer.write(blob)
|
|
|
|
elif args.cmd == "put":
|
|
with open(args.path, "rb") as handle:
|
|
blob = handle.read()
|
|
e10.write_file(args.name, blob, kind=args.kind)
|
|
print("%s -> controller (%d bytes)" % (args.name, len(blob)))
|
|
|
|
elif args.cmd == "load":
|
|
e10.load_file(args.name)
|
|
print("loaded %s" % args.name.upper())
|
|
|
|
elif args.cmd == "setvar":
|
|
e10.set_var(args.var, args.value)
|
|
print("%s = %r" % (args.var.upper(), args.value))
|
|
|
|
elif args.cmd == "mark":
|
|
e10.start_marking(simulate=args.simulate)
|
|
print("started, waiting for the head to come home...")
|
|
e10.wait_for_mark(on_pause=lambda: print(
|
|
"PAUSE line reached. press Start on the machine."))
|
|
print("cycle complete")
|
|
|
|
elif args.cmd == "reset":
|
|
e10.reset_error()
|
|
print("fault cleared")
|
|
|
|
elif args.cmd == "machine":
|
|
print(e10.get_machine().hex(" "))
|
|
|
|
elif args.cmd == "backup":
|
|
os.makedirs(args.outdir, exist_ok=True)
|
|
count = 0
|
|
failed = []
|
|
for entry in e10.list_files():
|
|
suffix = KIND_NAMES.get(entry["kind"], "kind%s" % entry["kind"])
|
|
# one unreadable file must not abandon the whole snapshot
|
|
try:
|
|
blob = e10.read_file(entry["name"], kind=entry["kind"])
|
|
except E10Error as err:
|
|
failed.append((entry["name"], str(err)))
|
|
print("%-12s %6s FAILED: %s"
|
|
% (entry["name"], "", err))
|
|
continue
|
|
path = os.path.join(
|
|
args.outdir, "%s.%s" % (entry["name"], suffix))
|
|
with open(path, "wb") as handle:
|
|
handle.write(blob)
|
|
print("%-12s %6d bytes -> %s%s" % (
|
|
entry["name"], len(blob), path,
|
|
" [CONTAINS CREDENTIAL]" if is_sensitive(entry["name"])
|
|
else ""))
|
|
count += 1
|
|
print("\n%d files captured, %d failed" % (count, len(failed)))
|
|
for name, err in failed:
|
|
print(" FAILED %-12s %s" % (name, err))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except E10Fault as fault:
|
|
sys.stderr.write("controller fault: status %s\n"
|
|
"run 'reset' before marking again\n"
|
|
% fault.status.hex(" "))
|
|
sys.exit(2)
|
|
except E10Error as err:
|
|
sys.stderr.write("e10: %s\n" % err)
|
|
sys.exit(1)
|