free_plot_server.py

#!/usr/bin/env python3
from __future__ import annotations

import glob
import json
import mimetypes
import os
import re
import shlex
import subprocess
import tempfile
from dataclasses import dataclass
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import unquote


ROOT = Path(__file__).parent.resolve()
WEB = ROOT / "web"
HOST = "127.0.0.1"
PORT = 8787
USB_HELPER = ROOT / "usb_direct.py"
VENV_PYTHON = ROOT / ".venv" / "bin" / "python"


@dataclass
class SerialPort:
    path: str
    label: str
    kind: str
    likely_usb: bool


class Handler(BaseHTTPRequestHandler):
    def do_GET(self) -> None:
        if self.path == "/api/ports":
            self.send_json(scan_ports())
            return

        requested = unquote(self.path.split("?", 1)[0])
        if requested == "/":
            requested = "/index.html"
        file_path = (WEB / requested.lstrip("/")).resolve()
        if not str(file_path).startswith(str(WEB)) or not file_path.is_file():
            self.send_error(404)
            return

        content_type = mimetypes.guess_type(file_path.name)[0] or "application/octet-stream"
        self.send_response(200)
        self.send_header("Content-Type", content_type)
        self.send_header("Cache-Control", "no-store, max-age=0")
        self.end_headers()
        self.wfile.write(file_path.read_bytes())

    def do_POST(self) -> None:
        if self.path == "/api/send":
            self.handle_serial_send()
            return
        if self.path == "/api/send-usb":
            self.handle_usb_send()
            return
        if self.path == "/api/probe-usb":
            self.handle_usb_probe()
            return
        self.send_error(404)

    def handle_serial_send(self) -> None:
        try:
            length = int(self.headers.get("Content-Length", "0"))
            payload = json.loads(self.rfile.read(length).decode("utf-8"))
            port = str(payload.get("port", ""))
            baud = str(payload.get("baud", "9600"))
            hpgl = str(payload.get("hpgl", ""))

            if not port.startswith("/dev/"):
                raise ValueError("Choose a valid serial port.")
            if baud not in {"9600", "19200", "38400", "57600", "115200"}:
                raise ValueError("Choose a supported baud rate.")
            if not hpgl:
                raise ValueError("No HPGL data was provided.")

            subprocess.run(["stty", "-f", port, baud, "cs8", "-cstopb", "-parenb"], check=False)
            with open(port, "wb", buffering=0) as handle:
                handle.write(hpgl.encode("ascii"))
            self.send_json({"ok": True})
        except Exception as exc:
            self.send_json({"ok": False, "error": str(exc)}, status=400)

    def handle_usb_send(self) -> None:
        try:
            length = int(self.headers.get("Content-Length", "0"))
            payload = json.loads(self.rfile.read(length).decode("utf-8"))
            vendor_id = int(str(payload.get("vendorId", "0")), 0)
            product_id = int(str(payload.get("productId", "0")), 0)
            cups_uri = str(payload.get("cupsUri", ""))
            hpgl = str(payload.get("hpgl", ""))

            if not hpgl:
                raise ValueError("No HPGL data was provided.")
            if cups_uri:
                send_cups_usb(cups_uri, hpgl)
                self.send_json({"ok": True, "method": "cups-usb"})
                return
            if not vendor_id or not product_id:
                raise ValueError("Choose a valid USB cutter.")
            if not VENV_PYTHON.exists():
                raise ValueError("The local USB helper environment is missing. Run: python3 -m venv .venv && .venv/bin/python -m pip install pyusb libusb1")
            if not USB_HELPER.exists():
                raise ValueError("The USB helper script is missing.")

            result = subprocess.run(
                [
                    str(VENV_PYTHON),
                    str(USB_HELPER),
                    "--vendor",
                    hex(vendor_id),
                    "--product",
                    hex(product_id),
                ],
                input=hpgl.encode("ascii"),
                capture_output=True,
                timeout=15,
            )
            if result.returncode != 0:
                error = result.stderr.decode("utf-8", "replace").strip() or result.stdout.decode("utf-8", "replace").strip()
                raise ValueError(error or "Direct USB send failed.")
            self.send_json({"ok": True})
        except Exception as exc:
            self.send_json({"ok": False, "error": str(exc)}, status=400)

    def handle_usb_probe(self) -> None:
        try:
            length = int(self.headers.get("Content-Length", "0"))
            payload = json.loads(self.rfile.read(length).decode("utf-8"))
            vendor_id = int(str(payload.get("vendorId", "0")), 0)
            product_id = int(str(payload.get("productId", "0")), 0)
            cups_uri = str(payload.get("cupsUri", ""))

            if cups_uri:
                self.send_json({"ok": True, "message": "CUPS USB cutter is available."})
                return
            if not vendor_id or not product_id:
                raise ValueError("Choose a valid USB cutter.")
            if not VENV_PYTHON.exists():
                raise ValueError("The local USB helper environment is missing. Run: python3 -m venv .venv && .venv/bin/python -m pip install pyusb libusb1")

            result = subprocess.run(
                [
                    str(VENV_PYTHON),
                    str(USB_HELPER),
                    "--vendor",
                    hex(vendor_id),
                    "--product",
                    hex(product_id),
                    "--probe",
                ],
                capture_output=True,
                timeout=15,
            )
            stdout = result.stdout.decode("utf-8", "replace").strip()
            stderr = result.stderr.decode("utf-8", "replace").strip()
            if result.returncode != 0:
                raise ValueError(stderr or stdout or "USB probe failed.")
            self.send_json({"ok": True, "message": stdout or "USB cutter is accessible."})
        except Exception as exc:
            self.send_json({"ok": False, "error": str(exc)}, status=400)

    def send_json(self, payload: object, status: int = 200) -> None:
        data = json.dumps(payload).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def log_message(self, format: str, *args: object) -> None:
        print(f"{self.address_string()} - {format % args}")


