new-site/scripts/workers/ink_signature_cli.py

#!/usr/bin/env python3
"""ink_signature_cli — produce/plot an ink signature for a signed filing.

End-to-end glue for the pen-plotter pipeline. Pulls a signed esign record's
captured signature_vector + signature_anchors from the DB, fits the strokes to
the form's signature box, and:

  * writes a G-code file (for the CR-10 V2 / any Marlin/GRBL plotter)
  * writes an SVG toolpath preview (verify before plotting)
  * optionally streams the G-code to the plotter over USB serial (--plot)

Calibration helper:
  * --test-box  draws just the signature-box outline so you can align the jig

This never plots unless --plot is passed (default is generate-only / dry-run),
so it is safe to run anywhere.

Examples:
  # Generate gcode + preview for order CO-ABCD1234's CMS-10114 signature:
  python scripts/workers/ink_signature_cli.py --order CO-ABCD1234 --doc cms10114

  # Calibrate the jig (draw the box outline) to /tmp, dry-run:
  python scripts/workers/ink_signature_cli.py --order CO-ABCD1234 --doc cms10114 --test-box

  # Actually plot it (sheet loaded in jig, printer on /dev/ttyUSB0):
  python scripts/workers/ink_signature_cli.py --order CO-ABCD1234 --doc cms10114 \
      --plot --port /dev/ttyUSB0 --home
"""
from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "scripts"))

# Import the plotter module directly (avoid workers.services package __init__,
# which pulls in heavy browser-automation deps we don't need here).
import importlib.util as _ilu

_spec = _ilu.spec_from_file_location(
    "ink_signature_plotter",
    ROOT / "scripts/workers/services/ink_signature_plotter.py",
)
ink = _ilu.module_from_spec(_spec)
sys.modules["ink_signature_plotter"] = ink
_spec.loader.exec_module(ink)


def _load_record(order: str, doc: str) -> dict:
    import psycopg2
    import psycopg2.extras
    dsn = os.environ.get("DATABASE_URL", "")
    if not dsn:
        sys.exit("DATABASE_URL not set")
    conn = psycopg2.connect(dsn)
    try:
        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
            cur.execute(
                """SELECT id, order_number, document_type, signature_type,
                          signature_vector, signature_anchors, status
                   FROM esign_records
                   WHERE order_number = %s AND document_type = %s
                     AND status = 'signed'
                   ORDER BY signed_at DESC LIMIT 1""",
                (order, doc),
            )
            row = cur.fetchone()
    finally:
        conn.close()
    if not row:
        sys.exit(f"No signed esign record for order={order} doc={doc}")
    return dict(row)


def _coerce_json(val):
    if val is None:
        return None
    if isinstance(val, (dict, list)):
        return val
    return json.loads(val)


