new-site/scripts/workers/services/ink_signature_plotter.py

"""Pen-plotter ink-signature pipeline: captured strokes -> G-code (and preview).

The Standard (no-login) CMS filing path requires an ORIGINAL ink signature on
paper ("Stamped, faxed or copied signatures will not be accepted"). To produce a
genuine wet-ink signature from a signature captured online, we redraw the
provider's own stroke paths onto the printed form with a pen mounted on a 3-axis
motion system (a Creality CR-10 V2 here, but any Marlin/GRBL machine works).

This module is HARDWARE-INDEPENDENT and fully testable without a plotter:

  strokes (normalized 0..1)            <- esign_records.signature_vector
        |  fit into the signature anchor box (PDF points)
        v
  paper-space coordinates (mm)         <- via PlotterConfig (bed origin + jig)
        |
        +--> emit_gcode()  -> Marlin/GRBL G-code for the CR-10 (Z = pen lift)
        +--> render_preview_svg()/_overlay_pdf() -> verify the strokes land on
             the cert line WITHOUT any hardware (this is our correctness test)

Coordinate frames
-----------------
* Capture box: signature pad canvas, origin top-left, x/y in [0,1].
* PDF: points, origin bottom-left (matches reportlab + our signature anchors).
* Plotter bed: millimetres, origin at the machine's homed corner (front-left for
  a CR-10). A paper jig fixes the sheet at a known offset from that corner, so
  PDF (0,0) of the sheet maps to (jig_x_mm, jig_y_mm) on the bed.

The signature anchor box (x, y, w, h, page_w, page_h in PDF points) tells us
exactly where on the page the signature must land — the same anchors the digital
stamper consumes. We fit the captured strokes into that box (preserving aspect),
then translate everything into bed mm using the jig offset.
"""
from __future__ import annotations

import io
import math
from dataclasses import dataclass, field
from typing import Any

PT_PER_INCH = 72.0
MM_PER_INCH = 25.4
PT_TO_MM = MM_PER_INCH / PT_PER_INCH  # 0.352777...


@dataclass(frozen=True)
class PlotterConfig:
    """Physical configuration of the pen plotter + paper jig.

    Defaults target a Creality CR-10 V2 (300x300x400 bed) with a US-Letter sheet
    held by a corner jig at the front-left of the bed. Tune jig_x_mm / jig_y_mm
    and pen heights during calibration with the test-box routine.
    """
    # Where PDF (0,0) — the sheet's bottom-left corner — sits on the bed, in mm
    # from the machine's homed origin.
    jig_x_mm: float = 20.0
    jig_y_mm: float = 20.0
    # Pen Z heights (mm). pen_down should be the height at which the pen touches
    # paper; with a spring-loaded holder a small over-press is safe.
    pen_up_mm: float = 3.0
    pen_down_mm: float = 0.0
    # Feed rates (mm/min).
    travel_feed: float = 3000.0   # pen-up moves
    draw_feed: float = 1200.0     # pen-down (drawing) moves
    z_feed: float = 600.0         # pen raise/lower
    # Bed safety envelope (mm). Emitter raises if a point would exceed these.
    bed_max_x_mm: float = 300.0
    bed_max_y_mm: float = 300.0
    # Resampling: drop points closer than this (mm) to keep G-code compact and
    # the pen smooth.
    min_segment_mm: float = 0.3
    # Pen-up/down via BLTouch/servo instead of Z (set servo_pen=True to emit
    # M280 deploy/stow instead of G1 Z moves — useful if you mount the pen to a
    # servo or repurpose the BLTouch as the actuator).
    servo_pen: bool = False
    servo_index: int = 0
    servo_down_angle: int = 10
    servo_up_angle: int = 90


@dataclass
class PlannedStroke:
    """A single pen-down polyline in bed mm (one continuous pen contact)."""
    points_mm: list[tuple[float, float]] = field(default_factory=list)


# ── Geometry: strokes (0..1) -> fitted PDF points -> bed mm ──────────────────

def _vector_bbox(strokes: list[list[dict]]) -> tuple[float, float, float, float]:
    """Bounding box (minx, miny, maxx, maxy) of normalized strokes."""
    xs = [p["x"] for s in strokes for p in s]
    ys = [p["y"] for s in strokes for p in s]
    if not xs:
        return 0.0, 0.0, 1.0, 1.0
    return min(xs), min(ys), max(xs), max(ys)


