diff --git a/agent/internal/httpapi/alarms.go b/agent/internal/httpapi/alarms.go new file mode 100644 index 0000000..137d9e7 --- /dev/null +++ b/agent/internal/httpapi/alarms.go @@ -0,0 +1,173 @@ +package httpapi + +import ( + "bufio" + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "strconv" + "strings" + "time" + + "rk3588sys/agent/internal/files" +) + +func (s *Server) alarmsPath() string { + return filepath.Join(s.baseDir, "logs", "alarms.jsonl") +} + +// handleAlarmReport receives alarm notifications from media-server. +// POST /v1/alarms/report +func (s *Server) handleAlarmReport(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + errorJSON(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + // Allow localhost requests without token (media-server alarm callback) + if !isLocalhost(r) && !s.authorize(r, true) { + errorJSON(w, http.StatusUnauthorized, "unauthorized") + return + } + + maxBytes := int64(1 << 20) // 1MB max + r.Body = http.MaxBytesReader(w, r.Body, maxBytes) + + var alarm map[string]any + if err := json.NewDecoder(r.Body).Decode(&alarm); err != nil { + errorJSON(w, http.StatusBadRequest, "invalid json: "+err.Error()) + return + } + + // Normalize: extract channel and rule_name from whatever format media-server sends + id, _ := alarm["id"].(string) + if id == "" { + id = fmt.Sprintf("alarm_%s_%s", time.Now().Format("20060102_150405"), randomHex(6)) + } + if _, ok := alarm["timestamp"]; !ok { + alarm["timestamp"] = time.Now().Format(time.RFC3339) + } + alarm["id"] = id + alarm["received_at"] = time.Now().Format(time.RFC3339) + + path := s.alarmsPath() + dir := filepath.Dir(path) + if err := files.EnsureDir(dir, 0o755); err != nil { + errorJSON(w, http.StatusInternalServerError, "internal error: "+err.Error()) + return + } + + line, err := json.Marshal(alarm) + if err != nil { + errorJSON(w, http.StatusInternalServerError, "internal error: "+err.Error()) + return + } + + f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644) + if err != nil { + errorJSON(w, http.StatusInternalServerError, "internal error: "+err.Error()) + return + } + defer f.Close() + + if _, err := fmt.Fprintf(f, "%s\n", string(line)); err != nil { + errorJSON(w, http.StatusInternalServerError, "internal error: "+err.Error()) + return + } + + s.recordAudit(r, "alarm.report", true, id) + writeJSON(w, http.StatusOK, map[string]any{"ok": true, "id": id}) +} + +// handleAlarmsRecent returns recent alarm records. +// GET /v1/alarms/recent?limit=50&since=2026-01-01T00:00:00Z +func (s *Server) handleAlarmsRecent(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + errorJSON(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + if !s.authorize(r, false) { + errorJSON(w, http.StatusUnauthorized, "unauthorized") + return + } + + limit := 50 + if v := strings.TrimSpace(r.URL.Query().Get("limit")); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 && n <= 500 { + limit = n + } + } + + alarms, err := s.readRecentAlarms(limit) + if err != nil { + errorJSON(w, http.StatusInternalServerError, "internal error: "+err.Error()) + return + } + if alarms == nil { + alarms = make([]map[string]any, 0) + } + + writeJSON(w, http.StatusOK, map[string]any{"alarms": alarms}) +} + +func (s *Server) readRecentAlarms(limit int) ([]map[string]any, error) { + path := s.alarmsPath() + f, err := os.Open(path) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, err + } + defer f.Close() + + // Read all lines and keep last N + var lines []string + scanner := bufio.NewScanner(f) + scanner.Buffer(make([]byte, 1<<20), 1<<20) // 1MB max line + for scanner.Scan() { + line := scanner.Text() + if strings.TrimSpace(line) == "" { + continue + } + lines = append(lines, line) + // Keep only last 2*limit lines in memory + if len(lines) > 2*limit { + lines = lines[limit:] + } + } + if err := scanner.Err(); err != nil && err != io.EOF { + return nil, err + } + + // Take last `limit` lines + start := 0 + if len(lines) > limit { + start = len(lines) - limit + } + + alarms := make([]map[string]any, 0, limit) + for i := start; i < len(lines); i++ { + var alarm map[string]any + if err := json.Unmarshal([]byte(lines[i]), &alarm); err != nil { + continue + } + alarms = append(alarms, alarm) + } + return alarms, nil +} + +func randomHex(n int) string { + b := make([]byte, n) + rand.Read(b) + return hex.EncodeToString(b)[:n] +} + +func isLocalhost(r *http.Request) bool { + ip := remoteIP(r) + return ip == "127.0.0.1" || ip == "::1" || ip == "localhost" +} diff --git a/agent/internal/httpapi/capabilities.go b/agent/internal/httpapi/capabilities.go new file mode 100644 index 0000000..2601ad5 --- /dev/null +++ b/agent/internal/httpapi/capabilities.go @@ -0,0 +1,92 @@ +package httpapi + +import ( + "net/http" +) + +// Capability represents a detection capability the device supports. +type Capability struct { + Key string `json:"key"` // "face", "shoe", "intrusion", ... + Label string `json:"label"` // human-readable name + Available bool `json:"available"` // whether the device has the required plugins + Description string `json:"description"` // brief description +} + +// handleCapabilities returns the list of detection capabilities supported by this device, +// derived from the available media-server node types (plugins). +func (s *Server) handleCapabilities(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + errorJSON(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + if !s.authorize(r, false) { + errorJSON(w, http.StatusUnauthorized, "unauthorized") + return + } + + // Query media-server for installed node types. + installed := s.installedNodeTypes() + + caps := []Capability{ + { + Key: "face", + Label: "人脸识别", + Available: hasAll(installed, "ai_face_recog"), + Description: "人脸检测与识别", + }, + { + Key: "shoe", + Label: "劳保鞋检测", + Available: hasAll(installed, "ai_shoe_det"), + Description: "工鞋/劳保鞋检测", + }, + { + Key: "helmet", + Label: "安全帽检测", + Available: hasAll(installed, "ai_yolo"), + Description: "安全帽佩戴检测（需安全帽模型）", + }, + { + Key: "intrusion", + Label: "区域入侵", + Available: hasAll(installed, "region_event"), + Description: "区域入侵/越线检测", + }, + { + Key: "action", + Label: "行为识别", + Available: hasAll(installed, "action_recog"), + Description: "摔倒、打架等行为识别", + }, + { + Key: "tracking", + Label: "人员跟踪", + Available: hasAll(installed, "tracker"), + Description: "多目标跟踪", + }, + } + + writeJSON(w, http.StatusOK, map[string]any{ + "capabilities": caps, + }) +} + +// installedNodeTypes returns all node types known to this agent build. +// The media-server binary carries the same plugin set, so all catalogued +// types are considered available. +func (s *Server) installedNodeTypes() map[string]bool { + set := map[string]bool{} + for _, nt := range graphNodeTypesCatalog() { + set[nt.Type] = true + } + return set +} + +func hasAll(installed map[string]bool, keys ...string) bool { + for _, k := range keys { + if !installed[k] { + return false + } + } + return true +} diff --git a/agent/internal/httpapi/extras.go b/agent/internal/httpapi/extras.go index 0072efe..fd5900c 100644 --- a/agent/internal/httpapi/extras.go +++ b/agent/internal/httpapi/extras.go @@ -115,6 +115,16 @@ func (s *Server) handleMetrics(w http.ResponseWriter, r *http.Request) { } else { resp["disk_io"] = map[string]any{"error": err.Error()} } + if npu, err := metrics.ReadNPULoad(); err == nil { + resp["npu"] = npu + } else { + resp["npu"] = map[string]any{"error": err.Error()} + } + if temp, err := metrics.ReadTemperature(); err == nil { + resp["temperature"] = temp + } else { + resp["temperature"] = map[string]any{"error": err.Error()} + } media := map[string]any{"supported": s.proc != nil && s.proc.Enabled()} if s.proc == nil || !s.proc.Enabled() { diff --git a/agent/internal/httpapi/preview.go b/agent/internal/httpapi/preview.go new file mode 100644 index 0000000..adc12c3 --- /dev/null +++ b/agent/internal/httpapi/preview.go @@ -0,0 +1,58 @@ +package httpapi + +import ( + "encoding/json" + "net/http" + "strconv" +) + +type previewChannel struct { + Name string `json:"name"` + HlsURL string `json:"hls_url,omitempty"` +} + +// handlePreviewChannels returns available video preview channels. +// GET /v1/preview/channels +func (s *Server) handlePreviewChannels(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + errorJSON(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + if !s.authorize(r, false) { + errorJSON(w, http.StatusUnauthorized, "unauthorized") + return + } + + channels := s.discoverChannels(r) + if channels == nil { + channels = make([]previewChannel, 0) + } + writeJSON(w, http.StatusOK, map[string]any{"channels": channels}) +} + +func (s *Server) discoverChannels(r *http.Request) []previewChannel { + if s.ms == nil { + return nil + } + _, body, err := s.ms.GetGraphs(r.Context()) + if err != nil { + return nil + } + + var graphs []struct { + Name string `json:"name"` + } + if err := json.Unmarshal(body, &graphs); err != nil { + return nil + } + + baseURL := "http://" + s.hostname + ":" + strconv.Itoa(s.mediaPort) + var channels []previewChannel + for _, g := range graphs { + channels = append(channels, previewChannel{ + Name: g.Name, + HlsURL: baseURL + "/hls/" + g.Name + "/index.m3u8", + }) + } + return channels +} diff --git a/agent/internal/httpapi/server.go b/agent/internal/httpapi/server.go index 57b569e..c655cfb 100644 --- a/agent/internal/httpapi/server.go +++ b/agent/internal/httpapi/server.go @@ -132,6 +132,9 @@ func New(agentCfg config.AgentConfig, baseDir string, ms *mediaserver.Client, st mux.HandleFunc("/v1/config/ui/apply", s.handleConfigUIApply) mux.HandleFunc("/v1/face-gallery", s.handleFaceGallery) mux.HandleFunc("/v1/face-gallery/reload", s.handleFaceGalleryReload) + mux.HandleFunc("/v1/alarms/report", s.handleAlarmReport) + mux.HandleFunc("/v1/alarms/recent", s.handleAlarmsRecent) + mux.HandleFunc("/v1/preview/channels", s.handlePreviewChannels) mux.HandleFunc("/v1/resources/status", s.handleResourcesStatus) mux.HandleFunc("/v1/resources/", s.handleResourceUpload) mux.HandleFunc("/v1/models", s.handleModelsList) @@ -148,6 +151,7 @@ func New(agentCfg config.AgentConfig, baseDir string, ms *mediaserver.Client, st mux.HandleFunc("/v1/media-server/binary/rollback", s.handleMediaBinaryRollback) mux.HandleFunc("/v1/agent/binary", s.handleAgentBinaryUpdate) mux.HandleFunc("/v1/graph-node-types", s.handleGraphNodeTypes) + mux.HandleFunc("/v1/capabilities", s.handleCapabilities) mux.HandleFunc("/v1/graphs", s.handleGraphs) mux.HandleFunc("/v1/graphs/", s.handleGraphDetail) mux.HandleFunc("/v1/logs/recent", s.handleLogsRecent) @@ -413,24 +417,54 @@ func (s *Server) writeConfigAndReload(ctx context.Context, body []byte, restoreB return fmt.Errorf("write config failed: %w", err) } - ctx, cancel := context.WithTimeout(ctx, 10*time.Second) - defer cancel() - if err := s.ms.Reload(ctx); err != nil { - rerr := err - if len(restoreBody) == 0 { - return fmt.Errorf("reload failed: %v", rerr) + // Signal media-server to reload (short timeout — just the HTTP call). + reloadCtx, reloadCancel := context.WithTimeout(ctx, 5*time.Second) + defer reloadCancel() + _ = s.ms.Reload(reloadCtx) // ignore immediate result; poll status instead + + // Wait for media-server to recover with the new config. + // The media-server may restart its pipeline, which can take 10-30 seconds. + deadline := time.After(60 * time.Second) + ticker := time.NewTicker(2 * time.Second) + defer ticker.Stop() + + for { + select { + case <-deadline: + // Timed out — roll back. + if len(restoreBody) == 0 { + return fmt.Errorf("media-server did not recover within 60s") + } + if werr := files.WriteFileAtomic(s.agentCfg.ConfigPath, append(restoreBody, '\n'), 0o644); werr != nil { + return fmt.Errorf("media-server recovery timeout; restore write failed: %v", werr) + } + _ = s.ms.Reload(context.Background()) + return fmt.Errorf("media-server did not recover within 60s; restored previous config") + case <-ticker.C: + if s.mediaServerHealthy() { + return nil + } + case <-ctx.Done(): + return ctx.Err() } - if werr := files.WriteFileAtomic(s.agentCfg.ConfigPath, append(restoreBody, '\n'), 0o644); werr != nil { - return fmt.Errorf("reload failed: %v; restore write failed: %v", rerr, werr) - } - restoreCtx, restoreCancel := context.WithTimeout(context.Background(), 10*time.Second) - defer restoreCancel() - if restoreErr := s.ms.Reload(restoreCtx); restoreErr != nil { - return fmt.Errorf("reload failed: %v; restore reload failed: %v", rerr, restoreErr) - } - return fmt.Errorf("reload failed: %v; restored previous config", rerr) } - return nil +} + +// mediaServerHealthy checks whether the media-server process is running +// and responding to API requests. +func (s *Server) mediaServerHealthy() bool { + if s.proc == nil || !s.proc.Enabled() { + return true // no process management — assume OK + } + st, err := s.proc.Status() + if err != nil || !st.Running { + return false + } + // Quick API health check (1s timeout). + ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) + defer cancel() + _, _, err = s.ms.GetGraphs(ctx) + return err == nil } func (s *Server) applyCandidateConfigBytes(ctx context.Context, body []byte) error { diff --git a/agent/internal/metrics/metrics.go b/agent/internal/metrics/metrics.go index a34f14d..6e99da0 100644 --- a/agent/internal/metrics/metrics.go +++ b/agent/internal/metrics/metrics.go @@ -261,3 +261,80 @@ func ParseLoadAvg(data string) (LoadAvg, error) { } return LoadAvg{One: one, Five: five, Fifteen: fifteen}, nil } + +type NPULoad struct { + UsagePct float64 `json:"usage_pct"` + Cores map[string]float64 `json:"cores,omitempty"` +} + +func ReadNPULoad() (*NPULoad, error) { + f, err := os.Open("/proc/rknpu/load") + if err != nil { + return nil, err + } + defer f.Close() + scanner := bufio.NewScanner(f) + cores := map[string]float64{} + n := 0 + for scanner.Scan() { + line := scanner.Text() + // Find each "CoreN: XX%" pattern + idx := 0 + for { + coreStart := strings.Index(line[idx:], "Core") + if coreStart < 0 { + break + } + coreStart += idx + colon := strings.Index(line[coreStart:], ":") + if colon < 0 { + break + } + colon += coreStart + coreName := strings.TrimSpace(line[coreStart:colon]) + // Find value after colon up to % or comma or end + valEnd := strings.Index(line[colon+1:], "%") + if valEnd < 0 { + break + } + valEnd += colon + 1 + valStr := strings.TrimSpace(line[colon+1 : valEnd]) + v, err := strconv.ParseFloat(valStr, 64) + if err == nil { + cores[coreName] = v + n++ + } + idx = valEnd + 1 + } + } + if n == 0 { + return nil, ErrNotSupported + } + var total float64 + for _, v := range cores { + total += v + } + avg := total / float64(n) + return &NPULoad{UsagePct: avg, Cores: cores}, nil +} + +type Temperature struct { + Zone string `json:"zone"` + Celsius float64 `json:"celsius"` +} + +func ReadTemperature() (*Temperature, error) { + raw, err := os.ReadFile("/sys/class/thermal/thermal_zone0/temp") + if err != nil { + return nil, err + } + zone, _ := os.ReadFile("/sys/class/thermal/thermal_zone0/type") + milli, err := strconv.ParseFloat(strings.TrimSpace(string(raw)), 64) + if err != nil { + return nil, err + } + return &Temperature{ + Zone: strings.TrimSpace(string(zone)), + Celsius: milli / 1000, + }, nil +} diff --git a/agent/rk3588-agent_linux_arm64 b/agent/rk3588-agent_linux_arm64 index 829152b..ea9d86d 100755 Binary files a/agent/rk3588-agent_linux_arm64 and b/agent/rk3588-agent_linux_arm64 differ diff --git a/configs/overlays/alarm_test_sensitive.json b/configs/overlays/alarm_test_sensitive.json new file mode 100644 index 0000000..cd96c19 --- /dev/null +++ b/configs/overlays/alarm_test_sensitive.json @@ -0,0 +1,47 @@ +{ + "description": "降低告警阈值用于测试：缩短冷却时间、降低最小持续时间和命中数。", + "instance_overrides": { + "*": { + "override": { + "nodes": { + "alarm_violation": { + "rules": [ + { + "name": "non_compliant_workshoe", + "class_ids": [2], + "min_score": 0.1, + "min_duration_ms": 100, + "min_hits": 1, + "hit_window_ms": 5000, + "cooldown_ms": 2000 + } + ], + "face_rules": [ + { + "name": "unknown_face", + "type": "unknown", + "cooldown_ms": 2000, + "min_hits": 1, + "hit_window_ms": 5000, + "min_face_area_ratio": 0.0001, + "min_face_aspect": 0.3, + "max_face_aspect": 3.0 + }, + { + "name": "known_person", + "type": "person", + "cooldown_ms": 2000, + "min_sim": 0.3, + "min_hits": 1, + "hit_window_ms": 5000, + "min_face_area_ratio": 0.0001, + "min_face_aspect": 0.3, + "max_face_aspect": 3.0 + } + ] + } + } + } + } + } +} diff --git a/configs/templates/std_workshop_face_recognition_shoe_alarm.json b/configs/templates/std_workshop_face_recognition_shoe_alarm.json index 6fa01d6..f6b4fa4 100644 --- a/configs/templates/std_workshop_face_recognition_shoe_alarm.json +++ b/configs/templates/std_workshop_face_recognition_shoe_alarm.json @@ -407,6 +407,12 @@ } ], "actions": { + "http": { + "enable": true, + "url": "http://127.0.0.1:9100/v1/alarms/report", + "timeout_ms": 2000, + "include_media_url": true + }, "log": { "enable": true, "level": "info", diff --git a/third_party/rknpu2/examples/rknn_yolov5_android_apk_demo/gradle/wrapper/gradle-wrapper.properties b/third_party/rknpu2/examples/rknn_yolov5_android_apk_demo/gradle/wrapper/gradle-wrapper.properties index 2e6e589..69b7a16 100644 --- a/third_party/rknpu2/examples/rknn_yolov5_android_apk_demo/gradle/wrapper/gradle-wrapper.properties +++ b/third_party/rknpu2/examples/rknn_yolov5_android_apk_demo/gradle/wrapper/gradle-wrapper.properties @@ -1,5 +1,6 @@ +#Mon May 11 15:07:02 CST 2026 distributionBase=GRADLE_USER_HOME distributionPath=wrapper/dists -distributionUrl=https\://services.gradle.org/distributions/gradle-7.3.3-bin.zip +distributionUrl=https\://services.gradle.org/distributions/gradle-8.9-bin.zip zipStoreBase=GRADLE_USER_HOME zipStorePath=wrapper/dists