""" 农场端示例客户端（给打码机厂家参考） 这台电脑一头连打码机，一头连我们的溯源服务器： 1. 从服务器下载码到本地（提前下载，断网也能继续打） 2. 逐条交给打码机打印，打完拍照检查 3. 经 WebSocket 回报打了哪些、打废了哪些；不合格的照片和抽样照片传回服务器 4. 连续检查不合格就在本地自动暂停（喷头堵了之类），平台也可以下令暂停或继续 5. 服务器随时可以问「现在打到哪了」，这里负责回答；本机命令行也实时显示进度 厂家要改的只有两个类： PrinterAdapter.print_code() 换成你们打码机的驱动调用 CameraInspector 换成相机取图，以及你们自己的检查方法（OCR 比对、二维码解码） 其余逻辑（本地记账、断线重连、回报补传、暂停）可以照搬，或者照着 4_protocol_trace_code_ws.md 用别的语言重写。 依赖：pip install websockets requests （可选 pip install pillow qrcode，示例才会生成照片） 跑法：python 3_client_farm_printer_sample.py --server <服务器地址> --farm --key --device P1 --print-as text 运行中在命令行输入（回车确认）：p 暂停 r 继续 s 状态 q 停止 """ import argparse import asyncio import collections import io import json import random import sqlite3 import ssl import sys import threading import time import uuid import certifi import requests import websockets LOW_WATER = 200 # 本地剩余码少于这个数就去服务器要新批次 BATCH_SIZE = 1000 # 每次申请多少个码（测试账号每批最多 1000，正式环境可用 --batch-size 调大） PAGE_SIZE = 5000 # 下载分页 REPORT_EVERY_SEC = 3 # 多久回报一次 HEARTBEAT_SEC = 30 FAULT_CONSECUTIVE = 3 # 连续几个检查不合格就自动暂停 SAMPLE_IMAGE_EVERY = 200 # 合格的照片每多少个抽一张上传；不合格的全部上传 class PrinterAdapter: """打码机驱动。示例里是模拟打印；厂家把 print_code 换成真机调用。""" def __init__(self, rate_per_sec=20.0, fail_rate=0.01, fault_after=0): self.rate = rate_per_sec self.fail_rate = fail_rate self.fault_after = fault_after # 模拟：打这么多个之后喷头堵塞，0 表示不模拟 self.since_clean = 0 self.ink_pct = 100.0 def print_code(self, item) -> bool: """ item = {"code": "F07A8K9Q", "text_lines": ["F07A", "8K9Q"], "qr_content": "https://.../T/F07A8K9Q", "print_as": "text" 或 "qr"} print_as == "text"：打两行，每行 4 位。 print_as == "qr"：把 qr_content 打成二维码。 返回 True 表示打码机报告已打出，False 表示打码机自己报错（会作为 void 回报）。 这个调用可以阻塞，外面用线程跑。 """ time.sleep(1.0 / self.rate) self.ink_pct = max(0.0, self.ink_pct - 0.001) self.since_clean += 1 # 模拟喷头堵塞：打码机自己不知道，照样返回 True，要靠拍照检查发现 item["_sim_bad"] = bool(self.fault_after) and self.since_clean > self.fault_after return random.random() >= self.fail_rate def after_resume(self): """暂停后继续时调用。真机可在这里做喷头清洗、自检。示例里当作已清洗。""" self.since_clean = 0 def status(self): return {"ink_pct": round(self.ink_pct, 2), "model": "SIMULATOR"} class CameraInspector: """ 打码后拍照并检查。厂家替换两处： capture() 从相机取一张照片，返回 (字节, "image/jpeg" 或 "image/png")，没有相机返回 (None, None) inspect() 判断这一枚打得合不合格，返回 (是否合格, 原因)。 例如：OCR 识别出的字与 item["text_lines"] 比对；二维码解码后与 item["qr_content"] 比对。 示例里的检查是模拟的：inspect() 并不分析照片，只读模拟打码机给的「喷头堵塞」标记； 照片用 Pillow 按这个标记画出来，只为演示上传和自动暂停。没装 Pillow 就不拍照。 实际检查建议：二维码用 cv2.QRCodeDetector 解码后与 qr_content 比对；8 位字码做 OCV（光学字符校验）， 核对打出的是否就是 text_lines，不必做通用 OCR；喷头堵塞可看字迹区域墨点覆盖率与横向断线。 厂家若已有带 OCV 的视觉检测相机，直接接它的结果。 """ def __init__(self): try: from PIL import Image, ImageDraw, ImageFont self.pil = (Image, ImageDraw, ImageFont) self.font = self._font(ImageFont) except ImportError: self.pil = None @staticmethod def _font(ImageFont): for name in ("consolab.ttf", "consola.ttf", "DejaVuSansMono-Bold.ttf", "Menlo.ttc"): try: return ImageFont.truetype(name, 64) except OSError: continue try: return ImageFont.load_default(size=64) except TypeError: return ImageFont.load_default() def capture(self, item): if not self.pil: return None, None Image, ImageDraw, _ = self.pil bad = item.get("_sim_bad") img = Image.new("RGB", (360, 240), (240, 222, 196)) d = ImageDraw.Draw(img) if item["print_as"] == "qr": try: import qrcode q = qrcode.QRCode(error_correction=qrcode.constants.ERROR_CORRECT_H, box_size=4, border=1) q.add_data(item["qr_content"]) q.make(fit=True) qimg = q.make_image(fill_color=(40, 28, 5), back_color=(201, 162, 39)).convert("RGB") qimg = qimg.resize((200, 200)) img.paste(qimg, (80, 20)) except ImportError: d.text((40, 90), "QR " + item["code"], fill=(40, 40, 40), font=self.font) else: ink = (150, 150, 150) if bad else (35, 38, 44) d.text((80, 38), item["text_lines"][0], fill=ink, font=self.font) d.text((80, 128), item["text_lines"][1], fill=ink, font=self.font) if bad: # 喷头堵塞的样子：几道横向缺墨 for _ in range(6): y = random.randint(20, 220) d.rectangle((0, y, 360, y + random.randint(4, 10)), fill=(240, 222, 196)) buf = io.BytesIO() img.save(buf, format="PNG") return buf.getvalue(), "image/png" def inspect(self, item, image): if item.get("_sim_bad"): return False, "字迹缺损，疑似喷头堵塞" if item["print_as"] == "text" else "二维码缺损，无法解码" return True, "" class LocalState: """本地 SQLite。断电重启后从这里接着打，不会重打，也不会漏报。""" def __init__(self, path): self.db = sqlite3.connect(path) self.db.executescript(""" CREATE TABLE IF NOT EXISTS codes( batch_id TEXT, seq INTEGER, code TEXT, qr_content TEXT, print_as TEXT, status TEXT DEFAULT 'todo', -- todo / sending / printed / void reported INTEGER DEFAULT 0, -- 1 = 已写进待发队列 done_at REAL, PRIMARY KEY(batch_id, seq)); CREATE TABLE IF NOT EXISTS batches(batch_id TEXT PRIMARY KEY, total INTEGER, downloaded INTEGER); CREATE TABLE IF NOT EXISTS outbox(request_id TEXT PRIMARY KEY, body TEXT, created REAL); """) # 上次断电时正在打的那条：不知道打没打出来。按已打处理，绝不重打（重打会出现两枚同码的蛋）。 self.db.execute("UPDATE codes SET status='printed' WHERE status='sending'") self.db.commit() def todo_count(self): return self.db.execute("SELECT COUNT(*) FROM codes WHERE status='todo'").fetchone()[0] def next_todo(self): r = self.db.execute( "SELECT c.batch_id, c.seq, c.code, c.qr_content, c.print_as FROM codes c " "JOIN batches b ON b.batch_id=c.batch_id WHERE c.status='todo' " "ORDER BY b.rowid, c.seq LIMIT 1").fetchone() if not r: return None return {"batch_id": r[0], "seq": r[1], "code": r[2], "text_lines": [r[2][:4], r[2][4:]], "qr_content": r[3], "print_as": r[4]} def set_status(self, batch_id, seq, status): self.db.execute("UPDATE codes SET status=?, done_at=? WHERE batch_id=? AND seq=?", (status, time.time(), batch_id, seq)) self.db.commit() def known_batches(self): return {r[0]: (r[1], r[2]) for r in self.db.execute("SELECT batch_id, total, downloaded FROM batches")} def save_page(self, batch_id, total, print_as, codes): self.db.execute("INSERT OR IGNORE INTO batches VALUES(?,?,0)", (batch_id, total)) self.db.executemany( "INSERT OR IGNORE INTO codes(batch_id, seq, code, qr_content, print_as, status) VALUES(?,?,?,?,?,?)", [(batch_id, c["seq"], c["code"], c["qr_content"], print_as, {"issued": "todo"}.get(c["status"], c["status"])) for c in codes]) # 服务器上已经是 printed / void 的码本地不再打，也不再回报 self.db.executemany( "UPDATE codes SET reported=1 WHERE batch_id=? AND seq=? AND status!='todo'", [(batch_id, c["seq"]) for c in codes if c["status"] != "issued"]) self.db.execute("UPDATE batches SET downloaded=(SELECT COUNT(*) FROM codes WHERE batch_id=?) WHERE batch_id=?", (batch_id, batch_id)) self.db.commit() def build_reports(self): """把打好 / 打废但还没回报的码，按批次合成连续区间，写进待发队列。""" msgs = [] for status, mtype in (("printed", "printed"), ("void", "void")): rows = self.db.execute( "SELECT batch_id, seq FROM codes WHERE status=? AND reported=0 ORDER BY batch_id, seq", (status,)).fetchall() by_batch = {} for b, s in rows: by_batch.setdefault(b, []).append(s) for b, seqs in by_batch.items(): if mtype == "void": msgs.append({"type": "void", "batch_id": b, "seqs": seqs, "reason": "print_failed"}) continue start = prev = seqs[0] for s in seqs[1:] + [None]: if s is not None and s == prev + 1: prev = s continue msgs.append({"type": "printed", "batch_id": b, "from_seq": start, "to_seq": prev, "ts": time.time()}) if s is not None: start = prev = s self.db.execute("UPDATE codes SET reported=1 WHERE status=? AND reported=0", (status,)) for m in msgs: m["request_id"] = uuid.uuid4().hex self.db.execute("INSERT INTO outbox VALUES(?,?,?)", (m["request_id"], json.dumps(m), time.time())) self.db.commit() def outbox(self): return [json.loads(r[0]) for r in self.db.execute("SELECT body FROM outbox ORDER BY created")] def acked(self, request_id): self.db.execute("DELETE FROM outbox WHERE request_id=?", (request_id,)) self.db.commit() def summary(self): counts = dict(self.db.execute("SELECT status, COUNT(*) FROM codes GROUP BY status").fetchall()) last = self.db.execute( "SELECT batch_id, seq FROM codes WHERE done_at IS NOT NULL ORDER BY done_at DESC LIMIT 1").fetchone() return {"local_todo": counts.get("todo", 0), "local_printed": counts.get("printed", 0), "local_void": counts.get("void", 0), "current_batch": last[0] if last else None, "current_seq": last[1] if last else None, "outbox_pending": self.db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0]} def batch_counts(self, batch_id): rows = self.db.execute("SELECT status, COUNT(*) FROM codes WHERE batch_id=? GROUP BY status", (batch_id,)) return dict(rows.fetchall()) class FarmClient: def __init__(self, args): self.http = args.server.rstrip("/") self.ws_url = self.http.replace("http", "ws", 1) + "/ws/printer" self.farm, self.key, self.device = args.farm, args.key, args.device self.print_as = args.print_as self.max_codes = args.max_codes self.batch_size = getattr(args, "batch_size", None) or BATCH_SIZE self.state = LocalState(args.state_db or f"client_state_{self.farm}_{self.device}.db") self.printer = PrinterAdapter(args.rate, args.fail_rate, getattr(args, "fault_after", 0) or 0) self.camera = CameraInspector() self.keyboard = getattr(args, "keyboard", True) and sys.stdin.isatty() self.tty = sys.stdout.isatty() self.ws = None self.next_download = 0.0 # 下载失败后等一会再试 self.inflight = {} # request_id -> 发出时间，已发未确认 self.uploads = asyncio.Queue(maxsize=200) self.printed_this_run = 0 self.void_this_run = 0 self.bad_streak = 0 self.recent = collections.deque() # 最近打印的时间戳，算速度 self.paused = False self.pause_by = self.pause_reason = None self.stopping = False self.running = "idle" self.batch_started = {} # batch_id -> 本次运行里打这批第一个码的时间 def headers(self): return {"X-Farm-Id": self.farm, "X-Api-Key": self.key} # ---------- 命令行显示 ---------- def log(self, msg): if self.tty: sys.stdout.write("\r\033[K") print(f"{time.strftime('%H:%M:%S')} {msg}", flush=True) def rate(self): now = time.time() while self.recent and now - self.recent[0] > 10: self.recent.popleft() return len(self.recent) / 10.0 def state_name(self): if self.stopping and self.running == "stopped": return "stopped" return "paused" if self.paused else self.running def state_dict(self): return {"state": self.state_name(), "pause_by": self.pause_by, "pause_reason": self.pause_reason, "printed_this_run": self.printed_this_run, "void_this_run": self.void_this_run, "rate": round(self.rate(), 1), **self.printer.status(), **self.state.summary()} def progress_line(self): s = self.state.summary() label = {"printing": "打印中", "no_codes": "等待码", "idle": "准备中", "stopped": "已停止"}.get(self.running, self.running) if self.paused: label = f"已暂停（{self.pause_reason}）" where = f"{s['current_batch']} 第 {s['current_seq']} 条" if s["current_batch"] else "尚未开始" return (f"[{label}] {where} | 本次已打 {self.printed_this_run} 作废 {self.void_this_run} | " f"{self.rate():.1f} 个/秒 | 本地剩余 {s['local_todo']} | 待回报 {s['outbox_pending']} | " f"{'在线' if self.ws else '离线'}") def log_batch_done(self, batch_id): """一批打完留一行，不被进度行覆盖。""" c = self.state.batch_counts(batch_id) if c.get("todo") or c.get("sending"): return line = f"[批次完成] {batch_id}：打印 {c.get('printed', 0)} 作废 {c.get('void', 0)}" start = self.batch_started.pop(batch_id, None) if start: sec = time.time() - start n = c.get("printed", 0) + c.get("void", 0) line += f"，本次用时 {int(sec // 60)} 分 {int(sec % 60)} 秒，平均 {n / max(sec, 1):.1f} 个/秒" self.log(line) async def progress_loop(self): n = 0 while True: await asyncio.sleep(1) n += 1 if self.tty: sys.stdout.write("\r\033[K" + self.progress_line()) sys.stdout.flush() elif n % 10 == 0: # 输出被重定向到文件时每 10 秒记一行 print(time.strftime("%H:%M:%S ") + self.progress_line(), flush=True) # ---------- 暂停 / 继续 / 停止 ---------- def pause(self, by, reason): if self.paused: return self.paused, self.pause_by, self.pause_reason = True, by, reason self.log(f"[暂停] {reason}。输入 r 回车继续") asyncio.get_running_loop().create_task(self.send_state("paused", by, reason)) def resume(self, by): if not self.paused: return self.paused, self.pause_by, self.pause_reason = False, None, None self.bad_streak = 0 self.printer.after_resume() self.log(f"[继续] {'平台指令' if by == 'server' else '现场操作'}") asyncio.get_running_loop().create_task(self.send_state("printing", by, None)) async def send_state(self, state, by, reason): if self.ws: try: await self.ws.send(json.dumps({"type": "state", "state": state, "by": by, "reason": reason, **{k: v for k, v in self.state_dict().items() if k != "state"}})) except Exception: pass # 离线时下一次心跳会带上当前状态 def on_key(self, line): cmd = line.strip().lower() if cmd == "p": self.pause("operator", "现场操作员暂停") elif cmd == "r": self.resume("operator") elif cmd == "s": self.log(json.dumps(self.state_dict(), ensure_ascii=False)) elif cmd == "q": self.log("[停止] 打完当前这一个、把回报发完就退出") self.stopping = True elif cmd: self.log("可用命令：p 暂停 r 继续 s 状态 q 停止") def start_keyboard(self, loop): def reader(): for line in sys.stdin: loop.call_soon_threadsafe(self.on_key, line) threading.Thread(target=reader, daemon=True).start() # ---------- HTTP：拿码 ---------- async def get(self, path, **params): r = await asyncio.to_thread(requests.get, f"{self.http}{path}", headers=self.headers(), params=params, timeout=60) r.raise_for_status() return r.json() async def download_codes(self): """先把服务器上分给本机、还没下完的批次下完；本地码不够再申请新批次。""" known = self.state.known_batches() for b in await self.get("/api/batches", device_id=self.device, open_only="true"): if b["print_as"] != self.print_as: continue # 喷墨机只打字，激光机只打二维码，不拿别的码型 total, downloaded = known.get(b["batch_id"], (b["count"], 0)) if downloaded < total: await self._download_batch(b["batch_id"], downloaded) if self.state.todo_count() < LOW_WATER: r = await asyncio.to_thread( requests.post, f"{self.http}/api/batches", headers=self.headers(), timeout=60, json={"count": self.batch_size, "print_as": self.print_as, "device_id": self.device}) if r.status_code >= 400: raise RuntimeError(f"申请批次被拒：{r.status_code} {r.text[:200]}") await self._download_batch(r.json()["batch_id"], 0) async def _download_batch(self, batch_id, offset): while True: page = await self.get(f"/api/batches/{batch_id}/codes", offset=offset, limit=PAGE_SIZE) if not page["codes"]: return self.state.save_page(batch_id, page["total"], page["print_as"], page["codes"]) offset = page["codes"][-1]["seq"] + 1 self.log(f"[下载] {batch_id} 到第 {offset}/{page['total']} 条") if offset >= page["total"]: return async def download_loop(self): """单独一个任务。网络慢或断了只影响补码，不影响用本地已有的码继续打。""" while True: if self.state.todo_count() < LOW_WATER and time.time() >= self.next_download: try: await self.download_codes() except Exception as e: self.log(f"[下载失败] {e}，本地还有 {self.state.todo_count()} 条，30 秒后再试") self.next_download = time.time() + 30 await asyncio.sleep(1) # ---------- HTTP：传照片 ---------- def queue_image(self, item, image, media, ok, reason): try: self.uploads.put_nowait((item["batch_id"], item["seq"], image, media, ok, reason)) except asyncio.QueueFull: pass # 网络跟不上时丢掉抽样照片，打印不能等照片 async def upload_loop(self): while True: batch_id, seq, image, media, ok, reason = await self.uploads.get() for attempt in range(3): try: r = await asyncio.to_thread( requests.post, f"{self.http}/api/images", data=image, timeout=30, headers={**self.headers(), "Content-Type": media}, params={"device_id": self.device, "batch_id": batch_id, "seq": seq, "ok": int(ok), "reason": reason}) if r.status_code < 400: break self.log(f"[照片被拒] {r.status_code} {r.text[:120]}") break except Exception: await asyncio.sleep(2 * (attempt + 1)) # ---------- 打印循环 ---------- async def print_loop(self): while not self.stopping and (self.max_codes is None or self.printed_this_run + self.void_this_run < self.max_codes): if self.paused: await asyncio.sleep(0.2) continue item = self.state.next_todo() if not item: self.running = "no_codes" await asyncio.sleep(1) continue self.running = "printing" self.state.set_status(item["batch_id"], item["seq"], "sending") # 先记再打 printed = await asyncio.to_thread(self.printer.print_code, item) image, media = await asyncio.to_thread(self.camera.capture, item) good, reason = self.camera.inspect(item, image) if printed else (False, "打码机报告打印失败") self.state.set_status(item["batch_id"], item["seq"], "printed" if good else "void") self.recent.append(time.time()) self.batch_started.setdefault(item["batch_id"], self.recent[-1]) if good: self.printed_this_run += 1 self.bad_streak = 0 if image and self.printed_this_run % SAMPLE_IMAGE_EVERY == 0: self.queue_image(item, image, media, True, "抽样") else: self.void_this_run += 1 self.bad_streak += 1 if image: self.queue_image(item, image, media, False, reason) if self.bad_streak >= FAULT_CONSECUTIVE: # 本地判断：连续不合格，先停下来，等人处理后输入 r 或平台下令继续 self.pause("local", f"连续 {self.bad_streak} 个检查不合格：{reason}") self.log_batch_done(item["batch_id"]) self.running = "stopped" # ---------- WebSocket：回报与应答 ---------- async def report_loop(self): while True: await asyncio.sleep(REPORT_EVERY_SEC) self.state.build_reports() await self.send_outbox() async def send_outbox(self, force=False): """发待发队列。已发出 15 秒还没收到 ack 的才重发；重连后全部重发（force）。""" if not self.ws: return # 断网时留在 outbox，重连后补发 now = time.time() for m in self.state.outbox(): if not force and now - self.inflight.get(m["request_id"], 0) < 15: continue try: await self.ws.send(json.dumps(m)) self.inflight[m["request_id"]] = now except Exception: return async def heartbeat_loop(self): while True: await asyncio.sleep(HEARTBEAT_SEC) if self.ws: try: await self.ws.send(json.dumps({"type": "heartbeat", **self.state_dict()})) except Exception: pass async def ws_loop(self): backoff = [1, 2, 5, 10, 30] attempt = 0 while True: try: # 用 certifi 的根证书，和 requests 一致。Windows 证书库里有坏证书时，默认 context 会报 ASN1 NOT_ENOUGH_DATA tls = ssl.create_default_context(cafile=certifi.where()) if self.ws_url.startswith("wss") else None async with websockets.connect(self.ws_url, open_timeout=10, ssl=tls) as ws: await ws.send(json.dumps({"type": "hello", "farm_id": self.farm, "api_key": self.key, "device_id": self.device, "software": "sample-client/0.2", "printer_model": self.printer.status().get("model")})) welcome = json.loads(await ws.recv()) if welcome.get("type") != "welcome": raise RuntimeError(f"服务器拒绝：{welcome.get('error') or welcome}") self.log(f"[连上] 服务器上本机未打完的批次 {len(welcome['open_batches'])} 个") self.ws, attempt = ws, 0 await self.send_outbox(force=True) # 上次没确认的全部重发，服务器按 request_id 去重 if self.paused: await self.send_state("paused", self.pause_by, self.pause_reason) async for raw in ws: await self.on_message(json.loads(raw)) except Exception as e: self.log(f"[断线] {e}") self.ws = None await asyncio.sleep(backoff[min(attempt, len(backoff) - 1)]) attempt += 1 async def on_message(self, msg): t = msg.get("type") if t == "ack": self.inflight.pop(msg.get("request_id"), None) if msg.get("ok"): self.state.acked(msg["request_id"]) if msg.get("duplicate"): self.log(f"[警告] 服务器说有 {msg['duplicate']} 个码被重复打印") else: self.log(f"[回报被拒] {msg.get('error')}") self.state.acked(msg["request_id"]) # 被拒的重发也不会过，留日志给人查 elif t == "status_request": await self.ws.send(json.dumps({"type": "status", "reply_to": msg["request_id"], **self.state_dict()})) elif t == "control": # 平台下令暂停或继续。多数情况由本地判断暂停，这里留着平台也能叫停 if msg.get("action") == "pause": self.pause("server", msg.get("reason") or "平台要求暂停") elif msg.get("action") == "resume": self.resume("server") await self.ws.send(json.dumps({"type": "control_ack", "reply_to": msg["request_id"], "ok": True, **self.state_dict()})) elif t == "batch_ready": self.log(f"[新批次] {msg['batch_id']}，{msg['count']} 个") self.next_download = 0.0 elif t == "error": self.log(f"[服务器错误] {msg.get('error')}") async def run(self): if self.keyboard: self.start_keyboard(asyncio.get_running_loop()) self.log("命令：p 暂停 r 继续 s 状态 q 停止（输入后回车）") if not self.camera.pil: self.log("没装 Pillow，示例不生成照片，只按打码机返回值判断（pip install pillow qrcode）") tasks = [asyncio.create_task(x) for x in (self.ws_loop(), self.download_loop(), self.report_loop(), self.heartbeat_loop(), self.upload_loop(), self.progress_loop())] await self.print_loop() await self.send_state("stopped", "operator", None) # 把剩下的回报发完再退出 self.state.build_reports() for _ in range(30): await self.send_outbox(force=True) if not self.state.outbox() and self.uploads.empty(): break await asyncio.sleep(1) self.log(f"[结束] 本次已打 {self.printed_this_run}，作废 {self.void_this_run}，{self.state.summary()}") for t in tasks: t.cancel() def main(): p = argparse.ArgumentParser(description="农场端示例客户端") p.add_argument("--server", required=True, help="如 https://softwareagentcenter.com/trace-code") p.add_argument("--farm", required=True) p.add_argument("--key", required=True) p.add_argument("--device", required=True, help="打码机编号，一台机器一个") p.add_argument("--print-as", default="text", choices=["text", "qr"]) p.add_argument("--state-db", default=None, help="本地记录文件，默认 client_state__.db，不要删") p.add_argument("--batch-size", type=int, default=BATCH_SIZE, help="每次申请多少个码") p.add_argument("--rate", type=float, default=20.0, help="模拟打印速度，个/秒") p.add_argument("--fail-rate", type=float, default=0.01, help="模拟打码机报错比例") p.add_argument("--fault-after", type=int, default=0, help="模拟：打这么多个之后喷头堵塞，用来演示自动暂停") p.add_argument("--max-codes", type=int, default=None, help="测试用：打够这么多个就停") asyncio.run(FarmClient(p.parse_args()).run()) if __name__ == "__main__": main()