AGV 信号优化复盘报告

项目中实际使用的 xPico 200 系列 AGV 网卡管理工具，基于 tkinter 图形界面操作，支持扫描设备、导出配置、批量修改漫游参数、查看实时连接状态。相比命令行版本，GUI 操作更直观高效。下方代码已对密码字段脱敏（实际使用时请填入真实密码）。

启动方式

python xpico_gui.py

启动后图形界面支持以下操作：

• 扫描设备 — 自动扫描网段内所有在线 xPico 设备并列出
• 导出配置 — 批量导出所有设备的 Radio/Roaming 配置，支持 CSV
• 批量修改 — 批量修改漫游参数（RSSI Floor / Trigger Delta）
• 查看状态 — 实时查看设备连接状态（SSID / BSSID / RSSI / Channel）
• 单台操作 — 点击设备可查看/修改单台配置

核心代码（xpico_gui.py 精简版）

#!/usr/bin/env python3
"""xPico 200 系列 AGV 网卡批量管理工具 - GUI 版本
功能：图形界面操作，支持扫描/导出/修改/查看单台设备配置
依赖：pip install requests  (tkinter 为 Python 内置)
"""

import csv, ipaddress, re, subprocess, sys, threading, webbrowser
import xml.etree.ElementTree as ET
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from tkinter import *
import tkinter.ttk as ttk
from tkinter import filedialog, messagebox
from tkinter.scrolledtext import ScrolledText

try:
    import requests
    from requests.auth import HTTPDigestAuth
except ImportError:
    print("[!] 需要安装 requests 库: pip install requests")
    sys.exit(1)

# ============== 配置区 ==============
SUBNET          = "172.21.58.0/23"
DEFAULT_USERNAME = "admin"
DEFAULT_PASSWORD = "********"   # 脱敏：实际部署填入 xPico 密码
TIMEOUT         = 3
MAX_WORKERS     = 30
PING_TIMEOUT    = 1

# ============== xPico HTTP API ==============
class XPicoClient:
    """通过 HTTP Digest Auth 与 xPico 200 通信"""
    def __init__(self, ip, username=DEFAULT_USERNAME, password=DEFAULT_PASSWORD, timeout=TIMEOUT):
        self.ip = ip
        self.base = f"http://{ip}"
        self.auth = HTTPDigestAuth(username, password)
        self.timeout = timeout
        self._session = None

    @property
    def session(self):
        if self._session is None:
            self._session = requests.Session()
        return self._session

    def _post(self, path, data=None, files=None):
        url = f"{self.base}{path}"
        try:
            if files:
                resp = self.session.post(url, auth=self.auth, files=files, timeout=self.timeout)
            elif data:
                resp = self.session.post(url, auth=self.auth, data=data, timeout=self.timeout)
            else:
                resp = self.session.post(url, auth=self.auth, timeout=self.timeout)
            return resp
        except requests.exceptions.RequestException:
            return None

    def is_xpico(self):
        resp = self._post("/export/config", data={"optionalGroupList": "Device"})
        if resp and resp.status_code == 200:
            text = resp.text
            if "configrecord" in text or "xPico" in text:
                return True
        return False

    def export_config(self, group_list=None):
        data = {}
        if group_list:
            data["optionalGroupList"] = group_list
        resp = self._post("/export/config", data=data)
        return resp.text if resp and resp.status_code == 200 else None

    def export_status(self, group_list=None):
        data = {}
        if group_list:
            data["optionalGroupList"] = group_list
        resp = self._post("/export/status", data=data)
        return resp.text if resp and resp.status_code == 200 else None

    def import_config(self, xml_content):
        resp = self._post("/import/config", files={"configrecord": ("config.xml", xml_content)})
        return resp.status_code == 200 if resp else False

    def save_config(self):
        resp = self._post("/action/status", data="group=Device&action=Save")
        return resp.status_code == 200 if resp else False

    def reboot(self):
        resp = self._post("/action/status", data="group=Device&action=Reboot")
        return resp.status_code == 200 if resp else False

