cantarire-etichetare/vendor/syncpulse-scale-bridge/scale_bridge/simulator.py

# SPDX-License-Identifier: GPL-3.0-or-later
# Copyright (C) 2026 Farshad Zamani — Cloudtart (https://cloudtart.com)
"""
Simulated scales — every protocol in the library, on a fake COM port.

For trying the program without hardware and for the tests. A simulated
device only understands the speed and framing it was configured with,
exactly like a real one: at another speed a continuous device produces
garbage and a polled one stays silent; a 7E1 device read as 8N1 shows its
parity bit in the top bit of each byte.

    --simulate mahak_mds                       COM90, 9600 8N1
    --simulate toledo_cont:4800:7E1,mt_sics    COM90 and COM91
"""
from __future__ import annotations

import time
from typing import Optional

from .serialio import LineSettings, Opener, open_real

FIRST_PORT = 90

# The weight walks through an empty scale, 550 g, a ~13 kg package that
# settles, and 750 g — (seconds, [kg values], stable).
SCHEDULE = [(6, [0.0], True), (8, [0.55], True), (3, [12.85, 13.15, 13.10], False),
            (8, [13.05], True), (6, [0.75], True)]


def weight_now(t0: float) -> tuple[float, bool]:
    total = sum(s for s, _, _ in SCHEDULE)
    t = (time.monotonic() - t0) % total
    for seconds, values, stable in SCHEDULE:
        if t < seconds:
            return values[int(t * 3) % len(values)], stable
        t -= seconds
    return 0.0, True


def _even_parity_bit(b: int) -> int:
    return (bin(b & 0x7F).count("1") & 1) << 7


def frame_for(protocol: str, kg: float, stable: bool) -> bytes:
    """What `protocol` sends for `kg` — the same shapes the parsers read."""
    if protocol == "mt_sics":
        return f"S {'S' if stable else 'D'} {kg:10.3f} kg\r\n".encode()
    if protocol == "and_std":
        return f"{'ST' if stable else 'US'},GS,{kg:+09.3f}kg\r\n".encode()
    if protocol == "toledo_cont":
        digits = f"{round(abs(kg) * 1000):06d}".encode()
        swa = 0x20 | 0x08 | 5                   # count-by x1, 0.00X
        swb = 0x20 | 0x10 | (0 if stable else 0x08) | (0x02 if kg < 0 else 0)
        swc = 0x20
        body = bytes([0x02, swa, swb, swc]) + digits + b"000000" + b"\r"
        return body + bytes([(-sum(b & 0x7F for b in body)) & 0x7F])
    if protocol == "sma":
        return b"\n 1G" + (b" " if stable else b"M") + b" " + f"{kg:10.3f}".encode() + b"kg \r"
    if protocol == "nci":
        return b"\n" + f"{kg:8.3f}kg".encode() + b"\r\n" + (b"0" if stable else b"1") + b"0\r\x03"
    if protocol == "ohaus":
        return f"{kg:10.3f} kg{'' if stable else ' ?'}\r\n".encode()
    if protocol == "sartorius":
        return f"+{kg:10.3f} {'kg' if stable else '  '}\r\n".encode()
    if protocol == "cas_enq":
        return b"\x01\x02" + (b"S" if stable else b"U") + f" {kg:7.3f}kg".encode() + b"\x03\x04"
    if protocol == "xk3190_tf0":
        body = (b"-" if kg < 0 else b"+") + f"{round(abs(kg) * 1000):06d}".encode() + b"3"
        x = 0
        for b in body:
            x ^= b
        return b"\x02" + body + f"{x >> 4:X}{x & 0xF:X}".encode() + b"\x03"
    if protocol == "xk3190_reversed":
        return f"{kg:08.3f}"[::-1].encode() + b"="
    if protocol == "label_print":
        # What an indicator wired to a TSC label printer sends on Print (TSPL).
        w = f"{kg:7.3f}"
        return ('CLS\r\nSIZE 57 mm, 40 mm\r\nGAP 2 mm,0\r\nDIRECTION 1\r\nREFERENCE 0,0\r\n'
                'CODEPAGE IranII\r\nTEXT 300,250,"10",270,1,1,"No: 6 SHIFT:2"\r\n'
                'TEXT 030,235,"10",0,1,2,"Order.Co :   222\r\n'
                'TEXT 030,080,"10",0,1,2,"1405/07/19 - 16:44"\r\n'
                'TEXT 030,160,"10",0,1,2,"Thickness : 0.0"\r\n'
                f'TEXT 020,000,"10",0,2,2,"{w}"\r\nTEXT 200,015,"10",0,2,1,"(Kg)"\r\n'
                f'BARCODE 350,280, "128",070,1,270,2,2, "  222\t{w}"\r\nPRINT 1\r\n\r\n').encode()
    if protocol == "mahak_mds":
        grams = round(kg * 1000)
        return bytes([0x58, (grams * 7 + 0x51) & 0xFF, 0x07, 0x07]) + grams.to_bytes(3, "big")
    return f"G  {kg:.3f} kg\r\n".encode()