def scan_ports() -> dict[str, object]:
    ports = list_serial_ports()
    return {
        "ports": [port.__dict__ for port in ports],
        "usbDevices": merge_usb_devices(list_usb_devices(), list_cups_usb_devices()),
    }


def list_serial_ports() -> list[SerialPort]:
    ports = sorted(set(glob.glob("/dev/cu.*") + glob.glob("/dev/tty.*")))
    serial_ports = [describe_port(port) for port in ports]
    return sorted(serial_ports, key=lambda port: (not port.likely_usb, port.kind != "callout", port.path.lower()))


def describe_port(path: str) -> SerialPort:
    name = os.path.basename(path)
    lower = name.lower()
    likely_usb = any(token in lower for token in ["usb", "wch", "ch34", "serial", "ftdi", "silabs", "modem"])
    kind = "callout" if path.startswith("/dev/cu.") else "terminal"
    label_parts = [path]
    if likely_usb:
        label_parts.append("likely USB")
    if kind == "callout":
        label_parts.append("preferred")
    return SerialPort(path=path, label=" - ".join(label_parts), kind=kind, likely_usb=likely_usb)


def list_usb_devices() -> list[dict[str, str]]:
    devices = list_usb_devices_system_profiler()
    if devices:
        return devices
    return list_usb_devices_ioreg()


def list_cups_usb_devices() -> list[dict[str, str]]:
    try:
        result = subprocess.run(
            ["/usr/libexec/cups/backend/usb"],
            check=False,
            capture_output=True,
            text=True,
            timeout=8,
        )
        devices: list[dict[str, str]] = []
        for line in result.stdout.splitlines():
            parts = shlex.split(line)
            if len(parts) < 4 or parts[0] != "direct" or not parts[1].startswith("usb://"):
                continue
            device_id = parts[4] if len(parts) > 4 else ""
            vendor = parse_device_id(device_id, "MFG")
            model = parse_device_id(device_id, "MDL") or parts[2]
            serial = parse_device_id(device_id, "SERN")
            devices.append(
                mark_usb_device(
                    {
                        "name": model.strip(),
                        "vendor": vendor.strip(),
                        "serial": serial.strip(),
                        "cupsUri": parts[1],
                        "likelySerial": True,
                    }
                )
            )
        return devices
    except Exception:
        return []


def merge_usb_devices(raw_devices: list[dict[str, str]], cups_devices: list[dict[str, str]]) -> list[dict[str, str]]:
    merged = raw_devices[:]
    for cups_device in cups_devices:
        matched = False
        for raw_device in merged:
            same_serial = cups_device.get("serial") and cups_device.get("serial") == raw_device.get("serial")
            same_name = cups_device.get("name") and cups_device.get("name") == raw_device.get("name")
            if same_serial or same_name:
                raw_device.update({key: value for key, value in cups_device.items() if value})
                raw_device["likelySerial"] = True
                matched = True
                break
        if not matched:
            merged.append(cups_device)
    return merged


def list_usb_devices_system_profiler() -> list[dict[str, str]]:
    try:
        result = subprocess.run(
            ["system_profiler", "SPUSBDataType", "-json"],
            check=False,
            capture_output=True,
            text=True,
            timeout=8,
        )
        if result.returncode != 0:
            return []
        payload = json.loads(result.stdout)
        devices: list[dict[str, str]] = []
        for root in payload.get("SPUSBDataType", []):
            collect_usb_items(root, devices)
        return devices
    except Exception:
        return []


