ske087
bff6d70b17
Aplicatie Flask + SQLite pentru productie alimentara: cantarirea componentelor unui produs finit si tiparirea etichetei prin QZ Tray. - produse finite (retete) formate din componente cantarite - designer vizual de etichete (mm/px, zona sigura, campuri variabile, cod QR cu continut din mai multe bucati) - legare produs finit -> sablon de eticheta - navigare pe doua taburi: Cantarire / Setari - administrare: componente, produse finite, sabloane, campuri, istoric - backup, resetare baza de date, checklist de pregatire, /healthz - CLI de administrare si serviciu systemd pentru productie - vendor/syncpulse-scale-bridge (GPL-3.0-or-later) pentru citirea cantarului
194 lines
7.5 KiB
Python
194 lines
7.5 KiB
Python
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Every parser against sample frames in the shapes the manufacturers document."""
|
|
import pytest
|
|
|
|
from scale_bridge.protocols import PROTOCOLS, convert, get_protocol, lines, norm_unit
|
|
from scale_bridge.simulator import frame_for
|
|
|
|
|
|
def one(protocol_id, data, params=None):
|
|
readings = get_protocol(protocol_id, params).readings(data)
|
|
assert readings, f"{protocol_id} read nothing from {data!r}"
|
|
return readings[-1]
|
|
|
|
|
|
def toledo_frame(swa, swb, swc, digits, tare=b"000000"):
|
|
body = bytes([0x02, swa, swb, swc]) + digits + tare + b"\r"
|
|
return body + bytes([(-sum(b & 0x7F for b in body)) & 0x7F])
|
|
|
|
|
|
# -- Mettler Toledo ----------------------------------------------------------
|
|
|
|
def test_mt_sics_stable_and_dynamic():
|
|
r = one("mt_sics", b"S S 0.550 kg\r\n")
|
|
assert (r.value, r.unit, r.stable) == (0.55, "kg", True)
|
|
r = one("mt_sics", b"S D 12.345 g\r\n")
|
|
assert (r.value, r.unit, r.stable) == (12.345, "g", False)
|
|
|
|
|
|
def test_mt_sics_overload_and_error():
|
|
assert one("mt_sics", b"S +\r\n").overload
|
|
assert PROTOCOLS["mt_sics"].readings(b"S I\r\nES\r\n") == []
|
|
|
|
|
|
def test_toledo_continuous_decimal_point_units_and_checksum():
|
|
# SWA: bit5 + count-by x1 (bit3) + dp code 3 (XXXXX.X); SWB: bit5 + kg (bit4)
|
|
r = one("toledo_cont", toledo_frame(0x2B, 0x30, 0x20, b"012345"))
|
|
assert (r.value, r.unit, r.stable, r.net, r.checksum_ok) == (1234.5, "kg", True, False, True)
|
|
# motion + negative + net, grams in SWC
|
|
r = one("toledo_cont", toledo_frame(0x2A, 0x20 | 0x08 | 0x02 | 0x01, 0x21, b"000550"))
|
|
assert (r.value, r.unit, r.stable, r.net) == (-550, "g", False, True)
|
|
# lb when SWB bit4 is clear, dp code 5 (X.XXX)
|
|
r = one("toledo_cont", toledo_frame(0x2D, 0x20, 0x20, b"001210"))
|
|
assert (r.value, r.unit) == (1.21, "lb")
|
|
|
|
|
|
def test_toledo_continuous_bad_checksum_is_flagged():
|
|
frame = bytearray(toledo_frame(0x2B, 0x30, 0x20, b"012345"))
|
|
frame[-1] ^= 0x01
|
|
assert one("toledo_cont", bytes(frame)).checksum_ok is False
|
|
|
|
|
|
# -- A&D and friends -----------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("frame,value,unit,stable,net", [
|
|
(b"ST,+00000.550 kg\r\n", 0.55, "kg", True, None),
|
|
(b"US,-00001.20 kg\r\n", -1.2, "kg", False, None),
|
|
(b"ST,GS,+0012.34kg\r\n", 12.34, "kg", True, False),
|
|
(b"ST,NT,+0001.500 lb\r\n", 1.5, "lb", True, True),
|
|
(b"QT,+000123.4 g\r\n", 123.4, "g", True, None),
|
|
])
|
|
def test_and_standard(frame, value, unit, stable, net):
|
|
r = one("and_std", frame)
|
|
assert (r.value, r.unit, r.stable, r.net) == (value, unit, stable, net)
|
|
|
|
|
|
def test_and_standard_overload():
|
|
r = one("and_std", b"OL,+9999999 kg\r\n")
|
|
assert r.overload and r.value is None
|
|
|
|
|
|
def test_sma_frames():
|
|
r = one("sma", b"\n 1G 0.550kg \r") # s r n m f, then the weight in 10 characters
|
|
assert (r.value, r.unit, r.stable, r.net) == (0.55, "kg", True, False)
|
|
r = one("sma", b"\n 1NM 12.500lb \r")
|
|
assert (r.value, r.unit, r.stable, r.net) == (12.5, "lb", False, True)
|
|
assert one("sma", b"\nO1G ----------lb \r").overload
|
|
|
|
|
|
def test_nci():
|
|
r = one("nci", b"\n 1.23lb\r\n00\r\x03")
|
|
assert (r.value, r.unit, r.stable) == (1.23, "lb", True)
|
|
r = one("nci", b"\n 1.23lb\r\n10\r\x03")
|
|
assert r.stable is False
|
|
|
|
|
|
def test_ohaus():
|
|
r = one("ohaus", b" 0.550 kg \r\n")
|
|
assert (r.value, r.unit, r.stable) == (0.55, "kg", True)
|
|
assert one("ohaus", b" 0.550 kg ? \r\n").stable is False
|
|
assert one("ohaus", b" 1.234 kg N\r\n").net is True
|
|
|
|
|
|
def test_sartorius_unit_missing_means_unstable():
|
|
r = one("sartorius", b"+ 0.550 kg \r\n")
|
|
assert (r.value, r.unit, r.stable) == (0.55, "kg", True)
|
|
r = one("sartorius", b"+ 0.550 \r\n")
|
|
assert (r.value, r.stable) == (0.55, False)
|
|
r = one("sartorius", b"N + 1.234 kg\r\n")
|
|
assert (r.value, r.net) == (1.234, True)
|
|
|
|
|
|
def test_cas_enq():
|
|
r = one("cas_enq", b"\x06\x01\x02S 0.550kg\x03\x04")
|
|
assert (r.value, r.unit, r.stable) == (0.55, "kg", True)
|
|
|
|
|
|
# -- XK3190 ---------------------------------------------------------------------
|
|
|
|
def test_xk3190_tf0_with_checksum():
|
|
body = b"+0012342"
|
|
x = 0
|
|
for b in body:
|
|
x ^= b
|
|
frame = b"\x02" + body + f"{x >> 4:X}{x & 0xF:X}".encode() + b"\x03"
|
|
r = one("xk3190_tf0", frame * 2)
|
|
assert (r.value, r.checksum_ok) == (12.34, True)
|
|
r = one("xk3190_tf0", b"\x02-0005003" + b"00" + b"\x03")
|
|
assert r.value == -0.5 and r.checksum_ok is False
|
|
|
|
|
|
def test_xk3190_reversed_stream():
|
|
r = one("xk3190_reversed", b"05.21000=05.21000=05.21000=")
|
|
assert r.value == 12.5
|
|
assert PROTOCOLS["xk3190_reversed"].readings(b"05.21000=") == []
|
|
|
|
|
|
# -- Mahak (field-tested) -----------------------------------------------------------
|
|
|
|
def test_mahak_field_capture():
|
|
# Captured on a Mahak MDS 13000 with a 550 g weight: 58 ?? 07 07 00 02 26
|
|
r = one("mahak_mds", bytes.fromhex("58 51 07 07 00 02 26"))
|
|
assert (r.value, r.unit) == (550.0, "g")
|
|
r = one("mahak_mds", bytes.fromhex("58 A3 07 07 00 02 EE"))
|
|
assert r.value == 750.0
|
|
|
|
|
|
def test_mahak_ignores_boot_message():
|
|
assert PROTOCOLS["mahak_mds"].readings(bytes.fromhex("42 03 04 01 42 4F 4F 54 0C 04 8C")) == []
|
|
|
|
|
|
# -- generic and custom ----------------------------------------------------------------
|
|
|
|
def test_generic_ascii():
|
|
r = one("ascii_generic", b"G 12.50 kg\r\n")
|
|
assert (r.value, r.unit) == (12.5, "kg")
|
|
assert one("ascii_generic", b"US 3.2 kg\r\n").stable is False
|
|
|
|
|
|
def test_custom_binary_and_ascii():
|
|
params = {"pattern": "58 ?? 07 07", "value_offset": 4, "value_length": 3, "unit": "g"}
|
|
r = one("custom_binary", bytes.fromhex("58 51 07 07 00 02 26"), params)
|
|
assert (r.value, r.unit) == (550.0, "g")
|
|
r = one("custom_ascii", b"W: 12.5\r\n", {"regex": r"W:\s*([-\d.]+)", "unit": "kg"})
|
|
assert (r.value, r.unit) == (12.5, "kg")
|
|
assert get_protocol("custom_ascii").readings(b"W: 1\r\n") == []
|
|
|
|
|
|
# -- the simulator speaks what the parsers read --------------------------------------------
|
|
|
|
@pytest.mark.parametrize("pid", ["mt_sics", "and_std", "toledo_cont", "sma", "nci", "ohaus", "sartorius",
|
|
"cas_enq", "xk3190_tf0", "xk3190_reversed", "mahak_mds"])
|
|
@pytest.mark.parametrize("kg,stable", [(0.55, True), (13.1, False)])
|
|
def test_simulator_round_trip(pid, kg, stable):
|
|
frame = frame_for(pid, kg, stable)
|
|
data = frame * 3 if pid == "xk3190_reversed" else frame
|
|
r = PROTOCOLS[pid].readings(data)[-1]
|
|
expected = kg * 1000 if pid == "mahak_mds" else kg
|
|
assert r.value == pytest.approx(expected)
|
|
if r.stable is not None and pid != "sartorius":
|
|
assert r.stable == stable
|
|
|
|
|
|
# -- helpers ----------------------------------------------------------------------------
|
|
|
|
def test_units():
|
|
assert norm_unit(b" KG ") == "kg" and norm_unit("lbs") == "lb" and norm_unit(None) == ""
|
|
assert convert(550, "g", "kg") == pytest.approx(0.55)
|
|
assert convert(1, "lb", "g") == pytest.approx(453.59237)
|
|
assert convert(1, "kg", "pcs") is None
|
|
|
|
|
|
def test_lines_split_on_terminators():
|
|
assert lines(b"\x02abc\r\ndef\x03\r\n\r\n") == [b"abc", b"def"]
|
|
|
|
|
|
def test_catalogue_is_consistent():
|
|
from scale_bridge.protocols import CONTINUOUS_IDS, POLL_IDS
|
|
for pid in CONTINUOUS_IDS + POLL_IDS:
|
|
assert pid in PROTOCOLS
|
|
for pid in POLL_IDS:
|
|
assert PROTOCOLS[pid].request, pid
|
|
# read-only: never zero / tare / reset / print-off letters on their own
|
|
assert PROTOCOLS[pid].request.strip(b"\r\n") not in (b"Z", b"T", b"R", b"P")
|