class SimulatedDevice:
    """Quacks like a pyserial Serial for what the bridge uses."""

    CONTINUOUS = {"toledo_cont", "xk3190_tf0", "xk3190_reversed", "ascii_generic"}
    # Sends only when somebody presses Print — here every PRINT_EVERY_S seconds.
    ON_PRINT = {"label_print"}
    PRINT_EVERY_S = 3.0
    REQUESTS = {"mt_sics": b"SI\r\n", "and_std": b"Q\r\n", "sma": b"\nW\r", "nci": b"W\r",
                "ohaus": b"IP\r\n", "sartorius": b"\x1bP\r\n", "mahak_mds": bytes.fromhex("4D090407"),
                "cas_enq": b"\x05"}
    BOOT = bytes.fromhex("42 03 04 01 42 4F 4F 54 0C 04 8C")

    def __init__(self, protocol: str, baudrate: int, framing: str, opened: LineSettings):
        self.protocol = protocol
        self.device_baud = baudrate
        self.device_framing = framing
        self.opened = opened
        self.is_open = True
        self._t0 = time.monotonic()
        self._out = bytearray()
        self._due: list[tuple[float, bytes]] = []
        self._next_emit = time.monotonic() + (self.PRINT_EVERY_S if protocol in self.ON_PRINT else 0)
        self._next_boot = time.monotonic() + 60
        self._garbage_seed = 0x5A

    # -- line matching -----------------------------------------------------
    @property
    def _baud_ok(self) -> bool:
        return self.opened.baudrate == self.device_baud

    @property
    def _framing_ok(self) -> bool:
        return self.opened.framing == self.device_framing

    def _as_received(self, data: bytes) -> bytes:
        """What the host's UART makes of the device's bytes."""
        if not self._baud_ok:
            out = bytearray()
            for _ in range(max(1, len(data) // 3)):
                self._garbage_seed = (self._garbage_seed * 73 + 41) & 0xFF
                out.append(self._garbage_seed | 0x80)
            return bytes(out)
        if self._framing_ok:
            return data
        if self.device_framing.startswith("7") and self.opened.bytesize == 8:
            return bytes((b & 0x7F) | _even_parity_bit(b) for b in data)
        if self.device_framing.startswith("8") and self.opened.bytesize == 7:
            return bytes(b & 0x7F for b in data)
        return data

    # -- pyserial surface --------------------------------------------------
    def _pump(self) -> None:
        now = time.monotonic()
        if self.protocol in self.CONTINUOUS and now >= self._next_emit:
            kg, stable = weight_now(self._t0)
            self._out += self._as_received(frame_for(self.protocol, kg, stable))
            self._next_emit = now + 0.2
        if self.protocol in self.ON_PRINT and now >= self._next_emit:
            kg, _ = weight_now(self._t0)
            self._out += self._as_received(frame_for(self.protocol, kg, True))
            self._next_emit = now + self.PRINT_EVERY_S
        if self.protocol == "mahak_mds" and now >= self._next_boot:
            self._out += self._as_received(self.BOOT)
            self._next_boot = now + 60
        ready = [d for d in self._due if d[0] <= now]
        self._due = [d for d in self._due if d[0] > now]
        for _, data in ready:
            self._out += self._as_received(data)

    def write(self, data: bytes) -> int:
        if not (self._baud_ok and self._framing_ok):
            return len(data)
        data = bytes(data)
        request = self.REQUESTS.get(self.protocol)
        kg, stable = weight_now(self._t0)
        if self.protocol == "cas_enq":
            if b"\x05" in data:
                self._due.append((time.monotonic() + 0.03, b"\x06"))
            if b"\x11" in data:
                self._due.append((time.monotonic() + 0.04, frame_for("cas_enq", kg, stable)))
        elif request and request in data:
            self._due.append((time.monotonic() + 0.04, frame_for(self.protocol, kg, stable)))
        return len(data)

    @property
    def in_waiting(self) -> int:
        self._pump()
        return len(self._out)

    def read(self, n: int = 1) -> bytes:
        self._pump()
        if not self._out:
            time.sleep(0.02)
            self._pump()
        chunk = bytes(self._out[:n])
        del self._out[:n]
        return chunk

    def reset_input_buffer(self) -> None:
        self._out.clear()

    def close(self) -> None:
        self.is_open = False


class SimulatorBank:
    """Fake ports COM90, COM91 …; every other port opens for real."""

    def __init__(self, spec: str):
        self.devices: dict[str, tuple[str, int, str]] = {}
        for i, item in enumerate(filter(None, (s.strip() for s in spec.split(",")))):
            parts = item.split(":")
            protocol = parts[0]
            baud = int(parts[1]) if len(parts) > 1 and parts[1] else 9600
            framing = (parts[2] if len(parts) > 2 and parts[2] else "8N1").upper()
            self.devices[f"COM{FIRST_PORT + i}"] = (protocol, baud, framing)

    def opener(self, fallback: Optional[Opener] = None) -> Opener:
        real = fallback or open_real

        def _open(settings: LineSettings):
            device = self.devices.get(settings.port.upper())
            if device is None:
                return real(settings)
            return SimulatedDevice(device[0], device[1], device[2], settings)
        return _open

    def ports(self) -> list[dict]:
        return [{"device": port, "description": f"Simulated scale ({proto}, {baud} {framing})",
                 "hwid": "SIMULATED", "vid": None, "pid": None, "manufacturer": "simulator",
                 "chip": None, "note": "", "kind": "simulated"}
                for port, (proto, baud, framing) in self.devices.items()]