def fit_strokes_to_box(
    vector: dict,
    box: dict,
    *,
    h_pad_frac: float = 0.04,
    v_pad_frac: float = 0.12,
    baseline_lift_pt: float = 1.5,
) -> list[list[tuple[float, float]]]:
    """Fit normalized capture strokes into the anchor box, in PDF points.

    Returns a list of strokes; each stroke is a list of (x_pt, y_pt) with the
    PDF convention (origin bottom-left). Aspect ratio is preserved, the ink is
    left-aligned and rests just above the box bottom (the signature rule), and
    the capture's top-left origin is flipped to PDF's bottom-left.
    """
    strokes = (vector or {}).get("strokes") or []
    if not strokes:
        return []

    minx, miny, maxx, maxy = _vector_bbox(strokes)
    src_w = max(maxx - minx, 1e-6)
    src_h = max(maxy - miny, 1e-6)

    bx, by = float(box["x"]), float(box["y"])
    bw, bh = float(box["w"]), float(box["h"])
    avail_w = bw * (1.0 - 2 * h_pad_frac)
    avail_h = bh * (1.0 - 2 * v_pad_frac)

    # Scale to fit, preserving aspect ratio.
    scale = min(avail_w / src_w, avail_h / src_h)
    drawn_w = src_w * scale
    drawn_h = src_h * scale

    # Left-aligned within the box; bottom-anchored just above the rule.
    off_x = bx + bw * h_pad_frac
    off_y = by + baseline_lift_pt + (avail_h - drawn_h) * 0.0  # rest on baseline

    out: list[list[tuple[float, float]]] = []
    for s in strokes:
        pts: list[tuple[float, float]] = []
        for p in s:
            nx = (p["x"] - minx) * scale
            ny = (p["y"] - miny) * scale
            x_pt = off_x + nx
            # Flip Y: capture origin is top-left, PDF origin is bottom-left.
            y_pt = off_y + (drawn_h - ny)
            pts.append((x_pt, y_pt))
        if pts:
            out.append(pts)
    return out


def pdf_points_to_bed_mm(
    strokes_pt: list[list[tuple[float, float]]],
    cfg: PlotterConfig,
) -> list[PlannedStroke]:
    """Translate fitted PDF-point strokes into bed millimetres via the jig offset."""
    planned: list[PlannedStroke] = []
    for s in strokes_pt:
        mm = [
            (cfg.jig_x_mm + x_pt * PT_TO_MM, cfg.jig_y_mm + y_pt * PT_TO_MM)
            for (x_pt, y_pt) in s
        ]
        mm = _resample(mm, cfg.min_segment_mm)
        if mm:
            planned.append(PlannedStroke(points_mm=mm))
    return planned


def _resample(points: list[tuple[float, float]], min_seg_mm: float) -> list[tuple[float, float]]:
    """Drop points closer than min_seg_mm to the previous kept point."""
    if not points:
        return []
    kept = [points[0]]
    for p in points[1:]:
        lx, ly = kept[-1]
        if math.hypot(p[0] - lx, p[1] - ly) >= min_seg_mm:
            kept.append(p)
    if len(kept) == 1 and len(points) > 1:
        kept.append(points[-1])
    return kept


def plan_signature(vector: dict, box: dict, cfg: PlotterConfig) -> list[PlannedStroke]:
    """Full geometry pipeline: normalized strokes + anchor box -> bed-mm strokes."""
    fitted = fit_strokes_to_box(vector, box)
    return pdf_points_to_bed_mm(fitted, cfg)


# ── G-code emission (Marlin/GRBL) ────────────────────────────────────────────

def emit_gcode(planned: list[PlannedStroke], cfg: PlotterConfig) -> str:
    """Emit Marlin/GRBL G-code that draws the planned strokes with the pen.

    Assumes the machine is homed (G28) so (0,0) is the bed corner; the jig offset
    is already baked into the planned coordinates. Pen up/down is done via Z
    moves (default) or a servo/BLTouch M280 (cfg.servo_pen).
    """
    out: list[str] = []
    w = out.append

    def pen_up():
        if cfg.servo_pen:
            w(f"M280 P{cfg.servo_index} S{cfg.servo_up_angle}  ; pen up")
            w("G4 P150")
        else:
            w(f"G1 Z{cfg.pen_up_mm:.2f} F{cfg.z_feed:.0f}  ; pen up")

    def pen_down():
        if cfg.servo_pen:
            w(f"M280 P{cfg.servo_index} S{cfg.servo_down_angle}  ; pen down")
            w("G4 P150")
        else:
            w(f"G1 Z{cfg.pen_down_mm:.2f} F{cfg.z_feed:.0f}  ; pen down")

    over_bed = any(
        x > cfg.bed_max_x_mm or y > cfg.bed_max_y_mm or x < 0 or y < 0
        for s in planned for (x, y) in s.points_mm
    )

    w("; --- Performance West ink-signature plot ---")
    w("; CR-10 V2 / Marlin (or GRBL); machine assumed homed (G28) before this.")
    if over_bed:
        w("; !! WARNING: some points fall outside the configured bed envelope.")
    w("G21  ; mm")
    w("G90  ; absolute positioning")
    pen_up()

    for s in planned:
        if not s.points_mm:
            continue
        x0, y0 = s.points_mm[0]
        w(f"G0 X{x0:.3f} Y{y0:.3f} F{cfg.travel_feed:.0f}  ; travel to stroke start")
        pen_down()
        for (x, y) in s.points_mm[1:]:
            w(f"G1 X{x:.3f} Y{y:.3f} F{cfg.draw_feed:.0f}")
        pen_up()

    w("G0 X0 Y0 F{:.0f}  ; park".format(cfg.travel_feed))
    w("; --- end plot ---")
    return "\n".join(out) + "\n"


