import json from collections import deque, Counter import os import sys from datetime import datetime # 将 LAS 根目录加入 sys.path sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) import core.alarm_plan_helper as helper from common.sys_comm import ( LOGDBG, LOGINFO, LOGWARN, LOGERR, EC, get_utc_time_ms, get_utc_time_s, get_bj_time_ms, get_bj_time_s, utc_to_bj_ms, bj_to_utc_ms, utc_to_bj_s, bj_to_utc_s ) from core.alarm_plan import EventAttr_SleepMonitoring class SleepReplayAnalyzer: def __init__(self, file_path, bed_rect_ltwh): self.file_path = file_path self.bed_rect_ltwh = bed_rect_ltwh # 窗口和状态 self.motion_buffer = deque() # 每条: (ts, motion_value) self.sleep_segments_raw_ = [] self.sleep_segments_ = [] self.last_target_point = None self.last_target_ts = None self.motion_stat_ = "leave" self.sleep_stat_ = "leave" # 判断目标点是否在床区域 def _is_point_in_bed(self, target_point): x, y, *_ = target_point l, t, w, h = self.bed_rect_ltwh return l <= x <= l + w and (t - h) <= y <= t # 判断目标点是否在床区域 def _is_point_in_bed_ex(self, target_point, expand=10): x, y, *_ = target_point l, t, w, h = self.bed_rect_ltwh l_expanded = l - expand r_expanded = l + w + expand top_expanded = t + expand bottom_expanded = t - h - expand return l_expanded <= x <= r_expanded and bottom_expanded <= y <= top_expanded # 根据窗口计算平滑运动状态 def _calc_motion_state_window(self, current_ts, threshold_peaceful=2, threshold_micro=4): try: # 清理超过60秒的数据 while self.motion_buffer and current_ts - self.motion_buffer[0][0] > 60: self.motion_buffer.popleft() # 平滑运动 if self.motion_buffer: smooth_motion = sum(v for _, v in self.motion_buffer) / len(self.motion_buffer) else: smooth_motion = 0 # 运动状态 if smooth_motion < threshold_peaceful: motion = "peaceful" elif smooth_motion < threshold_micro: motion = "micro" else: motion = "active" return motion, smooth_motion except Exception as e: LOGERR(f"_calc_motion_state_window error: error: {e}") # 🔹 新版本：根据时间窗口（10分钟）分析最近数据 def _analyze_recent_window(self, ts, smooth_motion, breath_rpm, breathe_stat): try: if not self.sleep_segments_raw_: return None # 保留最近10分钟（600秒）的数据 time_window = 600 # 秒 start_ts = ts - time_window recent_data = [d for d in self.sleep_segments_raw_ if d["ts"] >= start_ts] if not recent_data: return None stats = [d["sleep_stat"] for d in recent_data] counter = Counter(stats) total = len(stats) major_stat, count = counter.most_common(1)[0] # 若占比超过50%，用当前主要状态，否则保持上次状态 if count / total >= 0.5: final_sleep_stat = major_stat else: final_sleep_stat = self.sleep_stat_ sleep_segment = { "ts": ts, "sleep_stat": final_sleep_stat, "smooth_motion": round(smooth_motion, 3), "breath_rpm": breath_rpm, "breathe_stat": breathe_stat } return sleep_segment except Exception as e: LOGERR(f"_analyze_recent_window error: {e}") return None # 主分析逻辑 def analyze(self): if not os.path.exists(self.file_path): LOGERR(f"录像文件不存在: {self.file_path}") return last_analyze_ts = None # 上次分析时间 last_saved_minute = None with open(self.file_path, "r", encoding="utf-8") as f: lines = f.readlines() for line in lines: line = line.strip() if not line: continue try: entry = json.loads(line) ts = entry["ts"] / 1000.0 # 转秒 payload = entry["payload"] targets = payload.get("tracker_targets", []) breath_rpm = payload.get("health", {}).get("breath_rpm", 0.0) # --- 运动状态 --- motion = self.motion_stat_ if targets: if not self._is_point_in_bed(targets[0]): motion_value = 0 smooth_motion = 0 motion = "leave" else: motion_value = 0 if self.last_target_point is not None: x1, y1, z1, _ = self.last_target_point x2, y2, z2, _ = targets[0] motion_value = ((x2 - x1)**2 + (y2 - y1)**2 + (z2 - z1)**2)**0.5 self.motion_buffer.append((ts, motion_value)) motion, smooth_motion = self._calc_motion_state_window(ts) self.last_target_point = targets[0] self.last_target_ts = ts else: smooth_motion = 0 if self.last_target_point and self._is_point_in_bed(self.last_target_point): motion = "peaceful" else: motion = "leave" self.motion_buffer.append((ts, 0)) self.motion_stat_ = motion # --- 呼吸状态 --- if breath_rpm < 8: breathe_stat = "r0" elif breath_rpm < 12: breathe_stat = "r1" elif breath_rpm < 18: breathe_stat = "r2" elif breath_rpm < 25: breathe_stat = "r3" else: breathe_stat = "r4" i_motion = EventAttr_SleepMonitoring.motion_stat.index(self.motion_stat_) i_breath = EventAttr_SleepMonitoring.breathe_stat.index(breathe_stat) sleep_stat_raw = EventAttr_SleepMonitoring.sleep_stat[i_motion][i_breath] # --- 保存原始实时片段 --- self.sleep_segments_raw_.append({ "ts": ts, "sleep_stat": sleep_stat_raw, "smooth_motion": round(smooth_motion, 3), "breath_rpm": breath_rpm, "breathe_stat": breathe_stat }) # --- 每隔30秒分析一次 --- if (last_analyze_ts is None) or (ts - last_analyze_ts >= 30): analyzed_segment = self._analyze_recent_window(ts, smooth_motion, breath_rpm, breathe_stat) if analyzed_segment: self.sleep_segments_.append(analyzed_segment) self.sleep_stat_ = analyzed_segment["sleep_stat"] last_analyze_ts = ts # 更新时间戳 # --- 可选：按分钟对齐输出日志（非必要） dt = datetime.fromtimestamp(ts) current_minute = dt.replace(second=0, microsecond=0) if last_saved_minute != current_minute: LOGDBG(f"[{current_minute.strftime('%H:%M')}] 当前状态: {self.sleep_stat_}") last_saved_minute = current_minute except Exception as e: LOGERR(f"解析行失败: {line}, error: {e}") def export_report(self, out_file=None): if not out_file: out_file = os.path.splitext(self.file_path)[0] + "_sleep_report.json" try: with open(out_file, "w", encoding="utf-8") as f: json.dump(self.sleep_segments_, f, ensure_ascii=False, indent=2) LOGINFO(f"睡眠报告已生成: {out_file}") print(f"睡眠报告已生成: {out_file}") except Exception as e: LOGERR(f"保存睡眠报告失败: {e}") print(f"保存睡眠报告失败: {e}") # ---------------- 使用示例 ---------------- if __name__ == "__main__": start_ts = get_utc_time_s() if 1: dev_id = "94A9900B0B38" bed_rect = [0, 250, 200, 115] else: dev_id = "94A9900B0B80" bed_rect = [80, 130, 120, 210] file_path = f"demos/record/{dev_id}.raw.json" # file_path = f"demos/record/2025-10-13_21-03-13_94A9900B0B80.raw.json" analyzer = SleepReplayAnalyzer(file_path, bed_rect) analyzer.analyze() ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") analyzer.export_report(f"./demos/record/report_{dev_id}_{ts}.json") end_ts = get_utc_time_s() total_time = end_ts - start_ts print(f" 耗时：{total_time}秒") LOGINFO(f" 耗时：{total_time}秒")