def main():
    ap = argparse.ArgumentParser(description="Generate/plot an ink signature for a signed filing.")
    ap.add_argument("--order", required=True, help="order number, e.g. CO-ABCD1234")
    ap.add_argument("--doc", default="cms10114", help="document_type (cms10114, cms855i, ...)")
    ap.add_argument("--field", default="signer", help="signature anchor field to draw")
    ap.add_argument("--out-dir", default="/tmp", help="where to write gcode/svg")
    ap.add_argument("--profile", default="cr10",
                    help="machine profile: cr10 | axidraw | lineus (default cr10)")
    ap.add_argument("--port", default="/dev/ttyUSB0", help="USB serial port (marlin or lineus-USB)")
    ap.add_argument("--baud", type=int, default=115200)
    ap.add_argument("--lineus-host", default="line-us.local", help="Line-us WiFi host (TCP 1337)")
    ap.add_argument("--lineus-usb", action="store_true", help="send Line-us over USB serial (--port) instead of WiFi")
    ap.add_argument("--jig-x", type=float, default=None, help="override jig X offset mm")
    ap.add_argument("--jig-y", type=float, default=None, help="override jig Y offset mm")
    ap.add_argument("--pen-down", type=float, default=None, help="override pen-down Z mm")
    ap.add_argument("--pen-up", type=float, default=None, help="override pen-up Z mm")
    ap.add_argument("--servo-pen", action="store_true", help="force M280 servo pen (marlin profiles)")
    ap.add_argument("--solve-jig", action="store_true",
                    help="for small machines (lineus): compute + apply the jig offset that "
                         "brings the signature cell into reach, then proceed")
    ap.add_argument("--test-box", action="store_true", help="draw only the signature-box outline (calibration)")
    ap.add_argument("--plot", action="store_true", help="actually stream to the plotter (default: dry-run)")
    ap.add_argument("--home", action="store_true", help="prepend G28 home before plotting (marlin)")
    args = ap.parse_args()

    rec = _load_record(args.order, args.doc)
    anchors = _coerce_json(rec.get("signature_anchors")) or []
    box = next((a for a in anchors if a.get("field") == args.field), anchors[0] if anchors else None)
    if not box:
        sys.exit("Record has no signature_anchors — cannot position the signature.")

    # Start from the named profile, then apply any explicit overrides.
    try:
        cfg = ink.load_profile(args.profile)
    except ValueError as exc:
        sys.exit(str(exc))
    overrides = {}
    if args.jig_x is not None:    overrides["jig_x_mm"] = args.jig_x
    if args.jig_y is not None:    overrides["jig_y_mm"] = args.jig_y
    if args.pen_down is not None: overrides["pen_down_mm"] = args.pen_down
    if args.pen_up is not None:   overrides["pen_up_mm"] = args.pen_up
    if args.servo_pen:            overrides["servo_pen"] = True
    if overrides:
        from dataclasses import replace
        cfg = replace(cfg, **overrides)

    # Small-machine jig solve: position the signature into the arm's reach.
    # Load the captured signature vector up-front (needed for solve-jig too).
    vector = _coerce_json(rec.get("signature_vector"))

    if args.solve_jig:
        solved = ink.compute_jig_offset_for_box(box, cfg, vector=vector)
        print(f"[solve-jig] {json.dumps(solved)}")
        if not solved.get("ok"):
            sys.exit("solve-jig: signature does not fit the machine's reach — "
                     "use a larger machine or a smaller signature cell.")
        from dataclasses import replace
        cfg = replace(cfg, jig_x_mm=solved["jig_x_mm"], jig_y_mm=solved["jig_y_mm"])
        print(f"[solve-jig] applied jig offset: x={cfg.jig_x_mm}mm y={cfg.jig_y_mm}mm")

    out_dir = Path(args.out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    stem = f"{args.order}_{args.doc}"

    if args.test_box:
        gcode = ink.emit_test_box_gcode(cfg, box)
        gpath = out_dir / f"{stem}_testbox.gcode"
        gpath.write_text(gcode)
        print(f"[calibration] wrote test-box gcode: {gpath}")
        planned = ink.pdf_points_to_bed_mm(
            [[(box["x"], box["y"]), (box["x"] + box["w"], box["y"]),
              (box["x"] + box["w"], box["y"] + box["h"]), (box["x"], box["y"] + box["h"]),
              (box["x"], box["y"])]], cfg)
    else:
        if not vector:
            sys.exit("Record has no signature_vector (typed signature, or signed before vector capture). "
                     "Use the digital stamp, or re-collect the signature.")
        planned = ink.plan_signature(vector, box, cfg)
        gcode = ink.emit_gcode(planned, cfg)
        gpath = out_dir / f"{stem}_signature.gcode"
        gpath.write_text(gcode)
        print(f"wrote signature gcode: {gpath} "
              f"({len(planned)} strokes, {sum(len(s.points_mm) for s in planned)} points)")

    svg = ink.render_preview_svg(planned, cfg)
    spath = out_dir / f"{stem}_preview.svg"
    spath.write_text(svg)
    print(f"wrote toolpath preview: {spath}")

    # Reach check (gates plotting on small machines, informational on large).
    reach = ink.check_reach(planned, cfg)
    print(f"reach: {json.dumps(reach)}")
    if not reach["ok"]:
        print("WARNING: some points are out of the machine's reach. "
              "For lineus, re-run with --solve-jig (or adjust the jig).")
        if args.plot and cfg.dialect == "lineus":
            sys.exit("Refusing to plot out-of-reach signature on Line-us; fix the jig first.")

    if cfg.dialect == "lineus":
        result = ink.send_lineus(
            gcode,
            host=args.lineus_host,
            serial_port=(args.port if args.lineus_usb else None),
            baud=args.baud,
            dry_run=not args.plot,
        )
        print(f"lineus: {json.dumps(result)}")
    else:
        result = ink.send_gcode_serial(
            gcode, port=args.port, baud=args.baud,
            dry_run=not args.plot, home_first=args.home,
        )
        print(f"serial: {json.dumps(result)}")

    if not args.plot:
        print("NOTE: dry-run (default). Load a sheet in the jig and re-run with --plot to draw.")


if __name__ == "__main__":
    main()