def emit_test_box_gcode(cfg: PlotterConfig, box: dict) -> str:
    """Emit G-code that draws the OUTLINE of the signature box (no signature).

    Use this during calibration: tape a blank sheet in the jig, run this, and
    confirm the rectangle lands exactly on the form's signature line. Adjust
    jig_x_mm/jig_y_mm/pen heights until it does.
    """
    bx, by = float(box["x"]), float(box["y"])
    bw, bh = float(box["w"]), float(box["h"])
    corners_pt = [(bx, by), (bx + bw, by), (bx + bw, by + bh), (bx, by + bh), (bx, by)]
    planned = pdf_points_to_bed_mm([corners_pt], cfg)
    return emit_gcode(planned, cfg)


# ── Serial sender (Marlin/GRBL over USB) — gated, dry-run safe ───────────────

def send_gcode_serial(
    gcode: str,
    port: str = "/dev/ttyUSB0",
    baud: int = 115200,
    *,
    dry_run: bool = True,
    home_first: bool = False,
    line_timeout: float = 30.0,
) -> dict:
    """Stream G-code to a Marlin/GRBL controller over USB serial.

    Defaults to ``dry_run=True`` so it never moves hardware unless explicitly
    enabled — call with dry_run=False only when a sheet is loaded in the jig.

    Marlin acks each line with "ok"; we wait for it before sending the next line
    (simple, reliable flow control). Returns a summary dict.
    """
    lines = [ln for ln in (l.strip() for l in gcode.splitlines())
             if ln and not ln.startswith(";")]
    if home_first:
        lines = ["G28"] + lines

    if dry_run:
        return {
            "sent": False, "dry_run": True, "port": port,
            "lines": len(lines),
            "note": "DRY RUN — no serial I/O. Set dry_run=False to plot.",
        }

    try:
        import serial  # pyserial
    except ImportError as exc:  # pragma: no cover
        raise RuntimeError("pyserial is required to send G-code (pip install pyserial)") from exc

    import time
    ser = serial.Serial(port, baud, timeout=line_timeout)
    try:
        time.sleep(2.0)  # board reset on connect
        ser.reset_input_buffer()
        sent = 0
        for ln in lines:
            ser.write((ln + "\n").encode("ascii"))
            ser.flush()
            # Wait for the controller's "ok" ack.
            deadline = time.time() + line_timeout
            while time.time() < deadline:
                resp = ser.readline().decode("ascii", "ignore").strip().lower()
                if resp.startswith("ok") or resp.startswith("done"):
                    break
                if resp.startswith("error") or resp.startswith("!!"):
                    raise RuntimeError(f"controller error on '{ln}': {resp}")
            sent += 1
        return {"sent": True, "dry_run": False, "port": port, "lines": sent}
    finally:
        ser.close()


# ── Preview rendering (verify WITHOUT hardware) ──────────────────────────────

def render_preview_svg(
    planned: list[PlannedStroke],
    cfg: PlotterConfig,
    *,
    show_bed: bool = True,
) -> str:
    """Render the planned pen path as an SVG in bed mm (origin bottom-left).

    Pen-down strokes are solid; pen-up travels are dashed grey. Lets us eyeball
    the toolpath without a plotter.
    """
    W = cfg.bed_max_x_mm
    H = cfg.bed_max_y_mm
    parts = [
        f'<svg xmlns="http://www.w3.org/2000/svg" width="{W}mm" height="{H}mm" '
        f'viewBox="0 0 {W} {H}">',
        # Flip Y so bottom-left origin reads naturally.
        f'<g transform="translate(0,{H}) scale(1,-1)">',
    ]
    if show_bed:
        parts.append(f'<rect x="0" y="0" width="{W}" height="{H}" fill="#fff" stroke="#ccc" stroke-width="0.5"/>')

    prev_end: tuple[float, float] | None = None
    for s in planned:
        if not s.points_mm:
            continue
        if prev_end is not None:
            x0, y0 = s.points_mm[0]
            parts.append(
                f'<line x1="{prev_end[0]:.2f}" y1="{prev_end[1]:.2f}" '
                f'x2="{x0:.2f}" y2="{y0:.2f}" stroke="#bbb" stroke-width="0.2" '
                f'stroke-dasharray="1,1"/>'
            )
        d = "M " + " L ".join(f"{x:.2f},{y:.2f}" for (x, y) in s.points_mm)
        parts.append(f'<path d="{d}" fill="none" stroke="#10243f" stroke-width="0.6" stroke-linecap="round" stroke-linejoin="round"/>')
        prev_end = s.points_mm[-1]

    parts.append("</g></svg>")
    return "\n".join(parts)


