j0h

HMI_Listener.py

j0h
Jul 6th, 2026
99
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
Python 2.16 KB | None | 0 0
  1. #!/usr/bin/env python3
  2. """
  3. Listen to the HMI to figure out what registers it wants to initialize.
  4. based on OnTheLine arduino code.
  5. """
  6. import socket
  7. import argparse
  8.  
  9. HOST = "0.0.0.0"
  10. PORT = 2101
  11.  
  12.  
  13. def xor_bcc(data):
  14.     b = 0
  15.     for c in data:
  16.         b ^= c
  17.     return b
  18.  
  19.  
  20. def read_d(addr):
  21.     if addr == 570:
  22.         return 26
  23.  
  24.     if addr in (3498, 3499, 3500):
  25.         return 0
  26.  
  27.     return 0
  28.  
  29.  
  30. def make_reply(addr, nbytes):
  31.     words = nbytes // 2
  32.  
  33.     payload = b"\x06"          # ACK
  34.     payload += b"000"          # device=00, status=0
  35.  
  36.     for i in range(words):
  37.         payload += f"{read_d(addr+i):04X}".encode()
  38.  
  39.     bcc = xor_bcc(payload)
  40.     payload += f"{bcc:02X}".encode()
  41.     payload += b"\r"
  42.  
  43.     return payload
  44.  
  45.  
  46. def handle(sock):
  47.  
  48.     buf = b""
  49.  
  50.     while True:
  51.  
  52.         c = sock.recv(1)
  53.         if not c:
  54.             break
  55.  
  56.         buf += c
  57.  
  58.         if c != b"\r":
  59.             continue
  60.  
  61.         #print("RX:", buf)
  62.         print("RX HEX:  ", " ".join(f"{b:02X}" for b in buf))
  63.         print("RX ASCII:", repr(buf))
  64.  
  65.  
  66.         if len(buf) >= 14 and buf[0] == 0x05:
  67.  
  68.             cmd = buf[4:6]
  69.  
  70.             if cmd == b"RD":
  71.  
  72.                 addr = int(buf[6:10])
  73.                 nbytes = int(buf[10:12], 16)
  74.  
  75.                 print(f"READ D{addr} ({nbytes//2} words)")
  76.  
  77.                 reply = make_reply(addr, nbytes)
  78.  
  79.                 print("TX:", reply)
  80.  
  81.                 sock.sendall(reply)
  82.  
  83.             else:
  84.                 print("Unhandled command:", cmd)
  85.  
  86.         buf = b""
  87.  
  88.  
  89. def main():
  90.  
  91.     ap = argparse.ArgumentParser()
  92.  
  93.     ap.add_argument("--host", default=HOST)
  94.     ap.add_argument("--port", default=PORT, type=int)
  95.  
  96.     args = ap.parse_args()
  97.  
  98.     s = socket.socket()
  99.     s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  100.     s.bind((args.host, args.port))
  101.     s.listen(1)
  102.  
  103.     print(f"Listening on {args.host}:{args.port}")
  104.  
  105.     while True:
  106.  
  107.         c, a = s.accept()
  108.  
  109.         print("Connected:", a)
  110.  
  111.         try:
  112.             handle(c)
  113.         finally:
  114.             c.close()
  115.             print("Disconnected")
  116.  
  117.  
  118. if __name__ == "__main__":
  119.     main()
  120.  
Advertisement
Add Comment
Please, Sign In to add comment