宠物异常行为预警系统边缘计算与实时检测摘要本文深入讲解宠物异常行为预警系统的设计涵盖边缘计算架构、实时检测算法、多级告警机制、推送通知等完整技术方案。一、异常行为分类体系1.1 异常行为分级级别类型描述响应时间推送方式P0-紧急生命威胁抽搐、窒息、严重外伤即时电话短信APPP1-严重健康异常呕吐、腹泻、拒食24h5分钟短信APPP2-警告行为异常过度舔舐、异常叫声15分钟APP推送P3-提示生活异常饮水过多、活动量下降1小时APP通知1.2 异常行为特征库ANOMALY_FEATURES{vomiting:{level:P1,sensors:[camera,imu],description:呕吐行为,indicators:{body_motion:repeated_contraction,posture:head_low_neck_extended,duration_min:5,# 秒repetition:2}},seizure:{level:P0,sensors:[imu,heart_rate],description:抽搐/癫痫发作,indicators:{imu_pattern:high_freq_involuntary,heart_rate:elevated_irregular,duration_min:10,movement_intensity:5g}},excessive_licking:{level:P2,sensors:[camera,imu],description:过度舔舐可能皮肤问题,indicators:{body_part:same_area_repeated,frequency:10_times_per_hour,duration_total:30_min_per_day}},lethargy:{level:P1,sensors:[imu,activity],description:嗜睡/无精打采,indicators:{activity_level:30%_of_baseline,sleep_duration:18_hours,response_to_stimuli:delayed_or_none}},loss_of_appetite:{level:P1,sensors:[feeder,camera],description:食欲不振,indicators:{food_consumed:50%_of_normal,duration:24_hours,water_intake:normal_or_decreased}},pacing:{level:P2,sensors:[camera,imu],description:踱步/不安,indicators:{pattern:repetitive_path,duration:30_minutes,time_of_day:any}},hiding:{level:P2,sensors:[camera,location],description:躲藏可能生病或恐惧,indicators:{location:unusual_hiding_spot,duration:2_hours,social_avoidance:True}}}二、边缘计算架构2.1 边缘推理框架┌─────────────────────────────────────────────────┐ │ 边缘设备 (Jetson/RK3588) │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ 视频流 │ │ 传感器流 │ │ 音频流 │ │ │ │ 接收 │ │ 接收 │ │ 接收 │ │ │ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────────────────────────────────────┐ │ │ │ 特征提取层 │ │ │ │ 视觉特征 │ 运动特征 │ 声音特征 │ │ │ └─────────────────────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌─────────────────────────────────────────┐ │ │ │ 行为识别模型 │ │ │ │ YOLOv8 LSTM Transformer │ │ │ └─────────────────────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌─────────────────────────────────────────┐ │ │ │ 异常检测引擎 │ │ │ │ 规则引擎 异常检测模型 │ │ │ └─────────────────────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌─────────────────────────────────────────┐ │ │ │ 告警决策层 │ │ │ │ 告警聚合 │ 去重 │ 升级 │ 推送 │ │ │ └─────────────────────────────────────────┘ │ └─────────────────────────────────────────────────┘2.2 多线程流水线importthreadingimportqueuefromdataclassesimportdataclassfromtypingimportOptionalimporttimedataclassclassDetectionResult:timestamp:floatbehavior:strconfidence:floatlevel:strsource:strmetadata:dictclassEdgeAnomalyDetector:def__init__(self):self.video_queuequeue.Queue(maxsize30)self.sensor_queuequeue.Queue(maxsize100)self.audio_queuequeue.Queue(maxsize30)self.result_queuequeue.Queue(maxsize50)self.alert_queuequeue.Queue(maxsize20)# 加载模型self.yolo_modelself.load_yolo_model()self.lstm_modelself.load_lstm_model()self.anomaly_modelself.load_anomaly_model()# 状态管理self.behavior_buffer[]self.alert_history{}defstart(self):启动所有处理线程threads[threading.Thread(targetself.video_process_loop,daemonTrue),threading.Thread(targetself.sensor_process_loop,daemonTrue),threading.Thread(targetself.audio_process_loop,daemonTrue),threading.Thread(targetself.fusion_loop,daemonTrue),threading.Thread(targetself.alert_loop,daemonTrue),]fortinthreads:t.start()defvideo_process_loop(self):视频处理线程frame_buffer[]whileTrue:frameself.video_queue.get()# YOLO检测detectionsself.yolo_model.detect(frame)# 提取视觉特征featuresself.extract_visual_features(detections)frame_buffer.append(features)# 保持最近30帧iflen(frame_buffer)30:frame_buffer.pop(0)# LSTM时序分析iflen(frame_buffer)16:behaviorself.lstm_model.predict(frame_buffer[-16:])self.result_queue.put(DetectionResult(timestamptime.time(),behaviorbehavior[type],confidencebehavior[confidence],levelbehavior.get(level,P3),sourcevideo,metadata{bbox:detections}))defsensor_process_loop(self):传感器处理线程whileTrue:dataself.sensor_queue.get()# 提取运动特征activityself.classify_activity(data[imu])# 心率异常检测hr_anomalyself.detect_hr_anomaly(data[heart_rate])# 温度异常检测temp_anomalyself.detect_temp_anomaly(data[temperature])ifhr_anomalyortemp_anomaly:self.result_queue.put(DetectionResult(timestamptime.time(),behaviorhealth_anomaly,confidence0.8,levelP1,sourcesensor,metadata{heart_rate:data[heart_rate],temperature:data[temperature],activity:activity}))deffusion_loop(self):多模态融合线程results_buffer[]whileTrue:resultself.result_queue.get()results_buffer.append(result)# 保持最近100个结果iflen(results_buffer)100:results_buffer.pop(0)# 异常检测anomalyself.anomaly_model.detect(results_buffer)ifanomaly[is_anomaly]:# 告警聚合避免重复告警alert_keyf{anomaly[type]}_{anomaly.get(source,unknown)}ifself.should_alert(alert_key):self.alert_queue.put({type:anomaly[type],level:anomaly[level],confidence:anomaly[confidence],timestamp:time.time(),details:anomaly[details]})defalert_loop(self):告警处理线程whileTrue:alertself.alert_queue.get()# 根据级别选择推送方式ifalert[level]P0:self.send_phone_call(alert)self.send_sms(alert)self.send_app_push(alert)elifalert[level]P1:self.send_sms(alert)self.send_app_push(alert)elifalert[level]P2:self.send_app_push(alert)else:self.store_notification(alert)# 记录告警历史self.alert_history[alert[type]]time.time()defshould_alert(self,alert_key:str,cooldown:int300)-bool:告警去重同一类型告警冷却期last_alertself.alert_history.get(alert_key,0)returntime.time()-last_alertcooldown三、实时行为检测算法3.1 视觉行为检测importcv2importnumpyasnpfromultralyticsimportYOLOclassVisualBehaviorDetector:def__init__(self,model_path):self.modelYOLO(model_path)self.behavior_classes[sleeping,eating,drinking,playing,grooming,walking,sitting,standing,vomiting,seizure,pacing,hiding]defdetect_frame(self,frame):单帧检测resultsself.model(frame,verboseFalse)detections[]forrinresults:forboxinr.boxes:clsint(box.cls[0])conffloat(bo