Release 1.2.2: remove temporary debug instrumentation

README.md

@@ -12,7 +12,7 @@ UGV01 履带差速小车串口控制工程，提供底盘协议封装、手机
 
 ## 当前版本
 
-- 当前发布版本: `V1.2.0`
+- 当前发布版本: `1.2.2`
 - 控制方式: 基于底盘协议 `T=1` 的差速控制
 - Web 界面: 手机横屏优先的双摇杆控制台
 - FPV 方式: UVC 双目相机左视图实时视频
@@ -190,3 +190,4 @@ ROS 风格控制:
 - HTTPS 当前使用本地 CA 签发的证书；若希望浏览器将其视为受信任来源，需要安装并信任 `ca.crt`。
 - 相机采集采用独立后台线程，控制、反馈和网络设置仍保持原有服务链路。
 - 底盘 MCU 自带 AP 已通过命令关闭，并设置为开机关闭。
+- 已清理调试期加入的黑匣子、异常抓拍与调试接口，保留正式运行逻辑。

ugv01/controller.py

@@ -1,5 +1,4 @@
 import json
-from collections import deque
 from pathlib import Path
 import threading
 import time
@@ -39,7 +38,6 @@ class UGV01Controller:
         self.write_timeout = write_timeout
         self._serial: serial.Serial | None = None
         self._lock = threading.Lock()
-        self._send_history: deque[dict[str, Any]] = deque(maxlen=200)
 
     def open(self) -> None:
         with self._lock:
@@ -79,12 +77,6 @@ class UGV01Controller:
             message = json.dumps(payload, ensure_ascii=True, separators=(",", ":")) + "\n"
             ser.write(message.encode("utf-8"))
             ser.flush()
