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
113 lines
3.6 KiB
Python
113 lines
3.6 KiB
Python
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Auto-detection against simulated scales that only understand their own settings."""
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from scale_bridge.detect import Detector, Timings, readability
|
|
from scale_bridge.simulator import SimulatorBank
|
|
|
|
FAST = Timings(passive_s=0.5, probe_s=0.15, confirm_s=0.4, drain_s=0.02)
|
|
|
|
|
|
class Silent:
|
|
"""A port with nothing connected."""
|
|
in_waiting = 0
|
|
|
|
def __init__(self, settings):
|
|
self.settings = settings
|
|
|
|
def read(self, n=1):
|
|
time.sleep(0.01)
|
|
return b""
|
|
|
|
def write(self, data):
|
|
return len(data)
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class Chatter(Silent):
|
|
"""Readable text without any weight in it."""
|
|
|
|
def read(self, n=1):
|
|
time.sleep(0.01)
|
|
return b"HELLO WORLD\r\n" if self.settings.baudrate == 9600 else b""
|
|
|
|
@property
|
|
def in_waiting(self):
|
|
return 0
|
|
|
|
|
|
def detect(spec, port="COM90", thorough=False):
|
|
bank = SimulatorBank(spec)
|
|
events = []
|
|
result = Detector(bank.opener(), log=lambda code, **kw: events.append(code), thorough=thorough,
|
|
timings=FAST).run_port(port)
|
|
return result, events
|
|
|
|
|
|
@pytest.mark.parametrize("spec,protocol,baud,framing,mode", [
|
|
("mahak_mds", "mahak_mds", 9600, "8N1", "poll"),
|
|
("toledo_cont:4800:7E1", "toledo_cont", 4800, "7E1", "continuous"),
|
|
("xk3190_tf0:2400", "xk3190_tf0", 2400, "8N1", "continuous"),
|
|
("xk3190_reversed:1200", "xk3190_reversed", 1200, "8N1", "continuous"),
|
|
("mt_sics", "mt_sics", 9600, "8N1", "poll"),
|
|
("and_std:19200", "and_std", 19200, "8N1", "poll"),
|
|
("sma:9600:7E1", "sma", 9600, "7E1", "poll"),
|
|
("sartorius:1200:7E1", "sartorius", 1200, "7E1", "poll"),
|
|
("cas_enq", "cas_enq", 9600, "8N1", "poll"),
|
|
])
|
|
def test_finds_protocol_speed_and_framing(spec, protocol, baud, framing, mode):
|
|
result, events = detect(spec)
|
|
assert result["found"], events
|
|
assert result["protocol"] == protocol
|
|
assert result["serial"]["baudrate"] == baud
|
|
s = result["serial"]
|
|
assert f"{s['bytesize']}{s['parity']}{s['stopbits']}" == framing
|
|
assert result["mode"] == mode
|
|
assert 0 < result["confidence"] <= 0.99
|
|
assert events[0] == "port_start" and "found" in events
|
|
|
|
|
|
def test_thorough_search_reaches_7o1():
|
|
result, _ = detect("ohaus:9600:7O1", thorough=True)
|
|
assert result["found"] and result["protocol"] == "ohaus"
|
|
assert result["serial"]["parity"] == "O"
|
|
|
|
|
|
def test_nothing_connected():
|
|
detector = Detector(lambda s: Silent(s), timings=FAST)
|
|
result = detector.run_port("COM5")
|
|
assert result["found"] is False and result["unknown"] is None
|
|
|
|
|
|
def test_unknown_readable_data_is_reported():
|
|
detector = Detector(lambda s: Chatter(s), timings=FAST)
|
|
result = detector.run_port("COM5")
|
|
assert result["found"] is False
|
|
assert result["unknown"]["baud"] == 9600
|
|
assert "HELLO" in result["unknown"]["sample_text"]
|
|
|
|
|
|
def test_cancel_stops_quickly():
|
|
detector = Detector(lambda s: Silent(s), cancelled=lambda: True, timings=FAST)
|
|
started = time.monotonic()
|
|
result = detector.run_port("COM5")
|
|
assert time.monotonic() - started < 0.5
|
|
assert result["found"] is False and result["cancelled"] is True
|
|
|
|
|
|
def test_open_failure_is_reported():
|
|
def refuse(settings):
|
|
raise OSError("Access is denied")
|
|
result = Detector(refuse, timings=FAST).run_port("COM5")
|
|
assert result["found"] is False and "Access is denied" in result["error"]
|
|
|
|
|
|
def test_readability():
|
|
assert readability(b"ST,+0.55 kg\r\n") == 1.0
|
|
assert readability(bytes([0xD3, 0xD4])) == 0.0
|
|
assert readability(bytes([0xD3, 0xD4]), mask7=True) == 1.0
|