diff --git a/docs/03-tech-stack.md b/docs/03-tech-stack.md index 78f7a3e..e4d37b4 100644 --- a/docs/03-tech-stack.md +++ b/docs/03-tech-stack.md @@ -24,7 +24,7 @@ | 语言 | Go | 已定 | 面向长时间 RTSP 运行、并发控制和单一交付物。 | | 模型工件 | 固定输入尺寸的 ONNX Pose 模型 | 已定 | Go 不直接加载 `.pt`；模型由已验收 V1 权重导出并记录哈希。 | | 推理运行时 | `github.com/yalue/onnxruntime_go` v1.31.0 + ONNX Runtime CPU DLL | Spike 已验证 | Windows 上以显式 DLL 路径加载锁定的 1×3×640×640 Pose ONNX，已完成一次真实录像首帧推理；该包装器的 C API 为 26，V2 构建必须启用 CGO/MinGW，并在发布包锁定匹配 DLL 的来源与 SHA-256。 | -| 视频输入 | 受控的 `ffmpeg.exe` 子进程输出 `bgr24` 原始帧 | 回放已验证 | Go 回归命令已完整读取 317 帧本地录像；避免未配置的 GoCV/OpenCV 开发绑定。T-304 必须随应用锁定 FFmpeg/FFprobe 的版本、哈希与 RTSP 重连行为。 | +| 视频输入 | 受控的 `ffmpeg.exe` 子进程输出 `bgr24` 原始帧 | 回放已验证；RTSP 源单元测试已覆盖 | Go 回归命令已完整读取 317 帧本地录像；`internal/source` 已实现 FFprobe 元数据探测、FFmpeg 解码、有界重连、取消与最新帧投递，且错误状态不含 RTSP URL。避免未配置的 GoCV/OpenCV 开发绑定。T-304 仍须随应用锁定 FFmpeg/FFprobe 的版本、哈希，并在现场验证真实流重连。 | | 图像预处理与 Pose 解析 | Go 实现的 BGR→RGB、CHW、640 letterbox、NMS、17 点及坐标还原 | 合成契约与本机漏检回放已验证 | 采用 Ultralytics 的 114 补边、双线性缩放后转 uint8 的语义；Go 首帧输出两个人，与 V1 对应结果的 box 坐标差在已记录的 6 像素容差内。`demo/1.mp4` 是真实摔倒但在现有 V1 配置下漏检，不能称为“无报警反例”；仍需经同意、符合固定俯视大厅/走廊边界的正例录像完成事件级确认。 | | 演示 UI | `github.com/lxn/walk`（Windows 原生控件） | Spike 已验证 | 双 Tab 骨架可在 `CGO_ENABLED=0` 编译，避免 Fyne/OpenGL 首次编译的高复杂度；正式版仍须实现浅灰蓝底、白色卡片，红色仅表示确认摔倒。 | | 测试 | `go test ./...` + V1 回归录像 | 已定 | V2 行为必须与 V1 基线对照。 | diff --git a/docs/04-architecture.md b/docs/04-architecture.md index ddb39eb..16e53bf 100644 --- a/docs/04-architecture.md +++ b/docs/04-architecture.md @@ -54,6 +54,7 @@ V2 对应模块的当前落实与 T-304 目标如下： | --- | --- | --- | --- | | 回放解码器 | `v2/cmd/regression` | 用受控 FFmpeg 顺序读取 BGR 帧，供本机录像回归 | RTSP 重连、UI 绘制 | | 运行配置 | `v2/internal/config` | 校验公开 JSON、仅从环境变量解析 RTSP URL、生成非敏感配置版本 | 保存或显示 URL、账号、密码 | +| 实时视频源 | `v2/internal/source` | 用 FFprobe 获取尺寸/FPS，以 FFmpeg 输出 BGR 帧；有界重连、取消、来源状态和单槽最新帧投递 | ONNX、事件判定、UI 绘制；错误状态不得含 URL | | ONNX Pose | `v2/internal/pose` | 114 letterbox、RGB/CHW、NMS、关键点及坐标还原 | 人员 ID、摔倒结论 | | 事件引擎 | `v2/internal/fall` | 复现 V1 的跟踪、证据、四态状态机和 `FallEvent` | 视频解码、声音、文件 | | Windows UI | `v2/internal/ui` | Walk 顶部“监控/设置”Tab、渲染最新帧和已计算状态 | 直接读 RTSP、执行 ONNX 或事件规则 | diff --git a/docs/superpowers/plans/2026-07-22-t304-go-demo-delivery.md b/docs/superpowers/plans/2026-07-22-t304-go-demo-delivery.md index 7d5c978..9734717 100644 --- a/docs/superpowers/plans/2026-07-22-t304-go-demo-delivery.md +++ b/docs/superpowers/plans/2026-07-22-t304-go-demo-delivery.md @@ -98,7 +98,7 @@ git commit -m "feat(v2): add safe runtime configuration" - Create: `v2/internal/source/stream_test.go` - Modify: `docs/04-architecture.md` -- [ ] **Step 1: Write deterministic source tests before subprocess integration** +- [x] **Step 1: Write deterministic source tests before subprocess integration** ```go func TestLatestFrameDropsStaleFrame(t *testing.T) { @@ -114,12 +114,12 @@ func TestBackoffIsBoundedAndCancellationStopsRetry(t *testing.T) { } ``` -- [ ] **Step 2: Verify the tests fail** +- [x] **Step 2: Verify the tests fail** Run: `Set-Location v2; go test ./internal/source -run 'Test(Latest|Backoff)' -v` Expected: FAIL because `internal/source` does not exist. -- [ ] **Step 3: Implement probe and worker ownership** +- [x] **Step 3: Implement probe and worker ownership** ```go type Status string @@ -136,14 +136,12 @@ func Start(ctx context.Context, cfg Config, commands CommandFactory) (Stream, er Use `-rtsp_transport`, `-rw_timeout`, low-latency FFmpeg input flags when configured, and `-f rawvideo -pix_fmt bgr24 -` output. Capture stderr internally but publish only a generic status code/message such as `视频流已断开，正在重连`; it must never contain command arguments or the RTSP URL. Cancellation kills the current process and closes channels exactly once. -- [ ] **Step 4: Run source tests and a safe invalid-source smoke** +- [x] **Step 4: Run source tests** Run: `Set-Location v2; go test ./internal/source -v` -Expected: PASS. -Run: `go run ./cmd/silver-pose --config config.example.json` -Expected: clean missing-environment configuration error without an RTSP address. +Expected: PASS. The production command's missing-environment smoke is performed in Task 6 after `cmd/silver-pose` exists. -- [ ] **Step 5: Commit the source slice** +- [x] **Step 5: Commit the source slice** ```powershell git add v2/internal/source docs/04-architecture.md diff --git a/v2/internal/source/stream.go b/v2/internal/source/stream.go new file mode 100644 index 0000000..a4444a9 --- /dev/null +++ b/v2/internal/source/stream.go @@ -0,0 +1,288 @@ +// Package source owns the FFprobe/FFmpeg stream lifecycle for one camera. +package source + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "os/exec" + "strconv" + "strings" + "sync" + "time" +) + +type Status string + +const ( + Connecting Status = "CONNECTING" + Connected Status = "CONNECTED" + Retrying Status = "RETRYING" + Stopped Status = "STOPPED" +) + +// Config is intentionally separate from the persisted configuration. SourceURL +// exists only in process memory after config resolves its named environment variable. +type Config struct { + SourceURL string + FFmpegPath string + FFprobePath string + Transport string + Timeout time.Duration + LowLatency bool +} + +type Metadata struct { + Width int + Height int + FPS float64 +} + +type Frame struct { + BGR []byte + Width int + Height int + Timestamp time.Duration + Sequence uint64 +} + +type Update struct { + Status Status + Message string + Attempt int +} + +// Dependencies make lifecycle behavior testable without an FFmpeg installation +// or a real source URL. Nil functions use the safe default subprocess adapter. +type Dependencies struct { + Probe func(context.Context, Config) (Metadata, error) + Decode func(context.Context, Config, Metadata, func([]byte)) error + Wait func(context.Context, time.Duration) error +} + +type Stream struct { + frames chan Frame + statuses chan Update + done chan struct{} + once sync.Once +} + +// Start begins one reconnecting stream worker. Consumers receive at most one +// pending frame: a slow inference/UI path always observes the most recent frame. +func Start(ctx context.Context, config Config, dependencies Dependencies) *Stream { + stream := &Stream{ + frames: make(chan Frame, 1), + statuses: make(chan Update, 16), + done: make(chan struct{}), + } + if dependencies.Probe == nil { + dependencies.Probe = probe + } + if dependencies.Decode == nil { + dependencies.Decode = decode + } + if dependencies.Wait == nil { + dependencies.Wait = waitContext + } + go stream.run(ctx, config, dependencies) + return stream +} + +func (stream *Stream) Frames() <-chan Frame { return stream.frames } +func (stream *Stream) Statuses() <-chan Update { return stream.statuses } +func (stream *Stream) Done() <-chan struct{} { return stream.done } + +func (stream *Stream) run(ctx context.Context, config Config, dependencies Dependencies) { + defer stream.close() + started := time.Now() + attempt := 0 + sequence := uint64(0) + for { + if ctx.Err() != nil { + stream.publishStatus(Update{Status: Stopped, Message: "视频流已停止", Attempt: attempt}) + return + } + stream.publishStatus(Update{Status: Connecting, Message: "正在连接视频流", Attempt: attempt}) + metadata, err := dependencies.Probe(ctx, config) + if err == nil { + stream.publishStatus(Update{Status: Connected, Message: "视频流已连接", Attempt: attempt}) + err = dependencies.Decode(ctx, config, metadata, func(bgr []byte) { + sequence++ + publishLatest(stream.frames, Frame{ + BGR: append([]byte(nil), bgr...), + Width: metadata.Width, + Height: metadata.Height, + Timestamp: time.Since(started), + Sequence: sequence, + }) + }) + } + if ctx.Err() != nil { + stream.publishStatus(Update{Status: Stopped, Message: "视频流已停止", Attempt: attempt}) + return + } + stream.publishStatus(Update{Status: Retrying, Message: "视频流已断开，正在重连", Attempt: attempt}) + if err := dependencies.Wait(ctx, retryDelay(attempt)); err != nil { + stream.publishStatus(Update{Status: Stopped, Message: "视频流已停止", Attempt: attempt}) + return + } + attempt++ + } +} + +func (stream *Stream) close() { + stream.once.Do(func() { + close(stream.frames) + close(stream.statuses) + close(stream.done) + }) +} + +func (stream *Stream) publishStatus(update Update) { + select { + case stream.statuses <- update: + default: + // Status is advisory. A saturated observer must not block source cleanup. + } +} + +func publishLatest(frames chan Frame, frame Frame) { + select { + case frames <- frame: + return + default: + } + select { + case <-frames: + default: + } + select { + case frames <- frame: + default: + } +} + +func retryDelay(attempt int) time.Duration { + if attempt <= 0 { + return time.Second + } + if attempt >= 3 { + return 8 * time.Second + } + return time.Second << attempt +} + +func waitContext(ctx context.Context, duration time.Duration) error { + timer := time.NewTimer(duration) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + +func probe(ctx context.Context, config Config) (Metadata, error) { + if strings.TrimSpace(config.FFprobePath) == "" { + return Metadata{}, errors.New("FFprobe executable is not configured") + } + if strings.TrimSpace(config.SourceURL) == "" { + return Metadata{}, errors.New("video source is not configured") + } + args := []string{"-v", "error", "-select_streams", "v:0", "-show_entries", "stream=width,height,r_frame_rate", "-of", "json"} + args = appendInputOptions(args, config) + args = append(args, "-i", config.SourceURL) + output, err := exec.CommandContext(ctx, config.FFprobePath, args...).Output() + if err != nil { + return Metadata{}, errors.New("unable to probe video metadata") + } + var parsed struct { + Streams []struct { + Width int `json:"width"` + Height int `json:"height"` + FrameRate string `json:"r_frame_rate"` + } `json:"streams"` + } + if err := json.Unmarshal(output, &parsed); err != nil || len(parsed.Streams) == 0 { + return Metadata{}, errors.New("invalid video metadata") + } + metadata := Metadata{Width: parsed.Streams[0].Width, Height: parsed.Streams[0].Height} + metadata.FPS = parseFrameRate(parsed.Streams[0].FrameRate) + if metadata.Width <= 0 || metadata.Height <= 0 || metadata.FPS <= 0 { + return Metadata{}, errors.New("invalid video dimensions or frame rate") + } + return metadata, nil +} + +func decode(ctx context.Context, config Config, metadata Metadata, publish func([]byte)) error { + if strings.TrimSpace(config.FFmpegPath) == "" { + return errors.New("FFmpeg executable is not configured") + } + if strings.TrimSpace(config.SourceURL) == "" { + return errors.New("video source is not configured") + } + if metadata.Width <= 0 || metadata.Height <= 0 { + return errors.New("invalid video dimensions") + } + args := append([]string{"-hide_banner", "-loglevel", "error"}, appendInputOptions(nil, config)...) + args = append(args, "-i", config.SourceURL, "-an", "-sn", "-dn", "-f", "rawvideo", "-pix_fmt", "bgr24", "-") + command := exec.CommandContext(ctx, config.FFmpegPath, args...) + command.Stderr = io.Discard + stdout, err := command.StdoutPipe() + if err != nil { + return errors.New("unable to read video frames") + } + if err := command.Start(); err != nil { + return errors.New("unable to start video decoder") + } + frameSize := metadata.Width * metadata.Height * 3 + frame := make([]byte, frameSize) + for { + _, err := io.ReadFull(stdout, frame) + if err != nil { + _ = command.Wait() + if ctx.Err() != nil { + return ctx.Err() + } + return errors.New("video decoder stopped") + } + publish(frame) + } +} + +func appendInputOptions(args []string, config Config) []string { + transport := strings.ToLower(strings.TrimSpace(config.Transport)) + if transport == "tcp" || transport == "udp" { + args = append(args, "-rtsp_transport", transport) + } + if config.Timeout > 0 { + args = append(args, "-rw_timeout", strconv.FormatInt(config.Timeout.Microseconds(), 10)) + } + if config.LowLatency { + args = append(args, "-fflags", "nobuffer", "-flags", "low_delay") + } + return args +} + +func parseFrameRate(value string) float64 { + parts := strings.Split(value, "/") + if len(parts) == 2 { + numerator, numeratorErr := strconv.ParseFloat(parts[0], 64) + denominator, denominatorErr := strconv.ParseFloat(parts[1], 64) + if numeratorErr == nil && denominatorErr == nil && denominator > 0 { + return numerator / denominator + } + } + fps, err := strconv.ParseFloat(value, 64) + if err != nil || fps <= 0 { + return 0 + } + return fps +} + +func (metadata Metadata) String() string { + return fmt.Sprintf("%dx%d@%.3f", metadata.Width, metadata.Height, metadata.FPS) +} diff --git a/v2/internal/source/stream_test.go b/v2/internal/source/stream_test.go new file mode 100644 index 0000000..a171fea --- /dev/null +++ b/v2/internal/source/stream_test.go @@ -0,0 +1,92 @@ +package source + +import ( + "context" + "errors" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestPublishLatestDropsStaleFrame(t *testing.T) { + frames := make(chan Frame, 1) + publishLatest(frames, Frame{Sequence: 1}) + publishLatest(frames, Frame{Sequence: 2}) + + if got := <-frames; got.Sequence != 2 { + t.Fatalf("sequence = %d, want 2", got.Sequence) + } +} + +func TestRetryDelayIsBounded(t *testing.T) { + if got := retryDelay(0); got != time.Second { + t.Fatalf("first retry delay = %s", got) + } + if got := retryDelay(3); got != 8*time.Second { + t.Fatalf("fourth retry delay = %s", got) + } + if got := retryDelay(99); got != 8*time.Second { + t.Fatalf("bounded retry delay = %s", got) + } +} + +func TestStreamRetriesAndPublishesLatestCompletedFrame(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var decodeCalls int32 + decodedLatest := make(chan struct{}) + stream := Start(ctx, Config{SourceURL: "rtsp://operator:secret@example/101"}, Dependencies{ + Probe: func(context.Context, Config) (Metadata, error) { + return Metadata{Width: 2, Height: 1, FPS: 30}, nil + }, + Decode: func(ctx context.Context, _ Config, _ Metadata, publish func([]byte)) error { + call := atomic.AddInt32(&decodeCalls, 1) + if call == 1 { + publish([]byte{1, 1, 1, 1, 1, 1}) + return errors.New("temporary decoder failure") + } + publish([]byte{2, 2, 2, 2, 2, 2}) + close(decodedLatest) + cancel() + return ctx.Err() + }, + Wait: func(ctx context.Context, _ time.Duration) error { return ctx.Err() }, + }) + + select { + case <-decodedLatest: + case <-time.After(time.Second): + t.Fatal("decoder did not publish the second frame") + } + select { + case got := <-stream.Frames(): + if got.Sequence != 2 || got.BGR[0] != 2 { + t.Fatalf("frame = %#v", got) + } + case <-time.After(time.Second): + t.Fatal("did not receive latest decoded frame") + } + <-stream.Done() +} + +func TestStreamNeverPublishesDecoderErrorDetails(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + stream := Start(ctx, Config{SourceURL: "rtsp://operator:secret@example/101"}, Dependencies{ + Probe: func(context.Context, Config) (Metadata, error) { + return Metadata{}, errors.New("ffprobe rtsp://operator:secret@example/101 failed") + }, + Decode: func(context.Context, Config, Metadata, func([]byte)) error { return nil }, + Wait: func(ctx context.Context, _ time.Duration) error { + cancel() + return ctx.Err() + }, + }) + + for update := range stream.Statuses() { + if strings.Contains(update.Message, "secret") || strings.Contains(update.Message, "rtsp://") { + t.Fatalf("status leaked source details: %#v", update) + } + } +}