diff --git a/docs/04-architecture.md b/docs/04-architecture.md index 1c63c78..392b599 100644 --- a/docs/04-architecture.md +++ b/docs/04-architecture.md @@ -58,6 +58,7 @@ V2 对应模块的当前落实与 T-304 目标如下： | ONNX Pose | `v2/internal/pose` | 114 letterbox、RGB/CHW、NMS、关键点及坐标还原 | 人员 ID、摔倒结论 | | 事件引擎 | `v2/internal/fall` | 复现 V1 的跟踪、证据、四态状态机和 `FallEvent` | 视频解码、声音、文件 | | 画面与证据 | `v2/internal/render`、`v2/internal/alert` | 叠加人员骨架/状态，红色仅表示 CONFIRMED；按事件 ID 保存一次 PNG 和 JSONL | 读取 RTSP、重跑事件规则、直接操作 Walk 控件 | +| 实时装配 | `v2/internal/monitor` | 在后台串联帧→Pose→事件→渲染→证据；画面只保留最新完成帧，确认事件经独立队列送 UI | 直接操作 Walk 控件、保存凭证 | | Windows UI | `v2/internal/ui` | Walk 顶部“监控/设置”Tab、渲染最新帧和已计算状态 | 直接读 RTSP、执行 ONNX 或事件规则 | | 应用装配 | `v2/cmd/silver-pose`（T-304） | 管理 worker 生命周期、取消、最新帧投递和依赖注入 | 重写领域规则 | 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 026412a..3d32a58 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 @@ -208,7 +208,7 @@ git commit -m "feat(v2): add annotated evidence and alert dispatch" - Create: `v2/internal/monitor/monitor_test.go` - Modify: `v2/internal/spike/ort.go` only if an interface adapter is needed -- [ ] **Step 1: Write failing orchestration tests using fakes** +- [x] **Step 1: Write failing orchestration tests using fakes** ```go func TestMonitorPublishesLatestCompletedFrameAndOneAlert(t *testing.T) { @@ -227,12 +227,12 @@ func TestMonitorDoesNotCreateFallEvidenceOnSourceFailure(t *testing.T) { } ``` -- [ ] **Step 2: Verify the monitor tests fail** +- [x] **Step 2: Verify the monitor tests fail** Run: `Set-Location v2; go test ./internal/monitor -v` Expected: FAIL because `internal/monitor` does not exist. -- [ ] **Step 3: Implement the composition boundary** +- [x] **Step 3: Implement the composition boundary** ```go type PoseRuntime interface { Run([]float32) ([]float32, error); Close() } @@ -244,12 +244,12 @@ func (m *Monitor) Run(ctx context.Context) { /* source frame -> pose -> fall -> Create/open one ONNX Runtime instance before consuming frames and close it only when monitoring stops. Treat Pose/preprocess errors as non-fall status updates and continue/reconnect according to source state; do not terminate the Walk message loop. Only completed rendered frames enter the one-slot `Updates` channel, so a slow UI never queues stale video frames. -- [ ] **Step 4: Run monitor, fall and full Go tests** +- [x] **Step 4: Run monitor, fall and full Go tests** Run: `Set-Location v2; $env:CGO_ENABLED='1'; go test ./internal/monitor ./internal/fall ./...` Expected: PASS. -- [ ] **Step 5: Commit the monitor slice** +- [x] **Step 5: Commit the monitor slice** ```powershell git add v2/internal/monitor v2/internal/spike/ort.go diff --git a/v2/internal/monitor/monitor.go b/v2/internal/monitor/monitor.go new file mode 100644 index 0000000..59a1d03 --- /dev/null +++ b/v2/internal/monitor/monitor.go @@ -0,0 +1,191 @@ +// Package monitor composes source, pose, fall and evidence work off the UI thread. +package monitor + +import ( + "context" + "fmt" + "image" + "sync" + "time" + + "silverpose/v2/internal/alert" + "silverpose/v2/internal/fall" + "silverpose/v2/internal/pose" + "silverpose/v2/internal/render" + "silverpose/v2/internal/source" +) + +type FrameSource interface { + Frames() <-chan source.Frame + Statuses() <-chan source.Update +} + +type PoseRuntime interface { + Run([]float32) ([]float32, error) + Close() +} + +type EventProcessor interface { + Process(fall.Frame) fall.FrameResult +} + +type Update struct { + Image *image.RGBA + Result fall.FrameResult + Events []alert.Record + SourceStatus source.Status + SourceMessage string + Sequence uint64 +} + +type Monitor struct { + source FrameSource + runtime PoseRuntime + processor EventProcessor + confidence float32 + dispatcher *alert.Dispatcher + clock func() time.Time + updates chan Update + alerts chan alert.Record + done chan struct{} + closeOnce sync.Once +} + +func New( + stream FrameSource, + runtime PoseRuntime, + processor EventProcessor, + confidence float32, + dispatcher *alert.Dispatcher, + clock func() time.Time, +) (*Monitor, error) { + if stream == nil || runtime == nil || processor == nil || dispatcher == nil { + return nil, fmt.Errorf("monitor dependencies must be non-nil") + } + if confidence < 0 || confidence > 1 { + return nil, fmt.Errorf("pose confidence must be between 0 and 1") + } + if clock == nil { + clock = time.Now + } + return &Monitor{ + source: stream, runtime: runtime, processor: processor, confidence: confidence, + dispatcher: dispatcher, clock: clock, updates: make(chan Update, 1), alerts: make(chan alert.Record, 16), done: make(chan struct{}), + }, nil +} + +func (monitor *Monitor) Updates() <-chan Update { return monitor.updates } +func (monitor *Monitor) Done() <-chan struct{} { return monitor.done } +func (monitor *Monitor) Alerts() <-chan alert.Record { return monitor.alerts } + +// Run blocks until the source closes or the supplied context is cancelled. +// The caller owns the goroutine; all expensive work stays out of Walk's UI thread. +func (monitor *Monitor) Run(ctx context.Context) { + defer monitor.close() + defer monitor.runtime.Close() + frames := monitor.source.Frames() + statuses := monitor.source.Statuses() + currentStatus := source.Connecting + currentMessage := "正在连接视频流" + for frames != nil || statuses != nil { + select { + case <-ctx.Done(): + return + case status, open := <-statuses: + if !open { + statuses = nil + continue + } + currentStatus, currentMessage = status.Status, status.Message + monitor.publish(Update{SourceStatus: currentStatus, SourceMessage: currentMessage}) + case frame, open := <-frames: + if !open { + frames = nil + continue + } + statuses, currentStatus, currentMessage = monitor.drainStatuses(statuses, currentStatus, currentMessage) + monitor.processFrame(frame, currentStatus, currentMessage) + } + } +} + +func (monitor *Monitor) drainStatuses(statuses <-chan source.Update, current source.Status, message string) (<-chan source.Update, source.Status, string) { + for statuses != nil { + select { + case status, open := <-statuses: + if !open { + return nil, current, message + } + current, message = status.Status, status.Message + default: + return statuses, current, message + } + } + return nil, current, message +} + +func (monitor *Monitor) processFrame(frame source.Frame, status source.Status, message string) { + input, transform, err := pose.PreprocessBGR(frame.BGR, frame.Width, frame.Height, 640) + if err != nil { + monitor.publish(Update{SourceStatus: status, SourceMessage: "视频帧格式无效", Sequence: frame.Sequence}) + return + } + output, err := monitor.runtime.Run(input) + if err != nil { + monitor.publish(Update{SourceStatus: status, SourceMessage: "姿态推理失败", Sequence: frame.Sequence}) + return + } + people, err := pose.ParseYOLOv8Pose(output, transform, monitor.confidence, 0.70) + if err != nil { + monitor.publish(Update{SourceStatus: status, SourceMessage: "姿态结果无效", Sequence: frame.Sequence}) + return + } + result := monitor.processor.Process(fall.Frame{ + Timestamp: frame.Timestamp.Seconds(), Width: frame.Width, Height: frame.Height, Poses: people, + }) + capturedAt := monitor.clock().UTC() + canvas, err := render.Render(frame.BGR, frame.Width, frame.Height, result, capturedAt) + if err != nil { + monitor.publish(Update{Result: result, SourceStatus: status, SourceMessage: "画面渲染失败", Sequence: frame.Sequence}) + return + } + records := make([]alert.Record, 0, len(result.Events)) + for _, event := range result.Events { + record, err := monitor.dispatcher.Dispatch(event, canvas, capturedAt) + if err != nil { + monitor.publish(Update{Image: canvas, Result: result, SourceStatus: status, SourceMessage: "事件证据保存失败", Sequence: frame.Sequence}) + return + } + if record.Written { + records = append(records, record) + monitor.alerts <- record + } + } + monitor.publish(Update{ + Image: canvas, Result: result, Events: records, SourceStatus: status, SourceMessage: message, Sequence: frame.Sequence, + }) +} + +func (monitor *Monitor) publish(update Update) { + select { + case monitor.updates <- update: + return + default: + } + select { + case <-monitor.updates: + default: + } + select { + case monitor.updates <- update: + default: + } +} + +func (monitor *Monitor) close() { + monitor.closeOnce.Do(func() { + close(monitor.updates) + close(monitor.alerts) + close(monitor.done) + }) +} diff --git a/v2/internal/monitor/monitor_test.go b/v2/internal/monitor/monitor_test.go new file mode 100644 index 0000000..ffb53d2 --- /dev/null +++ b/v2/internal/monitor/monitor_test.go @@ -0,0 +1,103 @@ +package monitor + +import ( + "context" + "errors" + "testing" + "time" + + "silverpose/v2/internal/alert" + "silverpose/v2/internal/fall" + "silverpose/v2/internal/pose" + "silverpose/v2/internal/source" +) + +func TestMonitorPublishesLatestCompletedFrameAndOneAlert(t *testing.T) { + frames := make(chan source.Frame, 2) + frames <- source.Frame{BGR: []byte{1, 1, 1, 1, 1, 1}, Width: 2, Height: 1, Sequence: 1} + frames <- source.Frame{BGR: []byte{2, 2, 2, 2, 2, 2}, Width: 2, Height: 1, Sequence: 2} + close(frames) + statuses := make(chan source.Update, 1) + statuses <- source.Update{Status: source.Connected, Message: "视频流已连接"} + close(statuses) + processor := &fakeProcessor{result: fall.FrameResult{ + People: []fall.PersonAnalysis{{State: fall.Confirmed}}, + Events: []fall.Event{{EventID: "FALL-run-000001", TrackID: "P-0001", ConfigVersion: "cfg-safe", State: fall.Confirmed}}, + }} + runtime := &fakeRuntime{output: make([]float32, pose.YOLOPoseOutputValues)} + monitor, err := New(fakeSource{frames: frames, statuses: statuses}, runtime, processor, 0.25, alert.NewDispatcher(t.TempDir(), "lobby"), func() time.Time { + return time.Date(2026, 7, 22, 8, 0, 0, 0, time.UTC) + }) + if err != nil { + t.Fatal(err) + } + go monitor.Run(context.Background()) + <-monitor.Done() + + update, ok := <-monitor.Updates() + if !ok { + t.Fatal("monitor emitted no update") + } + if update.Sequence != 2 || update.Image == nil || update.SourceStatus != source.Connected { + t.Fatalf("update = %#v", update) + } + record, ok := <-monitor.Alerts() + if !ok || !record.Written || record.Event.EventID != "FALL-run-000001" { + t.Fatalf("alert record = %#v, open = %t", record, ok) + } + if !runtime.closed { + t.Fatal("runtime was not closed") + } +} + +func TestMonitorDoesNotCreateFallEvidenceOnSourceFailure(t *testing.T) { + frames := make(chan source.Frame) + close(frames) + statuses := make(chan source.Update, 1) + statuses <- source.Update{Status: source.Retrying, Message: "视频流已断开，正在重连"} + close(statuses) + processor := &fakeProcessor{} + monitor, err := New(fakeSource{frames: frames, statuses: statuses}, &fakeRuntime{}, processor, 0.25, alert.NewDispatcher(t.TempDir(), "lobby"), time.Now) + if err != nil { + t.Fatal(err) + } + go monitor.Run(context.Background()) + <-monitor.Done() + + update := <-monitor.Updates() + if update.SourceStatus != source.Retrying || len(update.Events) != 0 || processor.calls != 0 { + t.Fatalf("update = %#v, process calls = %d", update, processor.calls) + } +} + +type fakeSource struct { + frames <-chan source.Frame + statuses <-chan source.Update +} + +func (source fakeSource) Frames() <-chan source.Frame { return source.frames } +func (source fakeSource) Statuses() <-chan source.Update { return source.statuses } + +type fakeRuntime struct { + output []float32 + err error + closed bool +} + +func (runtime *fakeRuntime) Run(input []float32) ([]float32, error) { + if len(input) == 0 { + return nil, errors.New("missing input") + } + return runtime.output, runtime.err +} +func (runtime *fakeRuntime) Close() { runtime.closed = true } + +type fakeProcessor struct { + result fall.FrameResult + calls int +} + +func (processor *fakeProcessor) Process(frame fall.Frame) fall.FrameResult { + processor.calls++ + return processor.result +}