wrickertandClaude Opus 5.5
4e957509cc
FreeCAD/DXF to .rd conversion, named upload to controller memory with idle check and read-back verification, read-only status/file listing, and a fake controller for dry runs. Co-Authored-By: Claude Opus 5.5 <[email protected]>
123 lines
4.8 KiB
Python
Executable File
123 lines
4.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""rdlist.py: READ-ONLY queries to a Ruida controller: machine status and stored file names.
|
|
|
|
It can only send the byte strings in ALLOWED below; anything else raises before it reaches the
|
|
socket. Never send E8 00 (delete documents) or E8 03 (select document = START it): they sit right
|
|
next to E8 01 (document name) in the protocol. See SKILL.md, "Safety".
|
|
|
|
Protocol as MeerK40t's Ruida driver: swizzle magic 0x88, each datagram prefixed with a 16-bit sum,
|
|
40200 -> 50200. The controller ACKs (CC, C6 on the wire) and then sends a swizzled reply.
|
|
DA 00 04 00 status bits (0x01 job running, 0x02 part end, 0x01000000 moving)
|
|
DA 00 02 00 machine status (MeerK40t emulator: 22 idle, 23 paused, 21 running)
|
|
DA 00 02 05 total document count (read 0 with 49 files stored on our controller: unused)
|
|
E8 01 <n14> name of document n (1-based) -> reply E8 01 <n14> NAME 00
|
|
Confirmed on a real controller: name replies, end of list = no reply.
|
|
Status bit 0 (job running) confirmed on a real controller: set mid-cut (0x20018401), clear when idle (0x2001x600).
|
|
|
|
usage: rdlist.py --host <controller-ip> status + file list
|
|
rdlist.py --host 127.0.0.1 (fake controller)
|
|
"""
|
|
import argparse, socket, sys
|
|
from meerk40t.ruida.rdjob import swizzles_lut, encode14, decodeu35
|
|
|
|
MAGIC = 0x88
|
|
SRC_PORT, DST_PORT = 40200, 50200
|
|
TIMEOUT = 3.0
|
|
ACK, ENQ = 0xCC, 0xCE
|
|
SW, UN = swizzles_lut(MAGIC)
|
|
|
|
MEM_STATUS_BITS = b"\xda\x00\x04\x00"
|
|
MEM_MACHINE_STATUS = b"\xda\x00\x02\x00"
|
|
MEM_DOC_COUNT = b"\xda\x00\x02\x05"
|
|
MAX_DOCS = 200
|
|
ALLOWED = {MEM_STATUS_BITS, MEM_MACHINE_STATUS, MEM_DOC_COUNT} | {
|
|
b"\xe8\x01" + encode14(n) for n in range(1, MAX_DOCS + 1)}
|
|
RUNNING, MOVING = 0x00000001, 0x01000000
|
|
|
|
|
|
class NoReply(Exception):
|
|
pass
|
|
|
|
|
|
class Ruida:
|
|
def __init__(self, host):
|
|
self.host = host
|
|
self.s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
self.s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
self.s.bind(("", SRC_PORT))
|
|
self.s.settimeout(TIMEOUT)
|
|
|
|
def close(self):
|
|
self.s.close()
|
|
|
|
def query(self, cmd): # raises NoReply on timeout
|
|
"""Send one whitelisted read command; return the unswizzled reply payload."""
|
|
if cmd not in ALLOWED:
|
|
raise SystemExit(f"refusing to send non-whitelisted bytes {cmd.hex()}")
|
|
data = bytes(SW[b] for b in cmd)
|
|
self.s.sendto((sum(data) & 0xFFFF).to_bytes(2, "big") + data, (self.host, DST_PORT))
|
|
acked = False
|
|
while True:
|
|
try:
|
|
pkt, _ = self.s.recvfrom(1024)
|
|
except socket.timeout:
|
|
raise NoReply(f"no {'reply' if acked else 'ACK'} to {cmd.hex()} (timeout)")
|
|
r = bytes(UN[b] for b in pkt)
|
|
if len(r) == 1:
|
|
if r[0] == ACK:
|
|
acked = True
|
|
elif r[0] != ENQ:
|
|
raise SystemExit(f"{cmd.hex()}: unexpected reply {r.hex()}")
|
|
continue
|
|
if r[:len(cmd)] != cmd[:len(cmd)] and not (cmd[0] == 0xDA and r[:2] == b"\xda\x01"
|
|
and r[2:4] == cmd[2:4]):
|
|
raise SystemExit(f"{cmd.hex()}: reply {r.hex()} does not match the query")
|
|
return r
|
|
|
|
def mem(self, cmd):
|
|
r = self.query(cmd)
|
|
if len(r) < 9:
|
|
raise SystemExit(f"{cmd.hex()}: short reply {r.hex()}")
|
|
return decodeu35(r[4:9])
|
|
|
|
def status(self):
|
|
return self.mem(MEM_STATUS_BITS), self.mem(MEM_MACHINE_STATUS)
|
|
|
|
def files(self):
|
|
# This controller's doc-count register (DA 00 02 05) reads 0 with 49 files stored, so walk
|
|
# E8 01 n upward; the list ends at the first number that gets no reply (seen on a real controller).
|
|
names = []
|
|
for i in range(1, MAX_DOCS + 1):
|
|
try:
|
|
r = self.query(b"\xe8\x01" + encode14(i))
|
|
except NoReply:
|
|
if i == 1:
|
|
raise SystemExit("no reply to the first document-name query")
|
|
return names
|
|
names.append(r[4:].split(b"\x00")[0].decode("ascii", "replace"))
|
|
raise SystemExit(f"more than MAX_DOCS={MAX_DOCS} documents")
|
|
|
|
|
|
def idle(bits):
|
|
return not (bits & (RUNNING | MOVING))
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--host", required=True, help="controller IP (no default, on purpose)")
|
|
a = ap.parse_args()
|
|
r = Ruida(a.host)
|
|
try:
|
|
bits, ms = r.status()
|
|
print(f"status bits 0x{bits:08x} ({'IDLE' if idle(bits) else 'BUSY'}), machine status {ms}")
|
|
names = r.files()
|
|
print(f"{len(names)} file(s):")
|
|
for i, nm in enumerate(names, 1):
|
|
print(f" {i:3d} {nm}")
|
|
finally:
|
|
r.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|