def render_signature_on_pdf(
    pdf_bytes: bytes,
    vector: dict,
    anchors: list[dict],
    *,
    signer_field: str | None = None,
) -> bytes:
    """Stamp the VECTOR strokes onto the PDF at the anchor box(es).

    This is the digital twin of what the pen will draw — same geometry path as
    the plotter — so a visual diff against the real plotted sheet (or just an
    eyeball of this PDF) proves the toolpath lands on the cert line. Returns a
    flattened signed PDF.
    """
    from pypdf import PdfReader, PdfWriter
    from reportlab.pdfgen import canvas as rl_canvas

    reader = PdfReader(io.BytesIO(pdf_bytes))
    writer = PdfWriter()

    # Group anchors by page.
    by_page: dict[int, list[dict]] = {}
    for a in anchors:
        if signer_field and a.get("field") != signer_field:
            continue
        by_page.setdefault(int(a.get("page", 0)), []).append(a)

    for i, page in enumerate(reader.pages):
        page_anchors = by_page.get(i)
        if page_anchors:
            mb = page.mediabox
            pw = float(mb.width)
            ph = float(mb.height)
            buf = io.BytesIO()
            c = rl_canvas.Canvas(buf, pagesize=(pw, ph))
            c.setStrokeColorRGB(0.06, 0.14, 0.25)
            c.setLineWidth(1.3)
            c.setLineCap(1)
            c.setLineJoin(1)
            for box in page_anchors:
                # Scale anchor from authored page size to this page size.
                sx = pw / (box.get("page_w") or pw)
                sy = ph / (box.get("page_h") or ph)
                scaled = {
                    "x": box["x"] * sx, "y": box["y"] * sy,
                    "w": box["w"] * sx, "h": box["h"] * sy,
                }
                for stroke in fit_strokes_to_box(vector, scaled):
                    if len(stroke) < 2:
                        continue
                    path = c.beginPath()
                    path.moveTo(*stroke[0])
                    for pt in stroke[1:]:
                        path.lineTo(*pt)
                    c.drawPath(path, stroke=1, fill=0)
            c.save()
            buf.seek(0)
            overlay = PdfReader(buf)
            page.merge_page(overlay.pages[0])
        writer.add_page(page)

    out = io.BytesIO()
    writer.write(out)
    return out.getvalue()


if __name__ == "__main__":  # local demo: synth a signature, plan, preview, gcode
    import json
    # Synthetic cursive-ish signature (a few strokes), normalized 0..1.
    def wave(n, x0, x1, y0, amp, ph):
        return [{"x": x0 + (x1 - x0) * k / (n - 1),
                 "y": y0 + amp * math.sin(2 * math.pi * (k / (n - 1) * 2) + ph),
                 "t": k * 8} for k in range(n)]
    vector = {"v": 1, "w": 600, "h": 160, "strokes": [
        wave(40, 0.08, 0.45, 0.55, 0.18, 0.0),
        wave(40, 0.50, 0.92, 0.55, 0.15, 1.2),
    ]}
    box = {"field": "signer", "page": 4, "x": 64.0, "y": 471.0, "w": 402.0, "h": 19.0,
           "page_w": 612.0, "page_h": 792.0}

    cfg = PlotterConfig()
    planned = plan_signature(vector, box, cfg)
    print(f"planned strokes: {len(planned)}; points: {sum(len(s.points_mm) for s in planned)}")
    with open("/tmp/ink_preview.svg", "w") as f:
        f.write(render_preview_svg(planned, cfg))
    gcode = emit_gcode(planned, cfg)
    with open("/tmp/ink_signature.gcode", "w") as f:
        f.write(gcode)
    print("wrote /tmp/ink_preview.svg and /tmp/ink_signature.gcode")
    print("first 12 gcode lines:")
    print("\n".join(gcode.splitlines()[:12]))