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

feat(e10): SIC e10 part marker client, pattern generator and eDNC simulator

Browse Source 
Groundwork for replacing a Telesis pinstamp with a SIC Marking e10 while
keeping the existing eDNC/MarkTelesis operator workflow. MarkTelesis has a
hardcoded marker-type enum and eDNC is serial-only, so the plan is a
translator that impersonates a Telesis controller on a COM port. These are
the pieces that translator will be built from, all verified against a real
e10 (a P63, firmware "E10 P63 (20140924)").

scripts/e10-client.py
  e10-V6 protocol client. Marking path and file transfer both use the BINARY
  protocol: the text protocol uses space as its delimiter, so it cannot carry
  a field cleared to a single space nor the space runs in real marked text,
  and it has no checksum. Includes fault decoding against the manual's fig 1,
  and a guard that refuses to overflow a variable's declared buffer.

scripts/make-e10-pattern.py
  Builds marking files with DECLARED variables. The controller's own protocol
  cannot create one: NEWFILE/INSERTTEXTLINE/SAVEFILE writes the reference but
  leaves the variable count at zero, and SETVAR then fails. Template-based,
  copying through byte-for-byte everything the format notes do not explain.

scripts/simulate-ednc.py
  Parses a part program, substitutes tokens and emits the byte-exact frames
  MarkTelesis would send, then translates them for the e10 and can drive one.
  The wire format came from disassembling MarkTelesis.exe rather than from a
  capture: frame builder at 0x401cf0, and the type letters at their call
  sites. Note there is no station ID, unlike the Backup470 dialect.

scripts/fake-telesis.py
  Answers as a Telesis controller so MarkTelesis will hold a conversation
  long enough to be recorded, with switches to force NAK and timeout paths.

reference/e10/
  52-file snapshot off the bench controller, used as the corpus for reversing
  the MARK_V5.0 format and for testing without hardware. ConfigCtrl is
  excluded and gitignored - it stores the supervisor password in cleartext.
  GE part programs are excluded for the same reason.

Three hardware findings worth knowing:
  - The controller TRUNCATES SILENTLY to a variable's declared buffer size
    while SETVAR still answers ok, and GETVAR echoes the full value back so
    it cannot detect this. Guarded client-side against the size in the file.
  - It refuses to mark when a referenced variable is unset, which makes an
    unset field fail loudly rather than marking the previous part's data.
  - Setting variables does not modify the stored file, so backups will not
    churn on every marked part.
This commit is contained in:
cproudlock
2026-08-27 12:57:20 -04:00
parent f1755751bb
commit eb4d067f12
56 changed files with 1875 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

829
scripts/e10-client.py Normal file
View File

@@ -0,0 +1,829 @@
#!/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.
Stdlib only, termios - no pyserial on the bench box.
"""
import argparse
import os
import sys
import termios
import time
# 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)))
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 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="/dev/ttyUSB0", timeout=2.0, checksum=False,
debug=False):
self.device = device
self.timeout = timeout
self.checksum = checksum
self.debug = debug
self.fd = 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)
# Let anything still in flight from a previous session arrive, then
# throw it away. Without this the first transaction reads someone
# else's tail.
settle = time.monotonic() + 0.25
while time.monotonic() < settle:
try:
if not os.read(self.fd, 256):
time.sleep(0.01)
except BlockingIOError:
time.sleep(0.01)
termios.tcflush(self.fd, termios.TCIOFLUSH)
return self
def close(self):
if self.fd is not None:
os.close(self.fd)
self.fd = 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.
"""
try:
termios.tcflush(self.fd, termios.TCIFLUSH)
except OSError:
pass
def _write(self, data):
self._flush_input()
self._log("TX", data)
offset = 0
while offset < len(data):
offset += os.write(self.fd, data[offset:])
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""
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="/dev/ttyUSB0")
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.")
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()
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)