# ============== Radio/Roaming 配置提取 ==============
def extract_radio_config(parsed):
    radio = parsed.get("Radio", {})
    wlan  = _find_wlan_profile(parsed)
    return {
        "Radio Mode":           radio.get("Mode", "N/A"),
        "Band":                 radio.get("Band", "N/A"),
        "Roaming State":        radio.get("Roaming:State", "N/A"),
        "Scan Period":          radio.get("Roaming:Scan Period", "N/A"),
        "2.4GHz RSSI Floor":    radio.get("Roaming:Band 2.4GHz RSSI Floor", "N/A"),
        "2.4GHz Trigger Delta": radio.get("Roaming:Band 2.4GHz Trigger Delta", "N/A"),
        "5GHz RSSI Floor":      radio.get("Roaming:Band 5GHz RSSI Floor", "N/A"),
        "5GHz Trigger Delta":   radio.get("Roaming:Band 5GHz Trigger Delta", "N/A"),
        "WLAN TX Power Maximum": wlan.get("Advanced:TX Power Maximum", "N/A"),
    }

# ============== GUI 主窗口 ==============
class XPicoGUI:
    def __init__(self, root):
        self.root = root
        self.root.title("xPico 200 AGV 网卡管理工具")
        self.root.geometry("1100x750")
        self.devices = []        # 已发现的 xPico 设备列表
        self._build_ui()

    def _build_ui(self):
        # 顶部工具栏：扫描 / 导出 / 修改 / 状态 按钮
        toolbar = ttk.Frame(self.root)
        toolbar.pack(fill=X, padx=8, pady=4)
        ttk.Button(toolbar, text="🔍 扫描设备", command=self._scan).pack(side=LEFT, padx=4)
        ttk.Button(toolbar, text="📤 导出配置", command=self._export).pack(side=LEFT, padx=4)
        ttk.Button(toolbar, text="📝 批量修改", command=self._modify).pack(side=LEFT, padx=4)
        ttk.Button(toolbar, text="📊 连接状态", command=self._status).pack(side=LEFT, padx=4)
        ttk.Button(toolbar, text="📄 导出CSV", command=self._export_csv).pack(side=LEFT, padx=4)

        # 左侧设备列表
        left = ttk.Frame(self.root)
        left.pack(side=LEFT, fill=Y, padx=8)
        ttk.Label(left, text="设备列表", font=("", 11, "bold")).pack()
        self.dev_listbox = Listbox(left, width=28, height=30)
        self.dev_listbox.pack(fill=Y, expand=True)
        self.dev_listbox.bind("<<ListboxSelect>>", self._on_select)

        # 右侧详情面板
        right = ttk.Frame(self.root)
        right.pack(side=RIGHT, fill=BOTH, expand=True, padx=8)
        ttk.Label(right, text="设备详情", font=("", 11, "bold")).pack()
        self.detail = ScrolledText(right, wrap=WORD, font=("Consolas", 10))
        self.detail.pack(fill=BOTH, expand=True)

    def _scan(self):
        """扫描网段内所有 xPico 设备（多线程 ping + 验证）"""
        self.detail.delete("1.0", END)
        self.detail.insert(END, "正在扫描网段，请稍候...\n")
        threading.Thread(target=self._do_scan, daemon=True).start()

    def _do_scan(self):
        ips = [str(ip) for ip in ipaddress.ip_network(SUBNET, strict=False).hosts()]
        # 1) 并发 ping
        online = []
        with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
            futs = {pool.submit(ping_check, ip): ip for ip in ips}
            for f in as_completed(futs):
                if f.result():
                    online.append(futs[f])
        # 2) 并发验证是否为 xPico
        self.devices = []
        with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
            futs = {pool.submit(XPicoClient(ip).is_xpico, ): ip for ip in online}
            for f in as_completed(futs):
                if f.result():
                    self.devices.append(futs[f])
        # 3) 更新 UI
        self.root.after(0, self._refresh_list)

    def _on_select(self, event):
        """点击设备 → 加载该设备详细配置"""
        sel = self.dev_listbox.curselection()
        if not sel: return
        ip = self.devices[sel[0]]
        threading.Thread(target=lambda: self._load_device(ip), daemon=True).start()

    def _load_device(self, ip):
        client = XPicoClient(ip)
        xml = client.export_config("Radio")
        if xml:
            parsed = parse_config_xml(xml)
            radio = extract_radio_config(parsed)
            self.root.after(0, lambda: self._show_detail(ip, radio))

    # ... 导出/修改/状态方法类似，均通过 XPicoClient API 操作

💡 实际部署时，DEFAULT_PASSWORD 替换为 xPico 设备密码。GUI 版本相比命令行版本操作更直观，项目中实际使用 GUI 版本进行批量管理。