From e501d8095340adebe8f177ded99bdc4a6f2d6f93 Mon Sep 17 00:00:00 2001
From: zhanghao <774378400@qq.com>
Date: Mon, 3 Aug 2026 14:43:09 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E6=9C=BA=E6=A2=B0=E8=87=82=E6=8F=90?=
 =?UTF-8?q?=E7=A4=BA?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/plugins/websocket.js                      | 147 +++--
 .../register/components/Camera/index.vue      | 582 ++++++++++++------
 .../components/params/BasicNodeParams.vue     |  16 +-
 3 files changed, 500 insertions(+), 245 deletions(-)

diff --git a/src/plugins/websocket.js b/src/plugins/websocket.js
index 74de402..f3ca635 100644
--- a/src/plugins/websocket.js
+++ b/src/plugins/websocket.js
@@ -1,4 +1,8 @@
-import { throttle } from 'lodash';
+const MEDIA_HEADER_SIZE = 26;
+const MEDIA_MAGIC = [0x43, 0x4d, 0x56, 0x46]; // CMVF
+const MEDIA_MIME_TYPES = {
+  1: 'video/mp4; codecs="avc1.42C01E"'
+};
 
 class WebSocketManager {
   constructor(url, userId) {
@@ -10,7 +14,8 @@ class WebSocketManager {
     this.heartbeatTimer = null;
     this.isManualClose = false;
     this.messageQueue = [];
-    this.maxQueueSize = 1;
+    this.processingMessages = false;
+    this.maxQueueSize = 100;
     this.status = false;
   }
 
@@ -28,21 +33,28 @@ class WebSocketManager {
       this.status = true;
       this.startHeartbeat();
       this.processMessageQueue();
+      this.dispatch('open');
     };
 
-    this.ws.onmessage = (event) => {
-      if (this.messageQueue.length < this.maxQueueSize) {
-        this.messageQueue.push(event);
-        this.processMessageQueue();
-      } else {
-        console.warn('消息队列已满，丢弃消息');
+    this.ws.onmessage = async (event) => {
+      const data = event.data instanceof Blob ? await event.data.arrayBuffer() : event.data;
+      if (data instanceof ArrayBuffer && this.isMediaPacket(data)) {
+        this.dispatchMediaPacket(data);
+        return;
       }
+      if (this.messageQueue.length >= this.maxQueueSize) {
+        console.warn('WebSocket控制消息队列已满，丢弃最旧消息');
+        this.messageQueue.shift();
+      }
+      this.messageQueue.push(data);
+      this.processMessageQueue();
     };
 
     this.ws.onclose = () => {
       console.log('WebSocket 断开');
       this.status = false;
       this.stopHeartbeat();
+      this.dispatch('close');
       if (!this.isManualClose) {
         this.reconnect();
       }
@@ -53,65 +65,80 @@ class WebSocketManager {
     };
   }
 
-  async processMessageQueue() {
-    if (!this.ws || this.messageQueue.length === 0) return;
+  isMediaPacket(buffer) {
+    if (buffer.byteLength < MEDIA_HEADER_SIZE) return false;
+    const bytes = new Uint8Array(buffer, 0, MEDIA_MAGIC.length);
+    return MEDIA_MAGIC.every((value, index) => bytes[index] === value);
+  }
 
-    const event = this.messageQueue.shift();
+  dispatchMediaPacket(buffer) {
     try {
-      // 处理消息数据
-      console.log('收到消息数据类型:', event.data.constructor.name);
-      let compressedData;
-
-      // 处理 Blob 或 ArrayBuffer
-      if (event.data instanceof Blob) {
-        console.log('处理 Blob 数据，大小:', event.data.size);
-        compressedData = new Uint8Array(await event.data.arrayBuffer());
-      } else if (event.data instanceof ArrayBuffer) {
-        console.log('处理 ArrayBuffer 数据，大小:', event.data.byteLength);
-        compressedData = new Uint8Array(event.data);
-      } else {
-        throw new Error('不支持的数据类型: ' + event.data.constructor.name);
+      const view = new DataView(buffer);
+      const version = view.getUint8(4);
+      const mediaType = view.getUint8(5);
+      const timestamp = view.getUint32(8) * 0x100000000 + view.getUint32(12);
+      const sequence = view.getUint32(16);
+      const channelLength = view.getUint16(20);
+      const payloadLength = view.getUint32(22);
+      const payloadOffset = MEDIA_HEADER_SIZE + channelLength;
+      if (version !== 1 || payloadOffset + payloadLength !== buffer.byteLength) {
+        throw new Error('媒体数据包头无效');
       }
-
-      // 使用 DecompressionStream 解压 GZIP 数据
-      const decompressionStream = new DecompressionStream('gzip');
-      const writer = decompressionStream.writable.getWriter();
-      writer.write(compressedData);
-      writer.close();
-
-      const decompressedStream = decompressionStream.readable;
-      const reader = decompressedStream.getReader();
-      let decompressedData = new Uint8Array(0);
-
-      while (true) {
-        const { done, value } = await reader.read();
-        if (done) break;
-        const newData = new Uint8Array(decompressedData.length + value.length);
-        newData.set(decompressedData);
-        newData.set(value, decompressedData.length);
-        decompressedData = newData;
+      const channel = new TextDecoder().decode(new Uint8Array(buffer, MEDIA_HEADER_SIZE, channelLength));
+      const mimeType = MEDIA_MIME_TYPES[mediaType];
+      if (!mimeType) {
+        throw new Error(`不支持的媒体类型: ${mediaType}`);
       }
+      this.dispatch(channel, {
+        media: true,
+        mimeType,
+        sequence,
+        timestamp,
+        data: buffer.slice(payloadOffset, payloadOffset + payloadLength)
+      });
+    } catch (error) {
+      console.error('WebSocket媒体包解析失败:', error);
+    }
+  }
 
-      // 将解压后的数据转换为字符串
-      const textDecoder = new TextDecoder();
-      const decompressedString = textDecoder.decode(decompressedData);
-      console.log('解压后的数据:', JSON.parse(decompressedString));
-
-      // 解析 JSON
-      const { type, channel, payload,timestamp } = JSON.parse(decompressedString);
-      console.log('后端到前端延迟:', ((new Date()).getTime() - timestamp));
-
-      if (channel && this.events[channel]) {
-        this.events[channel].forEach(callback => callback(payload));
-      } else if (type && this.events[type]) {
-        this.events[type].forEach(callback => callback(payload));
+  async processMessageQueue() {
+    if (this.processingMessages || !this.ws) return;
+    this.processingMessages = true;
+    try {
+      while (this.messageQueue.length > 0) {
+        const data = this.messageQueue.shift();
+        let message;
+        if (typeof data === 'string') {
+          message = JSON.parse(data);
+        } else if (data instanceof ArrayBuffer) {
+          const decompressedStream = new Blob([data])
+            .stream()
+            .pipeThrough(new DecompressionStream('gzip'));
+          const decompressedData = await new Response(decompressedStream).arrayBuffer();
+          message = JSON.parse(new TextDecoder().decode(decompressedData));
+        } else {
+          throw new Error(`不支持的WebSocket数据类型: ${typeof data}`);
+        }
+        if (message.channel) {
+          this.dispatch(message.channel, message.payload);
+        } else if (message.type) {
+          this.dispatch(message.type, message.payload ?? message.content ?? message);
+        }
       }
-      setTimeout(() => this.processMessageQueue(), 0);
     } catch (err) {
-      console.error('WebSocket 消息解析失败:', err);
-      console.error('错误堆栈:', err.stack);
-      console.error('原始数据类型:', event.data.constructor.name);
-      console.error('原始数据大小:', event.data instanceof Blob ? event.data.size : event.data.byteLength || '未知');
+      console.error('WebSocket消息解析失败:', err);
+    } finally {
+      this.processingMessages = false;
+      if (this.messageQueue.length > 0) {
+        queueMicrotask(() => this.processMessageQueue());
+      }
+    }
+  }
+
+  dispatch(eventName, payload) {
+    const callbacks = this.events[eventName];
+    if (callbacks) {
+      callbacks.slice().forEach(callback => callback(payload));
     }
   }
 
diff --git a/src/views/device/register/components/Camera/index.vue b/src/views/device/register/components/Camera/index.vue
index 466ccb9..a7becbe 100644
--- a/src/views/device/register/components/Camera/index.vue
+++ b/src/views/device/register/components/Camera/index.vue
@@ -14,30 +14,20 @@
         </div>
       </el-col>
       <el-col :span="20" class="video-panel" v-if="form.stereoModule || form.rgbCamera">
-        <div v-if="form.stereoModule && form.rgbCamera" class="stream-container">
-          <div class="stream-item">
+        <div class="stream-container" :class="{ dual: form.stereoModule && form.rgbCamera }">
+          <div v-show="form.stereoModule" class="stream-item">
             <div class="stream-wrapper">
-              <video ref="depthVideo"  autoplay muted playsinline></video>
-              <video ref="depthVideoBuffer"  autoplay muted playsinline style="display: none;"></video>
+              <video ref="depthVideo" autoplay muted playsinline></video>
+              <span class="stream-state">{{ statusText(streamStatus.getDepthImageStream) }}</span>
             </div>
           </div>
-          <div class="stream-item">
+          <div v-show="form.rgbCamera" class="stream-item">
             <div class="stream-wrapper">
-              <video ref="colorVideo"  autoplay muted playsinline></video>
-              <video ref="colorVideoBuffer"  autoplay muted playsinline style="display: none;"></video>
+              <video ref="colorVideo" autoplay muted playsinline></video>
+              <span class="stream-state">{{ statusText(streamStatus.getRGBImageStream) }}</span>
             </div>
           </div>
         </div>
-        <div v-else class="single-stream-container">
-          <div v-if="form.stereoModule" class="stream-wrapper">
-            <video ref="depthVideo" width="640" height="480" autoplay muted playsinline></video>
-            <video ref="depthVideoBuffer" width="640" height="480" autoplay muted playsinline style="display: none;"></video>
-          </div>
-          <div v-if="form.rgbCamera" class="stream-wrapper">
-            <video ref="colorVideo" width="640" height="480" autoplay muted playsinline></video>
-            <video ref="colorVideoBuffer" width="640" height="480" autoplay muted playsinline style="display: none;"></video>
-          </div>
-        </div>
       </el-col>
       <el-col v-else :span="20" style="align-self: center;">
         <el-empty description="请选择模式" />
@@ -47,18 +37,17 @@
 </template>
 
 <script setup>
-import { inject, reactive, toRefs, watch, onMounted, onUnmounted, ref, nextTick } from 'vue';
+import { inject, reactive, toRefs, watch, onUnmounted, ref, nextTick } from 'vue';
 import { getRegister } from "@/api/device/register";
+import { getRgbStreamUrl } from "@/api/device/camera";
 import { useRoute } from "vue-router";
-import { debounce } from 'lodash'; // 引入 lodash 的防抖函数
 
-// 1. 定义 props
 const props = defineProps({
-  initialDeviceId: { // 示例 prop，用于从弹窗接收cameraId
+  initialDeviceId: {
     type: String,
     default: null,
   },
-  initialTerminalId: { // 示例 prop，用于从弹窗接收 terminalId
+  initialTerminalId: {
     type: String,
     default: null,
   },
@@ -78,175 +67,402 @@ const data = reactive({
 });
 const { form } = toRefs(data);
 const depthVideo = ref(null);
-const depthVideoBuffer = ref(null);
 const colorVideo = ref(null);
-const colorVideoBuffer = ref(null);
-const blobUrls = ref([]);
-const frameData = reactive({
-  getRGBImageStream: { count: 0, lastTime: 0 },
-  getDepthImageStream: { count: 0, lastTime: 0 },
+const streamStatus = reactive({
+  getRGBImageStream: 'idle',
+  getDepthImageStream: 'idle',
 });
+const mediaPlayers = new Map();
+const retryTimers = new Map();
+let rgbStreamGeneration = 0;
+let rgbStreamRetryTimer = null;
+let rgbStreamAbortController = null;
+const MAX_QUEUE_BYTES = 2 * 1024 * 1024;
+const STARTUP_BUFFER_SECONDS = 0.35;
+const TARGET_LIVE_LATENCY_SECONDS = 0.45;
+const MAX_PLAYED_BUFFER_SECONDS = 2;
+const PLAYBACK_SYNC_INTERVAL_MS = 200;
 
-function renderVideo(mainRef, bufferRef, data, channel, imageType) {
-  const mainVideo = mainRef.value;
-  const bufferVideo = bufferRef.value;
-  if (!mainVideo || !bufferVideo) {
-    console.error(`视频元素未找到 (${imageType}): 通道: ${channel}`);
-    return;
-  }
+const statusText = (status) => ({
+  connecting: '连接中',
+  live: '实时',
+  retrying: '重连中',
+  error: '播放异常',
+}[status] || '未连接');
 
-  const start = performance.now();
+function getVideoRef(method) {
+  return method === 'getRGBImageStream' ? colorVideo : depthVideo;
+}
+
+async function startDirectRgbStream() {
+  if (!data.terminalId || !data.cameraId || !colorVideo.value) return;
+  rgbStreamAbortController?.abort();
+  rgbStreamAbortController = null;
+  const generation = ++rgbStreamGeneration;
+  clearTimeout(rgbStreamRetryTimer);
+  rgbStreamRetryTimer = null;
+  streamStatus.getRGBImageStream = 'connecting';
   try {
-    if (typeof data === 'string') {
-      const newSrc = data.startsWith('blob:') ? data : `data:video/mp4;base64,${data}`;
-      
-      // 先加载到缓冲视频元素
-      bufferVideo.src = newSrc;
-      bufferVideo.onloadeddata = () => {
-        // 缓冲视频加载完成后，切换到主视频
-        if (mainVideo.src && mainVideo.src.startsWith('blob:')) {
-          URL.revokeObjectURL(mainVideo.src);
-          blobUrls.value = blobUrls.value.filter(url => url !== mainVideo.src);
-        }
-        mainVideo.src = bufferVideo.src;
-        blobUrls.value.push(newSrc);
-        bufferVideo.src = ''; // 清空缓冲视频源
-      };
-
-      frameData[imageType].count++;
-      const currentTime = performance.now();
-      if (frameData[imageType].lastTime && currentTime - frameData[imageType].lastTime >= 1000) {
-        const fps = frameData[imageType].count * 1000 / (currentTime - frameData[imageType].lastTime);
-        console.log(`渲染帧率 (${imageType}): ${fps.toFixed(2)} fps, 通道: ${channel}`);
-        frameData[imageType].count = 0;
-        frameData[imageType].lastTime = currentTime;
-      } else if (!frameData[imageType].lastTime) {
-        frameData[imageType].lastTime = currentTime;
-      }
-      console.log(`视频渲染耗时 (${imageType}): ${performance.now() - start}ms, 通道: ${channel}`);
-      if (performance.memory) {
-        console.log('当前内存使用:', {
-          usedJSHeapSize: (performance.memory.usedJSHeapSize / 1024 / 1024).toFixed(2) + ' MB',
-          totalJSHeapSize: (performance.memory.totalJSHeapSize / 1024 / 1024).toFixed(2) + ' MB',
-        });
-      }
-    } else {
-      console.error(`未知 payload 类型 (${imageType}):`, typeof data, '通道:', channel);
+    const response = await getRgbStreamUrl(data.terminalId, data.cameraId);
+    if (generation !== rgbStreamGeneration || !data.form.rgbCamera) return;
+    const relativeUrl = response.data;
+    const baseUrl = import.meta.env.VITE_APP_BASE_API.replace(/\/$/, '');
+    const video = colorVideo.value;
+    if (!video) return;
+    const player = createMediaPlayer(
+      'getRGBImageStream',
+      video,
+      'video/mp4; codecs="avc1.42C01E"'
+    );
+    rgbStreamAbortController = new AbortController();
+    consumeDirectRgbStream(
+      `${baseUrl}${relativeUrl}`,
+      generation,
+      player,
+      rgbStreamAbortController.signal
+    );
+  } catch (error) {
+    if (generation === rgbStreamGeneration && data.form.rgbCamera) {
+      console.error('Failed to open the direct RGB video stream:', error);
+      scheduleDirectRgbRestart();
     }
-  } catch (err) {
-    console.error(`视频渲染失败 (${imageType}):`, err, '通道:', channel);
   }
 }
 
-// 防抖订阅函数
-const subscribeDebounced = debounce(async (sub, terminalId, method) => {
-  if (!socket || !terminalId || !data.cameraId) {
-    console.warn(`订阅失败: socket=${!!socket}, terminalId=${terminalId}, cameraId=${data.cameraId}, method=${method}`);
+async function consumeDirectRgbStream(url, generation, player, signal) {
+  try {
+    const response = await fetch(url, { signal, cache: 'no-store' });
+    if (!response.ok || !response.body) {
+      throw new Error(`RGB video stream request failed: HTTP ${response.status}`);
+    }
+    const reader = response.body.getReader();
+    while (generation === rgbStreamGeneration && data.form.rgbCamera) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      if (!value?.byteLength || player.destroyed) continue;
+      const chunk = value.buffer.slice(value.byteOffset, value.byteOffset + value.byteLength);
+      player.queue.push(chunk);
+      player.queueBytes += chunk.byteLength;
+      if (player.queueBytes > MAX_QUEUE_BYTES) {
+        throw new Error(`RGB video stream queue exceeded ${MAX_QUEUE_BYTES} bytes`);
+      }
+      pumpMediaQueue(player);
+    }
+    if (!signal.aborted && generation === rgbStreamGeneration && data.form.rgbCamera) {
+      scheduleDirectRgbRestart();
+    }
+  } catch (error) {
+    if (error?.name !== 'AbortError' && generation === rgbStreamGeneration && data.form.rgbCamera) {
+      console.error('Direct RGB video stream interrupted:', error);
+      scheduleDirectRgbRestart();
+    }
+  }
+}
+
+function stopDirectRgbStream() {
+  rgbStreamGeneration++;
+  clearTimeout(rgbStreamRetryTimer);
+  rgbStreamRetryTimer = null;
+  rgbStreamAbortController?.abort();
+  rgbStreamAbortController = null;
+  destroyMediaPlayer('getRGBImageStream');
+  streamStatus.getRGBImageStream = 'idle';
+}
+
+function scheduleDirectRgbRestart() {
+  if (rgbStreamRetryTimer || !data.form.rgbCamera) return;
+  rgbStreamAbortController?.abort();
+  rgbStreamAbortController = null;
+  destroyMediaPlayer('getRGBImageStream');
+  streamStatus.getRGBImageStream = 'retrying';
+  rgbStreamRetryTimer = setTimeout(() => {
+    rgbStreamRetryTimer = null;
+    startDirectRgbStream();
+  }, 1000);
+}
+
+function createMediaPlayer(method, video, mimeType) {
+  if (!window.MediaSource || !MediaSource.isTypeSupported(mimeType)) {
+    streamStatus[method] = 'error';
+    throw new Error(`浏览器不支持媒体格式: ${mimeType}`);
+  }
+
+  destroyMediaPlayer(method);
+  const mediaSource = new MediaSource();
+  const objectUrl = URL.createObjectURL(mediaSource);
+  const state = {
+    method,
+    video,
+    mimeType,
+    mediaSource,
+    sourceBuffer: null,
+    objectUrl,
+    queue: [],
+    queueBytes: 0,
+    lastSequence: 0,
+    started: false,
+    syncTimer: null,
+    destroyed: false,
+  };
+  mediaPlayers.set(method, state);
+  video.src = objectUrl;
+
+  mediaSource.addEventListener('sourceopen', () => {
+    if (state.destroyed || state.sourceBuffer) return;
+    try {
+      const sourceBuffer = mediaSource.addSourceBuffer(mimeType);
+      state.sourceBuffer = sourceBuffer;
+      sourceBuffer.addEventListener('updateend', () => pumpMediaQueue(state));
+      sourceBuffer.addEventListener('error', () => restartMediaStream(method));
+      state.syncTimer = window.setInterval(() => maintainLivePlayback(state), PLAYBACK_SYNC_INTERVAL_MS);
+      pumpMediaQueue(state);
+    } catch (error) {
+      console.error('创建视频缓冲区失败:', error);
+      restartMediaStream(method);
+    }
+  }, { once: true });
+  return state;
+}
+
+function appendMediaPacket(method, packet) {
+  const video = getVideoRef(method).value;
+  if (!video) return;
+  let state = mediaPlayers.get(method);
+  if (!state || state.mimeType !== packet.mimeType || state.video !== video) {
+    state = createMediaPlayer(method, video, packet.mimeType);
+  }
+  if (state.lastSequence && packet.sequence !== state.lastSequence + 1) {
+    console.warn(`视频片段序号中断: ${state.lastSequence} -> ${packet.sequence}`);
+    scheduleRestart(method);
     return;
   }
-  const channel = `edgeCameraServiceImpl/${method}/${terminalId}/${data.cameraId}`;
-  console.log(sub ? '订阅' : '取消订阅', channel);
+  state.lastSequence = packet.sequence;
+  state.queue.push(packet.data);
+  state.queueBytes += packet.data.byteLength;
+  if (state.queueBytes > MAX_QUEUE_BYTES) {
+    console.warn(`视频缓冲积压超过${MAX_QUEUE_BYTES / 1024 / 1024}MB，重新建立实时流`);
+    scheduleRestart(method);
+    return;
+  }
+  pumpMediaQueue(state);
+}
+
+function pumpMediaQueue(state) {
+  const sourceBuffer = state.sourceBuffer;
+  if (state.destroyed || !sourceBuffer || sourceBuffer.updating) return;
+  try {
+    maintainLivePlayback(state);
+    const buffered = sourceBuffer.buffered;
+    if (buffered.length > 0 && state.started) {
+      const oldestBufferedTime = buffered.start(0);
+      const removeEnd = state.video.currentTime - 1;
+      if (removeEnd - oldestBufferedTime > MAX_PLAYED_BUFFER_SECONDS) {
+        sourceBuffer.remove(oldestBufferedTime, removeEnd);
+        return;
+      }
+    }
+    if (state.queue.length === 0) return;
+    const chunk = state.queue.shift();
+    state.queueBytes -= chunk.byteLength;
+    sourceBuffer.appendBuffer(chunk);
+    streamStatus[state.method] = 'live';
+  } catch (error) {
+    console.error('追加视频片段失败:', error);
+    restartMediaStream(state.method);
+  }
+}
+
+function restartMediaStream(method) {
+  if (method === 'getRGBImageStream') {
+    scheduleDirectRgbRestart();
+  } else {
+    scheduleRestart(method);
+  }
+}
+
+function maintainLivePlayback(state) {
+  const { sourceBuffer, video } = state;
+  if (state.destroyed || !sourceBuffer || sourceBuffer.updating || sourceBuffer.buffered.length === 0) return;
+
+  const buffered = sourceBuffer.buffered;
+  const rangeIndex = buffered.length - 1;
+  const liveStart = buffered.start(rangeIndex);
+  const liveEnd = buffered.end(rangeIndex);
+  const bufferedDuration = liveEnd - liveStart;
+
+  if (!state.started) {
+    if (bufferedDuration < STARTUP_BUFFER_SECONDS) return;
+    video.currentTime = Math.max(liveStart, liveEnd - TARGET_LIVE_LATENCY_SECONDS);
+    video.playbackRate = 1;
+    state.started = true;
+    video.play().catch(() => {});
+    return;
+  }
+
+  if (video.paused) {
+    video.play().catch(() => {});
+  }
+  if (video.seeking) return;
+
+  const latency = liveEnd - video.currentTime;
+  let playbackRate = 1;
+  if (latency > 1.4) {
+    playbackRate = 1.2;
+  } else if (latency > 1) {
+    playbackRate = 1.12;
+  } else if (latency > 0.7) {
+    playbackRate = 1.05;
+  } else if (latency < 0.28) {
+    playbackRate = 0.98;
+  }
+  if (Math.abs(video.playbackRate - playbackRate) > 0.001) {
+    video.playbackRate = playbackRate;
+  }
+}
+
+function destroyMediaPlayer(method) {
+  const state = mediaPlayers.get(method);
+  if (!state) return;
+  state.destroyed = true;
+  state.queue = [];
+  state.queueBytes = 0;
+  if (state.syncTimer) {
+    clearInterval(state.syncTimer);
+    state.syncTimer = null;
+  }
+  const video = state.video;
+  if (video) {
+    video.playbackRate = 1;
+    video.pause();
+    video.removeAttribute('src');
+    video.load();
+  }
+  URL.revokeObjectURL(state.objectUrl);
+  mediaPlayers.delete(method);
+}
+
+function channelFor(method) {
+  return `edgeCameraServiceImpl/${method}/${data.terminalId}/${data.cameraId}`;
+}
+
+async function setSubscription(enabled, method, force = false) {
+  if (!socket || !data.terminalId || !data.cameraId) return;
+  await nextTick();
+  const channel = channelFor(method);
+
+  if (enabled) {
+    if (data.callbacks[channel] && !force) return;
+    streamStatus[method] = 'connecting';
+    if (!data.callbacks[channel]) {
+      const callback = (payload) => {
+        if (payload?.media) {
+          appendMediaPacket(method, payload);
+        } else if (payload === '500') {
+          scheduleRestart(method);
+        }
+      };
+      // 先注册前端回调，再通知后端启动，避免丢失初始化片段。
+      socket.on(channel, callback);
+      data.callbacks[channel] = callback;
+    }
+  } else {
+    clearTimeout(retryTimers.get(method));
+    retryTimers.delete(method);
+  }
+
   socket.send({
     type: 'channel_subscription',
-    action: sub ? 'subscribe' : 'unsubscribe',
+    action: enabled ? 'subscribe' : 'unsubscribe',
     channel,
   });
-
-  if (sub) {
-    await nextTick();
-    const mainRef = method === 'getRGBImageStream' ? colorVideo : depthVideo;
-    const bufferRef = method === 'getRGBImageStream' ? colorVideoBuffer : depthVideoBuffer;
-    if (!mainRef.value || !bufferRef.value) {
-      console.error(`视频元素未找到: ${method}, 通道: ${channel}`);
-      return;
-    }
-
-    // 清理旧回调
-    if (data.callbacks[channel]) {
-      socket.off(channel, data.callbacks[channel]);
-      console.log('清理旧回调:', channel);
-    }
-
-    const callback = (data) => {
-      if (data !== '500') {
-        requestAnimationFrame(() => renderVideo(mainRef, bufferRef, data, channel, method));
-      } else {
-        // 错误处理：取消订阅后重新订阅
-        subscribeDebounced(false, terminalId, method);
-        subscribeDebounced(true, terminalId, method);
-      }
-    };
-    socket.on(channel, callback);
-    data.callbacks[channel] = callback;
-    console.log('注册新回调:', channel);
-  } else {
+  if (!enabled) {
     if (data.callbacks[channel]) {
       socket.off(channel, data.callbacks[channel]);
       delete data.callbacks[channel];
-      console.log('清理回调:', channel);
     }
+    destroyMediaPlayer(method);
+    streamStatus[method] = 'idle';
   }
-}, 300); // 300ms 防抖延迟
+}
+
+function scheduleRestart(method) {
+  if (retryTimers.has(method) || !data.form[method === 'getRGBImageStream' ? 'rgbCamera' : 'stereoModule']) return;
+  streamStatus[method] = 'retrying';
+  destroyMediaPlayer(method);
+  const timer = setTimeout(async () => {
+    retryTimers.delete(method);
+    await setSubscription(false, method);
+    await new Promise(resolve => setTimeout(resolve, 500));
+    if (data.form[method === 'getRGBImageStream' ? 'rgbCamera' : 'stereoModule']) {
+      await setSubscription(true, method);
+    }
+  }, 1000);
+  retryTimers.set(method, timer);
+}
+
+function handleSocketOpen() {
+  if (data.form.stereoModule && data.terminalId && data.cameraId) {
+    destroyMediaPlayer('getDepthImageStream');
+    setSubscription(true, 'getDepthImageStream', true);
+  }
+}
+
+function handleSocketClose() {
+  destroyMediaPlayer('getDepthImageStream');
+  if (data.form.stereoModule) {
+    streamStatus.getDepthImageStream = 'retrying';
+  }
+}
+
+socket?.on('open', handleSocketOpen);
+socket?.on('close', handleSocketClose);
 
-// 监听开关变化
 watch(() => data.form.stereoModule, async (newVal) => {
   if (!data.terminalId || !data.cameraId) return;
-  await nextTick();
-  subscribeDebounced(newVal, data.terminalId, 'getDepthImageStream');
+  await setSubscription(newVal, 'getDepthImageStream');
 });
 
 watch(() => data.form.rgbCamera, async (newVal) => {
   if (!data.terminalId || !data.cameraId) return;
-  await nextTick();
-  subscribeDebounced(newVal, data.terminalId, 'getRGBImageStream');
-});
-
-// 2. 使用 watchEffect 来响应 props 和路由的变化
-watchEffect(() => {
-  let deviceId = null;
-  // 优先使用 props 中的值（如果弹窗传入了）
-  if (props.initialDeviceId) {
-    deviceId = props.initialDeviceId;
+  if (newVal) {
+    await nextTick();
+    await startDirectRgbStream();
+  } else {
+    stopDirectRgbStream();
   }
-  // 如果 props 没有传入，再从路由获取（仅当cameraId和terminalId都还没设置时）
-  // 注意：这里需要根据你实际从路由获取参数的逻辑来调整
-  // 假设 route.path.split("/")[3] 是cameraId
-  if (!props.initialDeviceId && route.path.split("/")[3]) {
-    deviceId = route.path.split("/")[3];
-    // 调用 getRegister 并设置 cameraId
-    // 注意：这里的 getRegister 函数需要能够处理从路由获取的参数
-   
-  }
-   getRegister(deviceId).then(res => {
-      data.cameraId = res.data.deviceCode;
-      data.terminalId = res.data.idDeDeviceTerminalConfig;
-    }).catch(error => {
-      console.error("Error fetching register from route:", error);
-    });
-
 });
 
-onMounted(() => {
-  // getRegister(route.path.split("/")[3]).then(res => {
-  //   data.cameraId = res.data.deviceCode;
-  //   data.terminalId = res.data.idDeDeviceTerminalConfig;
-  // });
-});
+watch(
+  () => props.initialDeviceId || route.path.split('/')[3],
+  async (deviceId) => {
+    if (!deviceId) return;
+    try {
+      const response = await getRegister(deviceId);
+      data.cameraId = response.data.deviceCode;
+      data.terminalId = props.initialTerminalId || response.data.idDeDeviceTerminalConfig;
+      if (data.form.stereoModule) await setSubscription(true, 'getDepthImageStream');
+      if (data.form.rgbCamera) await startDirectRgbStream();
+    } catch (error) {
+      console.error('获取摄像头注册信息失败:', error);
+    }
+  },
+  { immediate: true }
+);
 
 onUnmounted(() => {
-  if (socket && data.terminalId && data.cameraId) {
-    const channels = [
-      `edgeCameraServiceImpl/getRGBImageStream/${data.terminalId}/${data.cameraId}`,
-      `edgeCameraServiceImpl/getDepthImageStream/${data.terminalId}/${data.cameraId}`,
-    ];
-    channels.forEach(channel => {
-      socket.off(channel, data.callbacks[channel]);
-      console.log('卸载时清理回调:', channel);
-    });
-    data.callbacks = {};
-  }
-  blobUrls.value.forEach(url => URL.revokeObjectURL(url));
-  blobUrls.value = [];
+  socket?.off('open', handleSocketOpen);
+  socket?.off('close', handleSocketClose);
+  stopDirectRgbStream();
+  ['getDepthImageStream'].forEach(method => {
+    clearTimeout(retryTimers.get(method));
+    if (data.terminalId && data.cameraId) {
+      const channel = channelFor(method);
+      socket?.send({ type: 'channel_subscription', action: 'unsubscribe', channel });
+      if (data.callbacks[channel]) socket.off(channel, data.callbacks[channel]);
+    }
+    destroyMediaPlayer(method);
+  });
+  data.callbacks = {};
 });
 </script>
 
@@ -276,34 +492,19 @@ onUnmounted(() => {
     flex-direction: column;
     flex-grow: 1;
     overflow: hidden;
-    .single-stream-container {
-      display: flex;
-      flex-direction: column;
-      justify-content: center;
-      align-items: center;
-      flex-grow: 1;
-      .stream-wrapper {
-        width: 100%;
-        height: 100%;
-        display: flex;
-        justify-content: center;
-        align-items: center;
-        overflow: hidden;
-        position: relative;
-        video {
-          width: auto;
-          height: 100%;
-          object-fit: contain;
-          transition: opacity 0.1s ease-in-out; /* 平滑过渡 */
-        }
-      }
-    }
     .stream-container {
       display: flex;
       flex-direction: column;
       flex-grow: 1;
+      min-height: 0;
+      &.dual {
+        .stream-item {
+          min-height: 50%;
+        }
+      }
       .stream-item {
         flex: 1;
+        min-height: 0;
         display: flex;
         flex-direction: column;
         justify-content: center;
@@ -317,14 +518,27 @@ onUnmounted(() => {
           overflow: hidden;
           position: relative;
           video {
-            width: auto;
+            width: 100%;
             height: 100%;
             object-fit: contain;
-            transition: opacity 0.1s ease-in-out; /* 平滑过渡 */
+            background: #000;
+          }
+          .stream-state {
+            position: absolute;
+            top: 12px;
+            right: 12px;
+            padding: 4px 9px;
+            border: 1px solid rgba(255, 255, 255, 0.2);
+            border-radius: 4px;
+            background: rgba(0, 0, 0, 0.58);
+            color: rgba(255, 255, 255, 0.9);
+            font-size: 12px;
+            line-height: 18px;
+            pointer-events: none;
           }
         }
       }
     }
   }
 }
-</style>
\ No newline at end of file
+</style>
diff --git a/src/views/flow/components/params/BasicNodeParams.vue b/src/views/flow/components/params/BasicNodeParams.vue
index ee8eef7..b79b875 100644
--- a/src/views/flow/components/params/BasicNodeParams.vue
+++ b/src/views/flow/components/params/BasicNodeParams.vue
@@ -219,7 +219,21 @@ const testRule = (property) => {
                 } else {
                     return Promise.reject(new Error('加速度必须大于速度'))
                 }
-            }, trigger: 'blur' }
+            }, trigger: 'change' }
+        ]
+
+    }
+
+    if (property?.name === 'velocity') {
+        return [
+            { required: true, message: '请输入参数值', trigger: 'blur' },
+            { validator: () => {
+                if (formData.nodeParams.find(param => param.name === 'acceleration')?.input > formData.nodeParams.find(param => param.name === 'velocity')?.input) {
+                    return Promise.resolve()
+                } else {
+                    return Promise.reject(new Error('加速度必须大于速度'))
+                }
+            }, trigger: 'change' }
         ]
 
     }