admin
4acec81657
- WANT_RESPONSE=True: Anzeige-Antworttelegramm als Handshake, Fehlercode wird ausgewertet (Rueckweg verifiziert: 02 80 81 80 30 03) - transport: Empfangspuffer vor jedem Senden leeren, damit read_response nur die frische Antwort sieht (Stoerbytes zwischen Telegrammen) - run_display: Anzeige-/Antwortfehler abfangen statt abzustuerzen - resp_test.py: Pruefwerkzeug fuer den Rueckkanal - PORT auf /dev/cu.usbserial-120 - .gitignore ergaenzt, __pycache__ aus der Versionierung entfernt Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Antwortleitung testen: Telegramme MIT Antwort-Anforderung senden.
|
|
|
|
Sendet im 1-Sekunden-Takt ein festes Telegramm mit gesetztem Antwort-Bit
|
|
(FC = 0x81). Nach jedem Empfang muss die Anzeige auf Pin 3 ihr Antworttelegramm
|
|
senden (erwartet: 02 80 81 80 30 03). Bricht bei ausbleibender Antwort NICHT ab,
|
|
damit man am Oszi in Ruhe messen kann; liest eine evtl. eintreffende Antwort mit
|
|
und zeigt sie als Hex.
|
|
|
|
python3 resp_test.py # 60 s, 1 s Takt
|
|
python3 resp_test.py --seconds 120
|
|
"""
|
|
|
|
import argparse
|
|
import sys
|
|
import time
|
|
|
|
import config
|
|
import migra
|
|
from transport import SerialTransport
|
|
|
|
|
|
def main(argv=None):
|
|
p = argparse.ArgumentParser(description=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
p.add_argument("--seconds", type=float, default=60.0)
|
|
p.add_argument("--interval", type=float, default=1.0)
|
|
p.add_argument("--port", default=config.PORT)
|
|
args = p.parse_args(argv)
|
|
|
|
du = migra.DataUnit().fill(config.BACKGROUND).charset(0, spaced=True)
|
|
du.cursor(0, 0); du.text("RESP TEST")
|
|
frame = migra.build_telegram(du, dest=config.DISPLAY_ADDR, src=config.HOST_ADDR,
|
|
checksum=False, response=True) # FC-Bit0 = 1
|
|
|
|
t = SerialTransport(port=args.port, baudrate=config.BAUDRATE, parity=config.PARITY,
|
|
bytesize=config.BYTESIZE, stopbits=config.STOPBITS, timeout=0.5)
|
|
|
|
print("Port :", args.port)
|
|
print("TX (%d B): %s" % (len(frame), " ".join("%02X" % b for b in frame)))
|
|
print("FC : 0x%02X (Bit0=1 -> Antwort anfordern)" % frame[3])
|
|
print("Erwartete Antwort auf Pin 3: 02 80 81 80 30 03")
|
|
print("Laufzeit %.0f s, Takt %.1f s. Abbruch mit Ctrl-C.\n" % (args.seconds, args.interval))
|
|
|
|
deadline = time.monotonic() + args.seconds
|
|
n = 0
|
|
got = 0
|
|
try:
|
|
while time.monotonic() < deadline:
|
|
n += 1
|
|
t._port.reset_input_buffer()
|
|
t.write(frame)
|
|
# kurz auf eine Antwort lauschen (blockiert das Oszi nicht)
|
|
end = time.monotonic() + 0.4
|
|
buf = bytearray()
|
|
while time.monotonic() < end:
|
|
c = t._port.read(1)
|
|
if c:
|
|
buf.extend(c)
|
|
if c[0] == migra.ETX:
|
|
break
|
|
elif buf:
|
|
break
|
|
if buf:
|
|
got += 1
|
|
print("#%02d RX: %s" % (n, " ".join("%02X" % b for b in buf)))
|
|
else:
|
|
print("#%02d RX: (nichts empfangen)" % n)
|
|
# bis zum naechsten Sekundentakt warten
|
|
rest = args.interval - 0.4
|
|
if rest > 0:
|
|
time.sleep(rest)
|
|
except KeyboardInterrupt:
|
|
print("\nabgebrochen")
|
|
finally:
|
|
t.close()
|
|
|
|
print("\n%d Telegramme gesendet, %d Antworten empfangen." % (n, got))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|