cantarire-etichetare/cantarire/probe.py

"""Monitor de port serial: ce trimite de fapt cantarul.

Instrument pentru un model nou (ex. Bizerba BC800): deschide portul in mod
ascultare (nu scrie nimic catre cantar), strange cadrele brute si le trece
prin toate protocoalele cunoscute. Rezultatul spune daca formatul este deja
suportat sau daca trebuie adaugat un protocol nou.
"""
from __future__ import annotations

import re
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import Any, Optional

VENDOR_DIR = Path(__file__).resolve().parent.parent / "vendor" / "syncpulse-scale-bridge"

BAUD_RATES = (9600, 4800, 19200, 2400, 38400, 1200, 57600, 115200)
FRAMINGS = ("8N1", "7E1", "7O1", "8E1", "8O1")
MAX_FRAMES = 60


class ProbeError(RuntimeError):
    """Portul nu poate fi deschis sau setarile sunt invalide."""


def _bridge():
    path = str(VENDOR_DIR)
    if path not in sys.path:
        sys.path.insert(0, path)
    from scale_bridge import serialio  # noqa: PLC0415

    return serialio


def valid_framing(text: str) -> str:
    value = (text or "").strip().upper()
    if not re.fullmatch(r"[5678][NEOMS][12]", value):
        raise ProbeError(f"framing invalid: {text!r} (ex. 8N1, 7E1)")
    return value


def list_ports() -> list[dict[str, Any]]:
    serialio = _bridge()
    return serialio.list_serial_ports()


def decode(data: bytes) -> list[dict[str, Any]]:
    """Ce protocoale din biblioteca recunosc datele primite."""
    _bridge()
    from scale_bridge.protocols import PROTOCOLS  # noqa: PLC0415

    guesses: list[dict[str, Any]] = []
    if not data:
        return guesses
    for pid, proto in PROTOCOLS.items():
        if pid.startswith("custom"):
            continue
        try:
            readings = [r for r in proto.readings(data) if r.value is not None]
        except Exception:  # noqa: BLE001 - un parser nu trebuie sa rupa monitorul
            continue
        if not readings:
            continue
        last = readings[-1]
        guesses.append({
            "protocol": pid,
            "name": proto.name,
            "kind": proto.kind,
            "count": len(readings),
            "value": last.value,
            "unit": last.unit,
            "stable": last.stable,
            "specificity": proto.specificity,
        })
    guesses.sort(key=lambda g: (-g["specificity"], -g["count"]))
    return guesses


def probe_port(
    port: str,
    baud: int = 9600,
    framing: str = "8N1",
    seconds: float = 5.0,
    opener: Optional[Any] = None,
) -> dict[str, Any]:
    """Asculta portul `seconds` secunde si intoarce cadrele brute."""
    serialio = _bridge()
    framing = valid_framing(framing)
    if not port:
        raise ProbeError("Alege portul serial.")
    if not 50 <= int(baud) <= 921600:
        raise ProbeError(f"baud invalid: {baud}")
    seconds = min(max(float(seconds), 0.5), 60.0)

    serial_cfg = {
        "port": port,
        "baudrate": int(baud),
        "bytesize": int(framing[0]),
        "parity": framing[1],
        "stopbits": int(framing[2]),
    }
    try:
        settings = serialio.LineSettings.from_dicts(serial_cfg, {})
    except ValueError as exc:
        raise ProbeError(str(exc)) from exc

    worker = serialio.PortWorker(settings, opener or serialio.open_real)
    worker.start()
    deadline = time.monotonic() + seconds
    try:
        while time.monotonic() < deadline:
            if worker.last_error and not worker.is_open:
                break
            time.sleep(0.05)
    finally:
        worker.stop()
        worker.join(timeout=3)

    chunks = worker.recent(MAX_FRAMES)
    joined = b"".join(data for _, data in chunks)
    now = serialio.now_ms()
    return {
        "ok": bool(chunks) or bool(worker.is_open),
        "port": settings.port,
        "settings": settings.describe(),
        "baud": int(baud),
        "framing": framing,
        "seconds": seconds,
        "error": None if (chunks or worker.is_open) else worker.last_error,
        "bytes": len(joined),
        "frames": [
            {"age_ms": max(0, now - t), "hex": serialio.format_hex(data),
             "text": serialio.printable(data), "size": len(data)}
            for t, data in chunks
        ],
        "guesses": decode(joined),
        "raw": joined,
    }


def sweep(
    port: str,
    seconds: float = 2.0,
    framings: tuple[str, ...] = ("8N1", "7E1"),
    bauds: tuple[int, ...] = BAUD_RATES,
    opener: Optional[Any] = None,
) -> list[dict[str, Any]]:
    """Incearca mai multe setari de linie si pastreaza doar cele cu date."""
    results = []
    for framing in framings:
        for baud in bauds:
            try:
                result = probe_port(port, baud, framing, seconds, opener)
            except ProbeError as exc:
                results.append({"baud": baud, "framing": framing, "ok": False,
                                "error": str(exc), "bytes": 0, "guesses": []})
                continue
            results.append({
                "baud": baud,
                "framing": framing,
                "ok": result["ok"],
                "error": result["error"],
                "bytes": result["bytes"],
                "frames": len(result["frames"]),
                "guesses": result["guesses"],
                "raw": result["raw"],
            })
            if result["bytes"] and result["guesses"]:
                # Un protocol recunoscut este un rezultat bun: nu mai insistam.
                pass
    return results


def save_capture(folder: str | Path, name: str, frames: list[dict[str, Any]]) -> Path:
    """Scrie cadrele ca hex, un cadru pe linie (formatul din tests/captures)."""
    safe = re.sub(r"[^A-Za-z0-9_-]+", "-", str(name or "capture")).strip("-") or "capture"
    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    target = Path(folder) / f"{safe}-{stamp}.hex"
    target.parent.mkdir(parents=True, exist_ok=True)
    lines = [f["hex"] for f in frames if f.get("hex")]
    target.write_text("\n".join(lines) + "\n", encoding="ascii")
    return target