#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 远程显示查看器 - 带 GUI 的实时显存查看与按键模拟 用法: python remo_disp_viewer.py <装置IP> """ import socket import struct import threading import time import sys import tkinter as tk from tkinter import ttk, messagebox from typing import Optional, Tuple # 协议常量 TAG_CLIENT = 0xAA TAG_DEVICE = 0xBB PORT = 7003 CMD_KEEPLIVE = 0 CMD_INIT = 1 CMD_KEY = 2 CMD_LCDMEM = 3 KEY_U, KEY_D, KEY_L, KEY_R = 0x02, 0x40, 0x10, 0x08 KEY_ENT, KEY_ESC, KEY_F1, KEY_F2 = 0x20, 0x01, 0x04, 0x80 def calc_crc(data: bytes) -> int: crc = 0 for b in data: crc ^= b return crc & 0xFF def build_frame(cmd: int, data: bytes) -> bytes: length = len(data) header = bytes([TAG_CLIENT, cmd & 0xFF, (length >> 8) & 0xFF, length & 0xFF]) return header + data + bytes([calc_crc(data)]) def parse_frame(raw: bytes) -> Optional[Tuple[int, bytes]]: if len(raw) < 6 or raw[0] != TAG_DEVICE: return None length = (raw[2] << 8) | raw[3] if len(raw) < 5 + length + 1: return None data = raw[4:4 + length] if calc_crc(data) != raw[4 + length]: return None return (raw[1], data) class RemoDispViewer: def __init__(self, host: str, port: int = PORT): self.host = host self.port = port self._sock: Optional[socket.socket] = None self._init_info: Optional[dict] = None self._running = False self._refresh_thread: Optional[threading.Thread] = None def connect(self) -> bool: try: self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self._sock.settimeout(3.0) self._sock.connect((self.host, self.port)) self._sock.settimeout(0.5) return True except Exception: return False def disconnect(self): self._running = False if self._sock: try: self._sock.close() except Exception: pass self._sock = None def _send(self, cmd: int, data: bytes = b"") -> bool: if not self._sock: return False try: self._sock.sendall(build_frame(cmd, data)) return True except Exception: return False def _recv(self) -> Optional[Tuple[int, bytes]]: if not self._sock: return None try: header = self._sock.recv(5) if len(header) < 5: return None length = (header[2] << 8) | header[3] rest = self._sock.recv(length + 1) if len(rest) < length + 1: return None return parse_frame(header + rest) except (socket.timeout, BlockingIOError): return None except Exception: return None def request_init(self) -> Optional[dict]: if not self._send(CMD_INIT): return None for _ in range(20): result = self._recv() if result and result[0] == CMD_INIT: data = result[1] if len(data) >= 29: self._init_info = { "width": (data[0] << 8) | data[1], "height": (data[2] << 8) | data[3], "mem_size": (data[5] << 24) | (data[6] << 16) | (data[7] << 8) | data[8], } return self._init_info time.sleep(0.05) return None def send_key(self, key: int) -> bool: return self._send(CMD_KEY, bytes([key])) def fetch_screen(self) -> Optional[bytes]: info = self._init_info if not info: info = self.request_init() if not info: return None mem_size = info["mem_size"] buffer = bytearray(mem_size) pos = 0 while pos < mem_size: if not self._send(CMD_LCDMEM, struct.pack(">I", pos)): return None for _ in range(30): result = self._recv() if result and result[0] == CMD_LCDMEM: payload = result[1] if len(payload) >= 4: start = struct.unpack(">I", payload[:4])[0] chunk = payload[4:] buffer[start : start + len(chunk)] = chunk pos = start + len(chunk) break time.sleep(0.02) else: return None return bytes(buffer) class ViewerApp: def __init__(self, host: str): self.host = host self.client = RemoDispViewer(host) self.root = tk.Tk() self.root.title(f"远程显示 - {host}") self.root.geometry("400x320") self._canvas = None self._photo = None self._scale = 2 self._build_ui() def _build_ui(self): # 连接区 frm = ttk.Frame(self.root, padding=5) frm.pack(fill=tk.X) ttk.Label(frm, text=f"目标: {self.host}:7003").pack(side=tk.LEFT) self._status = ttk.Label(frm, text="未连接") self._status.pack(side=tk.RIGHT) # 画布 self._canvas = tk.Canvas(self.root, width=320, height=320, bg="black") self._canvas.pack(padx=5, pady=5) # 按键区 btn_frm = ttk.Frame(self.root, padding=5) btn_frm.pack(fill=tk.X) keys = [ ("↑", KEY_U), ("↓", KEY_D), ("←", KEY_L), ("→", KEY_R), ("确认", KEY_ENT), ("取消", KEY_ESC), ("F1", KEY_F1), ("F2", KEY_F2), ] for label, key in keys: ttk.Button(btn_frm, text=label, width=6, command=lambda k=key: self._on_key(k)).pack(side=tk.LEFT, padx=2) # 刷新 ttk.Button(btn_frm, text="刷新", command=self._refresh).pack(side=tk.LEFT, padx=2) self.root.protocol("WM_DELETE_WINDOW", self._on_close) def _connect(self) -> bool: if self.client.connect(): if self.client.request_init(): self._status.config(text="已连接") return True self.client.disconnect() self._status.config(text="连接失败") return False def _on_key(self, key: int): if not self.client._sock: if not self._connect(): messagebox.showerror("错误", "请先连接装置") return self.client.send_key(key) def _refresh(self): if not self.client._sock: if not self._connect(): messagebox.showerror("错误", "连接失败") return data = self.client.fetch_screen() if not data: messagebox.showerror("错误", "拉取显存失败") return info = self.client._init_info w, h = info["width"], info["height"] # 单色转图像 try: from PIL import Image, ImageTk img = Image.new("1", (w, h)) pix = img.load() for y in range(h): for x in range(w): bi = (y * w + x) // 8 bit = 7 - (x % 8) pix[x, y] = 255 if (bi < len(data) and (data[bi] >> bit) & 1) else 0 img = img.resize((w * self._scale, h * self._scale), Image.NEAREST) self._photo = ImageTk.PhotoImage(img) self._canvas.delete("all") self._canvas.create_image(0, 0, anchor=tk.NW, image=self._photo) except ImportError: # 无 PIL：用简单矩形模拟 self._canvas.delete("all") step = max(1, 320 // w) for y in range(min(h, 160)): for x in range(min(w, 160)): bi = (y * w + x) // 8 bit = 7 - (x % 8) if bi < len(data) and (data[bi] >> bit) & 1: self._canvas.create_rectangle( x * step, y * step, (x + 1) * step, (y + 1) * step, fill="white", outline="" ) def _on_close(self): self.client.disconnect() self.root.destroy() def run(self): self._connect() self._refresh() self.root.mainloop() def main(): if len(sys.argv) < 2: print("用法: python remo_disp_viewer.py <装置IP>") sys.exit(1) app = ViewerApp(sys.argv[1]) app.run() if __name__ == "__main__": main()