inspection-host-computer/server/cameraVideoStream.js

import { spawn } from 'child_process'
import { createCameraGrpcClient } from './grpcClient.js'

const FIRST_FRAME_TIMEOUT_MS = 10000
const FRAME_TYPES = ['U8C1', 'U16C1', 'U8C3', 'U16C3', 'F16C1', 'F32C1']

function normalizeCodec(codec = '') {
  return codec.toLowerCase().replace(/[^a-z0-9]/g, '')
}

function inputArguments(frame) {
  const codec = normalizeCodec(frame.codec)

  if (['jpeg', 'jpg', 'mjpeg', 'imagejpeg'].includes(codec)) return ['-f', 'mjpeg']
  if (['png', 'imagepng'].includes(codec)) return ['-f', 'image2pipe', '-vcodec', 'png']
  if (['h264', 'avc', 'avc1'].includes(codec)) return ['-f', 'h264']
  if (['h265', 'hevc', 'hev1', 'hvc1'].includes(codec)) return ['-f', 'hevc']

  const width = Number(frame.width)
  const height = Number(frame.height)
  if (!Number.isInteger(width) || width <= 0 || !Number.isInteger(height) || height <= 0) {
    throw new Error(`未压缩视频帧的尺寸无效: ${frame.width}x${frame.height}`)
  }

  const declaredPixelFormats = {
    gray: 'gray',
    gray8: 'gray',
    gray16: 'gray16le',
    gray16le: 'gray16le',
    rgb24: 'rgb24',
    bgr24: 'bgr24',
    rgb48: 'rgb48le',
    rgb48le: 'rgb48le',
    bgr48: 'bgr48le',
    bgr48le: 'bgr48le',
    grayf32: 'grayf32le',
    grayf32le: 'grayf32le'
  }
  const rawPixelFormats = {
    U8C1: 'gray',
    U16C1: 'gray16le',
    U8C3: 'rgb24',
    U16C3: 'rgb48le',
    F32C1: 'grayf32le'
  }
  const frameType = typeof frame.type === 'number' ? FRAME_TYPES[frame.type] : frame.type
  const pixelFormat = declaredPixelFormats[codec] || rawPixelFormats[frameType]
  if (!pixelFormat) {
    throw new Error(`暂不支持未压缩帧类型 ${frameType || frame.type}，请让相机返回 JPEG/H.264/H.265`)
  }

  return [
    '-f', 'rawvideo',
    '-pixel_format', pixelFormat,
    '-video_size', `${width}x${height}`,
    '-framerate', String(Number(frame.source_fps) || 25)
  ]
}

function createTranscoder(frame) {
  const fps = Math.max(1, Number(frame.source_fps) || 25)
  const ffmpeg = process.env.FFMPEG_PATH || 'ffmpeg'
  const args = [
    '-hide_banner', '-loglevel', 'warning',
    ...inputArguments(frame),
    '-i', 'pipe:0',
    '-an',
    '-c:v', 'libx264',
    '-preset', 'ultrafast',
    '-tune', 'zerolatency',
    '-pix_fmt', 'yuv420p',
    '-g', String(fps),
    '-keyint_min', String(fps),
    '-sc_threshold', '0',
    '-f', 'mp4',
    '-movflags', 'frag_keyframe+empty_moov+default_base_moof',
    'pipe:1'
  ]
  return spawn(ffmpeg, args, { windowsHide: true, stdio: ['pipe', 'pipe', 'pipe'] })
}

function message(error) {
  return error?.details || error?.message || String(error)
}

export function streamCameraVideo(req, res) {
  const ip = typeof req.query.ip === 'string' ? req.query.ip.trim() : ''
  const deviceId = typeof req.query.deviceId === 'string' ? req.query.deviceId.trim() : ''
  if (!ip || !deviceId) {
    res.status(400).json({ code: 400, message: 'ip 和 deviceId 不能为空' })
    return
  }

  const client = createCameraGrpcClient(ip)
  const grpcStream = client.getRgbImageStream()
  let transcoder
  let closed = false
  let responseStarted = false
  let stderr = ''

  const close = () => {
    if (closed) return
    closed = true
    clearTimeout(firstFrameTimer)
    grpcStream.cancel()
    client.close()
    if (transcoder && !transcoder.killed) {
      transcoder.stdin.destroy()
      transcoder.kill()
    }
  }

  const fail = (error) => {
    if (closed) return
    console.error('[camera] video stream failed:', message(error))
    if (!responseStarted) {
      res.status(502).json({ code: 502, message: `相机视频流不可用: ${message(error)}` })
    } else {
      res.destroy(error instanceof Error ? error : undefined)
    }
    close()
  }

  const firstFrameTimer = setTimeout(
    () => fail(new Error('等待相机首帧超时')),
    FIRST_FRAME_TIMEOUT_MS
  )

  grpcStream.on('data', (feedback) => {
    if (closed) return
    if (feedback?.header?.success === false) {
      fail(new Error(feedback.header.error_message || '相机拒绝视频流请求'))
      return
    }

    const frame = feedback?.color_frame
    if (!frame?.data?.length) return

    if (!transcoder) {
      clearTimeout(firstFrameTimer)
      try {
        transcoder = createTranscoder(frame)
      } catch (error) {
        fail(error)
        return
      }

      res.status(200)
      res.set({
        'Content-Type': 'video/mp4',
        'Cache-Control': 'no-store, no-cache, must-revalidate',
        Connection: 'keep-alive',
        'X-Accel-Buffering': 'no'
      })
      responseStarted = true
      transcoder.stdout.pipe(res)
      transcoder.stderr.on('data', (chunk) => {
        stderr = (stderr + chunk.toString()).slice(-4000)
      })
      transcoder.stdin.on('error', (error) => {
        if (!closed && error.code !== 'EPIPE') fail(error)
      })
      transcoder.on('error', fail)
      transcoder.on('exit', (code, signal) => {
        if (!closed && code !== 0) {
          fail(new Error(`FFmpeg 异常退出 (${code ?? signal}): ${stderr.trim()}`))
        }
      })
    }

    if (!transcoder.stdin.write(frame.data)) {
      grpcStream.pause()
      transcoder.stdin.once('drain', () => grpcStream.resume())
    }
  })

  grpcStream.on('error', (error) => {
    // CANCELLED 是浏览器断开后主动清理产生的正常状态。
    if (!closed && error.code !== 1) fail(error)
  })
  grpcStream.on('end', () => {
    if (!closed) {
      transcoder?.stdin.end()
      if (!responseStarted) fail(new Error('相机未返回任何视频帧'))
    }
  })

  res.on('close', close)
  grpcStream.write({
    header: { device_id: deviceId },
    eof: false
  })
}