def list_usb_devices_ioreg() -> list[dict[str, str]]:
    try:
        result = subprocess.run(
            ["ioreg", "-p", "IOUSB", "-l", "-w", "0"],
            check=False,
            capture_output=True,
            text=True,
            timeout=8,
        )
        if result.returncode != 0:
            return []

        devices: list[dict[str, str]] = []
        current: dict[str, str] | None = None
        for line in result.stdout.splitlines():
            match = re.search(r"\+-o\s+(.+?)@", line)
            if match:
                if current and current.get("name"):
                    devices.append(mark_usb_device(current))
                current = {"name": match.group(1).strip()}
                continue

            if current is None:
                continue

            for key, output_key in [
                ("USB Product Name", "name"),
                ("kUSBProductString", "name"),
                ("USB Vendor Name", "vendor"),
                ("kUSBVendorString", "vendor"),
                ("USB Serial Number", "serial"),
                ("kUSBSerialNumberString", "serial"),
            ]:
                value = quoted_ioreg_value(line, key)
                if value:
                    current[output_key] = value

            for key, output_key in [("idProduct", "productId"), ("idVendor", "vendorId")]:
                value = integer_ioreg_value(line, key)
                if value:
                    current[output_key] = value

        if current and current.get("name"):
            devices.append(mark_usb_device(current))
        return devices
    except Exception:
        return []


def collect_usb_items(item: dict[str, object], devices: list[dict[str, str]]) -> None:
    name = str(item.get("_name", "")).strip()
    vendor = str(item.get("manufacturer", item.get("vendor_id", ""))).strip()
    product_id = str(item.get("product_id", "")).strip()
    vendor_id = str(item.get("vendor_id", "")).strip()
    if name:
        devices.append(
            {
                "name": name,
                "vendor": vendor,
                "productId": product_id,
                "vendorId": vendor_id,
                "likelySerial": any(
                    token in " ".join([name, vendor, product_id, vendor_id]).lower()
                    for token in ["serial", "ch34", "wch", "ftdi", "silicon", "usb2.0-ser"]
                ),
            }
        )
    for child in item.get("_items", []) or []:
        if isinstance(child, dict):
            collect_usb_items(child, devices)


def mark_usb_device(device: dict[str, str]) -> dict[str, str]:
    text = " ".join(
        [
            device.get("name", ""),
            device.get("vendor", ""),
            device.get("serial", ""),
            device.get("productId", ""),
            device.get("vendorId", ""),
        ]
    ).lower()
    device["likelySerial"] = any(
        token in text
        for token in ["serial", "ch34", "wch", "ftdi", "silicon", "usb2.0-ser", "stm32", "xili"]
    )
    return device


def parse_device_id(device_id: str, key: str) -> str:
    match = re.search(rf"(?:^|;){re.escape(key)}:([^;]*)", device_id)
    return match.group(1) if match else ""


def send_cups_usb(cups_uri: str, hpgl: str) -> None:
    if not cups_uri.startswith("usb://"):
        raise ValueError("Invalid USB printer URI.")
    with tempfile.NamedTemporaryFile(delete=False, suffix=".hpgl") as handle:
        handle.write(hpgl.encode("ascii"))
        path = handle.name
    env = os.environ.copy()
    env["DEVICE_URI"] = cups_uri
    try:
        result = subprocess.run(
            ["/usr/libexec/cups/backend/usb", "1", "freeplot", "FreePlot", "1", "", path],
            env=env,
            capture_output=True,
            timeout=20,
        )
        if result.returncode != 0:
            stderr = result.stderr.decode("utf-8", "replace").strip()
            stdout = result.stdout.decode("utf-8", "replace").strip()
            raise ValueError(stderr or stdout or "CUPS USB backend failed.")
    finally:
        try:
            os.unlink(path)
        except OSError:
            pass


def quoted_ioreg_value(line: str, key: str) -> str:
    match = re.search(rf'"{re.escape(key)}"\s*=\s*"([^"]+)"', line)
    return match.group(1).strip() if match else ""


def integer_ioreg_value(line: str, key: str) -> str:
    match = re.search(rf'"{re.escape(key)}"\s*=\s*(\d+)', line)
    return match.group(1).strip() if match else ""


def main() -> None:
    os.chdir(ROOT)
    server = ThreadingHTTPServer((HOST, PORT), Handler)
    print(f"Free Plot running at http://{HOST}:{PORT}")
    print("Press Ctrl+C to stop.")
    server.serve_forever()


if __name__ == "__main__":
    main()