Add V1 protocol discovery and upgrade support

scripts/device_upgrade.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""设备 V2 基础信息预检与 EPFP/4001 固件升级工具。"""
+"""V1/V2 身份预检与 EPFP/4001 固件升级工具。"""
 
 from __future__ import annotations
 
@@ -21,6 +21,8 @@ from typing import Any, Dict, Iterable, List, Optional, Tuple
 INFO_PORT = 8942
 UPGRADE_PORT = 8943
 INFO_COMMAND = 30101
+V1_INFO_PORT = 4941
+V1_INFO_COMMAND = 0x5A0E2104
 UPGRADE_COMMAND = 4001
 MAX_JSON_SIZE = 1024 * 1024
 MAX_STATUS_SIZE = 1024 * 1024
@@ -116,6 +118,30 @@ def query_device_info(host: str, timeout: float, logger: EventLogger) -> Dict[st
     return response
 
 
+def query_v1_device_info(host: str, timeout: float, logger: EventLogger) -> Dict[str, Any]:
+    """通过 V1 工厂 TCP/2104 获取设备身份与固件字段。"""
+    logger.emit("v1_info_request", host=host, port=V1_INFO_PORT,
+                command="0x%08X" % V1_INFO_COMMAND)
+    with socket.create_connection((host, V1_INFO_PORT), timeout=timeout) as sock:
+        sock.settimeout(timeout)
+        sock.sendall(struct.pack("!II", V1_INFO_COMMAND, 0))
+        header = recv_exact(sock, 8)
+        command, size = struct.unpack("!II", header)
+        if command != V1_INFO_COMMAND or size > MAX_JSON_SIZE:
+            raise UpgradeError("V1 固件信息响应 CMD 或长度异常")
+        payload = recv_exact(sock, size)
+    try:
+        response = json.loads(payload.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise UpgradeError("V1 固件信息响应不是有效 JSON：%s" % exc) from exc
+    if not isinstance(response, dict) or response.get("code") != 0:
+        raise UpgradeError("V1 固件信息查询失败或响应格式异常")
+    if not response.get("sn") or not response.get("dev_type"):
+        raise UpgradeError("V1 固件信息缺少 SN 或 dev_type，拒绝升级")
+    logger.emit("device_info", protocol="V1", response=response)
+    return response
+
+
 def response_data(response: Dict[str, Any]) -> Dict[str, Any]:
     """取得响应 data 对象，格式错误时拒绝继续。"""
 
@@ -379,9 +405,11 @@ def build_parser() -> argparse.ArgumentParser:
     """创建命令行参数解析器。"""
 
     parser = argparse.ArgumentParser(
-        description="V2 设备 30101 身份预检与 EPFP/4001 ROM 升级；默认仅预检。"
+        description="设备身份预检与 EPFP/4001 ROM 升级；默认仅预检。"
     )
     parser.add_argument("--host", required=True, help="目标设备 IPv4/主机名")
+    parser.add_argument("--protocol", choices=("v1", "v2"), default="v2",
+                        help="身份查询协议；固件传输均使用 EPFP/4001")
     parser.add_argument("--expected-sn", required=True, help="必须精确匹配的完整设备 SN")
     parser.add_argument("--expected-model", help="可选，必须出现在设备型号字段中的型号")
     parser.add_argument("--rom", required=True, type=Path, help="ROM 文件绝对或相对路径")
@@ -422,10 +450,23 @@ def main(argv: Optional[List[str]] = None) -> int:
         logger.emit("mode", execute=bool(args.execute), log_file=str(logger.path))
         rom = validate_rom(args.rom)
         logger.emit("rom", **rom)
-        response = query_device_info(args.host, args.connect_timeout, logger)
-        data, models = validate_identity(response, args.expected_sn, args.expected_model)
+        if args.protocol == "v1":
+            response = query_v1_device_info(args.host, args.connect_timeout, logger)
+            if response.get("sn") != args.expected_sn:
+                raise UpgradeError("设备 SN 不匹配：期望 %s，实际 %r" % (args.expected_sn, response.get("sn")))
+            if args.expected_model and args.expected_model not in str(response.get("dev_type", "")):
+                raise UpgradeError("设备型号不匹配：期望 %s，实际 %r" % (args.expected_model, response.get("dev_type")))
+            data = {"sn": response.get("sn"), "inner": {"type": response.get("dev_type"),
+                    "version": response.get("firmware")}, "outer": {"version": response.get("dev_ver")},
+                    "hardware": {"version": response.get("hardware")}}
+            models = [str(response["dev_type"])]
+        else:
+            response = query_device_info(args.host, args.connect_timeout, logger)
+            data, models = validate_identity(response, args.expected_sn, args.expected_model)
         summary = {
             "host": args.host,
+            "identity_protocol": args.protocol,
+            "upgrade_protocol_compatibility": "V1 身份查询 + EPFP/4001；仍须依据目标固件资料确认兼容性",
             "sn": data.get("sn"),
             "mac": data.get("mac"),
             "models": models,