-            self._send_history.append(
-                {
-                    "ts": round(time.time(), 3),
-                    "payload": payload,
-                }
-            )
 
     def request(
         self,
@@ -181,11 +173,6 @@ class UGV01Controller:
             raw = ser.read(waiting).decode("utf-8", errors="replace")
         return self._parse_lines(raw)
 
-    def send_history(self) -> list[dict[str, Any]]:
-        with self._lock:
-            return list(self._send_history)
-
-
 class UGV01MotionService:
     def __init__(
         self,
@@ -239,7 +226,6 @@ class UGV01MotionService:
         self._last_steering = 0.0
         self._last_throttle = 0.0
         self._feedback_flow_interval_ms = idle_feedback_interval_ms
-        self._command_log: deque[dict[str, Any]] = deque(maxlen=80)
 
     def start(self) -> None:
         self.controller.open()
@@ -275,17 +261,6 @@ class UGV01MotionService:
             self._last_steering = 0.0
             self._last_throttle = 0.0
             self._sync_feedback_flow_interval_locked()
-            self._record_command(
-                source="system",
-                owner=None,
-                gear=None,
-                steering=None,
-                throttle=None,
-                left=0.0,
-                right=0.0,
-                accepted=True,
-                note="force_stop",
-            )
 
     def stop_if_owner(self, owner: str) -> None:
         with self._lock:
@@ -297,17 +272,6 @@ class UGV01MotionService:
         gear_name = gear if gear in self.speed_profiles else "medium"
         with self._lock:
             self._last_gear = gear_name
-            self._record_command(
-                source="system",
-                owner=None,
-                gear=gear_name,
-                steering=None,
-                throttle=None,
-                left=self._last_payload[0],
-                right=self._last_payload[1],
-                accepted=True,
-                note="set_preferred_gear",
-            )
         return self.status()
 
     def apply_joystick(self, steering: float, throttle: float, gear: str, owner: str) -> dict[str, float | str]:
@@ -321,17 +285,6 @@ class UGV01MotionService:
                 and not self._stopped
                 and (now - self._last_command_at) < self.idle_timeout
             ):
-                self._record_command(
-                    source="web",
-                    owner=owner,
-                    gear=gear if gear in self.speed_profiles else "medium",
-                    steering=steering,
-                    throttle=throttle,
-                    left=self._last_payload[0],
-                    right=self._last_payload[1],
-                    accepted=False,
-                    note="rejected_not_active_owner",
-                )
                 return {
                     "gear": gear if gear in self.speed_profiles else "medium",
                     "steering": steering,
@@ -351,17 +304,6 @@ class UGV01MotionService:
             self._last_steering = float(payload["steering"])
             self._last_throttle = float(payload["throttle"])
             self._sync_feedback_flow_interval_locked()
-            self._record_command(
-                source="web",
-                owner=owner,
-                gear=payload["gear"],
-                steering=payload["steering"],
-                throttle=payload["throttle"],
-                left=left,
-                right=right,
-                accepted=True,
-                note="apply_joystick",
-            )
         payload["left"] = left
         payload["right"] = right
         payload["accepted"] = True
@@ -384,17 +326,6 @@ class UGV01MotionService:
             self._last_steering = 0.0
             self._last_throttle = 0.0
             self._sync_feedback_flow_interval_locked()
-            self._record_command(
-                source="wtb",
-                owner="wtb",
-                gear=gear_name,
-                steering=None,
-                throttle=None,
-                left=left,
-                right=right,
-                accepted=True,
-                note=f"apply_wtb({left_norm:.3f},{right_norm:.3f})",
-            )
         return {
             "gear": gear_name,
             "left": left,
@@ -422,10 +353,6 @@ class UGV01MotionService:
                 "speed_profiles": self.speed_profiles,
             }
 
-    def command_log(self) -> list[dict[str, Any]]:
-        with self._lock:
-            return list(self._command_log)
-
     def raw_state(self) -> dict[str, Any]:
         with self._lock:
             return {
@@ -453,17 +380,6 @@ class UGV01MotionService:
                     self._last_steering = 0.0
                     self._last_throttle = 0.0
                     self._sync_feedback_flow_interval_locked()
-                    self._record_command(
-                        source="system",
-                        owner=None,
-                        gear=None,
-                        steering=None,
-                        throttle=None,
-                        left=0.0,
-                        right=0.0,
-                        accepted=True,
-                        note="idle_timeout_stop",
-                    )
                     continue
                 if self._stopped:
                     continue
@@ -558,35 +474,6 @@ class UGV01MotionService:
             right = target_right
         return left, right
 
-    def _record_command(
-        self,
-        *,
-        source: str,
-        owner: str | None,
-        gear: str | None,
-        steering: float | None,
-        throttle: float | None,
-        left: float,
-        right: float,
-        accepted: bool,
-        note: str,
-    ) -> None:
-        self._command_log.append(
-            {
-                "ts": round(time.time(), 3),
-                "source": source,
-                "owner": owner,
-                "gear": gear,
-                "steering": steering,
-                "throttle": throttle,
-                "left": left,
-                "right": right,
-                "accepted": accepted,
-                "note": note,
-            }
-        )
-
-
 def latest_dict(items: list[Any], target_t: int) -> dict[str, Any] | None:
     for item in reversed(items):
         if isinstance(item, dict) and item.get("T") == target_t:
@@ -666,11 +553,6 @@ class UGV01FeedbackService:
         self._latest_base: dict[str, Any] | None = None
         self._base_items: list[Any] = []
         self._updated_at = 0.0
-        self._summary_log: deque[dict[str, Any]] = deque(maxlen=120)
-        self._raw_items_log: deque[dict[str, Any]] = deque(maxlen=240)
-        self._trace_log: deque[dict[str, Any]] = deque(maxlen=800)
-        self._anomaly_captures: deque[dict[str, Any]] = deque(maxlen=24)
-        self._last_capture_at = 0.0
 
     def start(self) -> None:
         self.controller.open()
@@ -704,18 +586,6 @@ class UGV01FeedbackService:
                 "updated_at": self._updated_at,
             }
 
-    def summary_log(self) -> list[dict[str, Any]]:
-        with self._lock:
-            return list(self._summary_log)
-
-    def debug_snapshot(self) -> dict[str, Any]:
-        with self._lock:
-            return {
-                "recent_trace": list(self._trace_log),
-                "recent_raw_items": list(self._raw_items_log),
-                "captures": list(self._anomaly_captures),
-            }
-
     def _read_loop(self) -> None:
         while not self._stop_event.wait(0.03):
             items = self.controller.read_available()
@@ -728,72 +598,3 @@ class UGV01FeedbackService:
                 self._latest_base = latest
                 self._base_items = [latest]
                 self._updated_at = time.time()
-                self._raw_items_log.append(
-                    {
-                        "ts": round(self._updated_at, 3),
-                        "items": items,
-                    }
-                )
-                summary = summarize_base_feedback(latest)
-                self._summary_log.append({"ts": round(self._updated_at, 3), "summary": summary})
-                self._append_trace(summary)
-
-    def _append_trace(self, summary: dict[str, Any]) -> None:
-        state = self.state_supplier() if self.state_supplier else {}
-        sample = {
-            "ts": round(self._updated_at, 3),
-            "command": {
-                "left": state.get("left"),
-                "right": state.get("right"),
-                "gear": state.get("gear"),
-                "steering": state.get("steering"),
-                "throttle": state.get("throttle"),
-                "source": state.get("control_source"),
-                "owner": state.get("control_owner"),
-                "stopped": state.get("stopped"),
-            },
-            "feedback": summary,
-        }
-        self._trace_log.append(sample)
-        self._maybe_capture_anomaly(sample)
-
-    def _maybe_capture_anomaly(self, sample: dict[str, Any]) -> None:
-        command = sample["command"]
-        feedback = sample["feedback"]
-        if command.get("stopped"):
-            return
-        left_cmd = command.get("left")
-        right_cmd = command.get("right")
-        left_fb = feedback.get("left_speed")
-        right_fb = feedback.get("right_speed")
-        if not isinstance(left_cmd, (int, float)) or not isinstance(right_cmd, (int, float)):
-            return
-        if not isinstance(left_fb, (int, float)) or not isinstance(right_fb, (int, float)):
-            return
-        command_mag = max(abs(left_cmd), abs(right_cmd))
-        straight_cmd = abs(left_cmd - right_cmd) <= 0.003 and command_mag >= 0.01
-        if not straight_cmd:
-            return
-        feedback_mag = max(abs(left_fb), abs(right_fb))
-        excessive = feedback_mag > command_mag + 0.05
-        asymmetric = abs(left_fb - right_fb) > 0.05
-        if not (excessive or asymmetric):
-            return
-        now = self._updated_at
-        if (now - self._last_capture_at) < 0.8:
-            return
-        self._last_capture_at = now
-        reason_parts: list[str] = []
-        if excessive:
-            reason_parts.append("feedback_over_command")
-        if asymmetric:
-            reason_parts.append("feedback_asymmetric")
-        self._anomaly_captures.append(
-            {
-                "ts": round(now, 3),
-                "reason": ",".join(reason_parts),
-                "sample": sample,
-                "context": list(self._trace_log)[-40:],
-                "raw_context": list(self._raw_items_log)[-40:],
-            }
-        )

ugv01/webapp.py

@@ -114,26 +114,6 @@ async def status():
     return payload
 
 
-@app.get("/api/debug/commands")
-async def debug_commands():
-    return {
-        "ok": True,
-        "current": motion_service.status(),
-        "recent": motion_service.command_log(),
-        "serial_writes": controller.send_history(),
-        "feedback_recent": feedback_service.summary_log(),
-    }
-
-
-@app.get("/api/debug/control-trace")
-async def debug_control_trace():
-    return {
-        "ok": True,
-        "current": motion_service.status(),
-        **feedback_service.debug_snapshot(),
-    }
-
-
 @app.post("/api/stop")
 async def stop(payload: StopPayload | None = None):
     if wtb_service.is_authorized():