diff --git a/.gitignore b/.gitignore index c7f8c0e..d85b05d 100644 --- a/.gitignore +++ b/.gitignore @@ -1,11 +1,21 @@ # 敏感配置和缓存 -config.py +/config.py token_cache.json sync.log # 下载的录像（每天几百 MB） recordings/ +# 本地视频分析产出（运行时生成） +analysis.db +analysis.db.bak.* +thumbnails/ +transcoded/ + +# 模型权重（体积大，首次运行自动下载） +analysis/models/ +yolo11n.pt + # Python __pycache__/ *.pyc @@ -20,3 +30,16 @@ __pycache__/ # 临时 *.tmp /tmp/ + +# 重复视频检测报告（运行时生成） +dedup_report.txt + +# Playwright MCP 运行日志/快照 +.playwright-mcp/ + +# 开发截图 +webapp-check.png +webapp-index.png + +# Frida +*.frida diff --git a/README.md b/README.md new file mode 100644 index 0000000..ca19f98 --- /dev/null +++ b/README.md @@ -0,0 +1,170 @@ +# ezviz-sync — 萤石摄像头录像同步与智能分析 + +从萤石（EZVIZ）摄像头同步 SD 卡报警录像到本地，对录像做 AI 分析（人物检测、摔倒检测、活动语义打标），并提供 Web 界面浏览和回放。 + +目标设备：萤石 CB60，序列号 `BH5455783`。 + +## 目录结构 + +``` +ezviz-sync/ +├── automate.py # 主力下载脚本：ADB 自动化模拟器内萤石 App 批量下载 +├── sync.py # [已废弃] 开放平台 API 下载方案，保留作备选 +├── probe.py # API 探测调试脚本 +├── analyze.py # 视频分析主入口（7 阶段流水线） +├── config.py # 萤石开放平台配置（密钥，已 gitignore） +├── config.example.py # 配置模板 +├── requirements.txt # 直接依赖 +├── requirements.lock # 锁定版本的完整依赖（pip freeze） +├── recordings/ # 下载的录像，按 设备号/日期/ 归档 +├── analysis.db # SQLite 元数据库（运行时生成） +├── thumbnails/ # 事件缩略图（运行时生成） +├── transcoded/ # Web 播放转码缓存（运行时生成） +├── analysis/ # 分析模块 +│ ├── db.py # SQLite 建表与 CRUD +│ ├── config.py # 分析参数配置（阈值、模型名、路径） +│ ├── video_scanner.py # 扫描录像文件入库 + MD5 查重 +│ ├── motion_filter.py # MOG2 背景建模检测活动窗口 +│ ├── person_tracker.py # YOLO11 + ByteTrack 人物跟踪 +│ ├── fall_detector.py # MediaPipe Pose 摔倒检测 +│ ├── activity_tagger.py # Ollama VLM 活动语义打标 +│ ├── thumbnail.py # 事件缩略图抽取 +│ ├── timestamp_reader.py # OSD 时间戳水印识别（模板匹配） +│ ├── backfill_start_time.py # 精确时间戳回填 +│ ├── dedup_finder.py # 重复视频检测（生成报告） +│ ├── dedup_cleanup.py # 重复视频清理（默认 dry-run） +│ ├── cleanup_orphans.py # 清理文件已删除的数据库孤儿记录 +│ ├── hashing.py # MD5 计算 +│ ├── ollama_client.py # Ollama VLM 调用封装 +│ ├── frida_spawn.py # Frida SSL pinning 绕过（抓包用） +│ ├── unpin.js # Frida 脱壳脚本 +│ ├── models/ # MediaPipe pose 模型权重 +│ └── templates/ # OSD 数字识别模板库 +├── webapp/ # Flask Web 界面 +│ ├── app.py # 路由：事件列表、统计、视频流、缩略图 +│ ├── transcode_cache.py # 按需转码缓存（HEVC→H264） +│ ├── templates/index.html +│ └── static/ # app.js, style.css +└── yolo11n.pt # YOLO11 目标检测权重（首次运行自动下载） +``` + +## 环境依赖 + +| 工具 | 用途 | 安装 | +|---|---|---| +| Python 3.12 | 运行环境 | `pyenv install 3.12` 或系统包管理器 | +| ffmpeg / ffprobe | 视频处理、抽帧、转码 | `brew install ffmpeg` | +| ImageMagick (`magick`) | 截图像素分析（automate.py 蓝勾/行检测） | `brew install imagemagick` | +| Android SDK + ADB | 模拟器控制 | `brew install --cask android-platform-tools` | +| Android 模拟器 | 运行萤石 App | Android Studio AVD 或第三方模拟器 | +| Ollama | 本地 VLM 推理（活动打标） | `brew install ollama && ollama pull qwen2.5vl:7b` | + +## 安装 + +```bash +# 1. 克隆项目 +git clone && cd ezviz-sync + +# 2. 创建虚拟环境 +python3.12 -m venv .venv +source .venv/bin/activate + +# 3. 安装依赖（用 lock 文件保证版本一致） +pip install -r requirements.lock + +# 4. 配置萤石开放平台 +cp config.example.py config.py +# 编辑 config.py 填入 AppKey / AppSecret / 设备序列号 / 验证码 + +# 5. 初始化数据库（首次运行 analyze.py 自动建表） +python analyze.py --only-stage scan +``` + +## 使用方法 + +### 1. 下载录像 + +```bash +# 下载昨天的录像（默认） +python automate.py + +# 下载最近 7 天的录像 +python automate.py 7 +``` + +下载流程：启动模拟器内萤石 App → 导航到指定日期 → 批量下载 → adb pull 到本地 `recordings/设备号/日期/`。 + +### 2. 分析录像 + +```bash +# 跑完整流水线（scan → backfill → motion → person → fall → activity → thumbnail） +python analyze.py + +# 只跑某个阶段 +python analyze.py --only-stage motion + +# 小批量测试 +python analyze.py --only-stage person --limit 10 + +# 强制重跑某个阶段（回退已完成的视频状态） +python analyze.py --only-stage activity --force +``` + +流水线状态机：`pending → motion_done → person_done → fall_done → activity_done`，每个阶段只处理上一阶段完成的视频，重复运行幂等。 + +### 3. 启动 Web 界面 + +```bash +python -m webapp.app +# 打开 http://127.0.0.1:5050 +``` + +功能：按事件类型/日期筛选、点击事件跳转播放对应时间点、转码缓存管理。 + +### 4. 维护工具 + +```bash +# 检测重复视频（只生成报告，不删除） +python -m analysis.dedup_finder + +# 清理重复视频（先预览，确认后加 --execute） +python -m analysis.dedup_cleanup +python -m analysis.dedup_cleanup --execute + +# 清理数据库中文件已不存在的孤儿记录 +python -m analysis.cleanup_orphans # dry-run +python -m analysis.cleanup_orphans --execute +``` + +## 分析流水线详解 + +| 阶段 | 输入状态 | 输出状态 | 方法 | 说明 | +|---|---|---|---|---| +| scan | — | pending/corrupt/duplicate | ffprobe + MD5 | 扫描 recordings/ 入库，按 MD5 标记重复 | +| backfill | pending | pending（不变） | 模板匹配 OCR | 读取画面 OSD 时间戳，回填精确 start_time | +| motion | pending | motion_done | MOG2 背景建模 | 检测有活动的时间窗口，空场景跳过后续重模型 | +| person | motion_done | person_done | YOLO11n + ByteTrack | 只在活动窗口内跑人物检测跟踪 | +| fall | person_done | fall_done | MediaPipe Pose | 只在有人的时间段跑姿态估计，规则引擎判摔倒 | +| activity | fall_done | activity_done | Ollama Qwen2.5-VL | 活动窗口抽帧，VLM 分类为 走动/进食/坐下/站立/离开/其他 | +| thumbnail | — | — | ffmpeg 抽帧 | 给每条事件生成缩略图 | + +## 关键设计决策 + +- **下载用 ADB 自动化而非开放平台 API**：开放平台 API 的 FLV 回放流把 24 小时录像拼接成一个文件，丢失了分段信息和精确时间戳；ADB 方案直接下载 App 内的分段报警录像，文件名含时间戳。 +- **分析流水线分层过滤**：先用轻量 MOG2 预筛活动窗口，再在窗口内跑 YOLO，有人的时间段才跑姿态估计和 VLM，避免对空场景浪费算力。 +- **摔倒检测用规则引擎而非 VLM**：安全信号需要可解释、可调阈值、宁可多报不漏报，MediaPipe 关键点 + "直立骤变横躺并持续"规则比模型黑盒更可靠。 +- **Web 播放按需转码**：源文件是 HEVC 编码且 mp4 容器 sample entry 标签缺失，浏览器无法直接播放；第一次播放时用 VideoToolbox 硬件转码为 H264 并缓存，之后复用。 + +## 常见问题 + +**Q: automate.py 报错 `magick: command not found`** +A: 没装 ImageMagick。`brew install imagemagick`。automate.py 的蓝勾检测和卡片行定位都依赖 `magick` 命令。 + +**Q: analyze.py 跑 activity 阶段很慢/卡住** +A: 检查 Ollama 是否在运行：`ollama ps`。模型 `qwen2.5vl:7b` 需要约 6GB 显存/内存。 + +**Q: Web 端视频播放不了** +A: 首次播放会触发转码（HEVC→H264），大文件可能需要几十秒。转码缓存在 `transcoded/`，可在 Web 界面清空。 + +**Q: 数据库里有很多 duplicate 记录** +A: automate.py 早期版本有重复下载问题（已修复），同一段录像被下载成多份。scan 阶段按 MD5 自动标记 duplicate 并跳过分析。可用 `dedup_cleanup.py` 清理多余文件。 diff --git a/analysis/activity_tagger.py b/analysis/activity_tagger.py new file mode 100644 index 0000000..96199e3 --- /dev/null +++ b/analysis/activity_tagger.py @@ -0,0 +1,121 @@ +"""在 motion_filter 找出的活动窗口内按固定间隔抽帧，调用本地 Ollama VLM +做限定词表活动分类，连续同标签的采样点合并成一条 activity 事件。 +""" +import json +from datetime import datetime, timedelta + +import cv2 + +from analysis.config import ACTIVITY_SAMPLE_INTERVAL_SEC, ACTIVITY_MERGE_GAP_SEC, PROJECT_ROOT +from analysis.db import get_all_videos, insert_event, update_video_status +from analysis.ollama_client import classify_frame + + +def _sample_labels(file_path, windows, fps): + """在活动窗口内按固定秒数间隔抽帧分类，返回 [(t, label, raw_text), ...]。""" + cap = cv2.VideoCapture(str(file_path)) + if not cap.isOpened(): + return [] + + results = [] + for start_sec, end_sec in windows: + t = start_sec + while t <= end_sec: + cap.set(cv2.CAP_PROP_POS_FRAMES, int(t * fps)) + ret, frame = cap.read() + if ret: + label, raw_text = classify_frame(frame) + results.append((t, label, raw_text)) + t += ACTIVITY_SAMPLE_INTERVAL_SEC + + cap.release() + return results + + +def _merge_labeled_samples(samples): + """连续同标签的采样点合并成一段 {start, end, label, raw_texts}。""" + segments = [] + for t, label, raw_text in samples: + if (segments and segments[-1]["label"] == label + and (t - segments[-1]["end"]) <= ACTIVITY_MERGE_GAP_SEC): + segments[-1]["end"] = t + segments[-1]["raw_texts"].append(raw_text) + else: + segments.append({"start": t, "end": t, "label": label, "raw_texts": [raw_text]}) + return segments + + +def process_video(video): + """返回 (segment_count, failed)。""" + try: + windows = json.loads(video["motion_windows_json"] or "[]") + except (json.JSONDecodeError, TypeError): + windows = [] + if not windows: + update_video_status(video["id"], "activity_done") + return 0, False + + file_path = PROJECT_ROOT / video["file_path"] + if not file_path.exists(): + update_video_status(video["id"], "activity_failed", error_message="文件不存在") + return 0, True + + try: + fps = video["fps"] or 15.0 + samples = _sample_labels(file_path, windows, fps) + except Exception as e: + update_video_status(video["id"], "activity_failed", error_message=str(e)) + return 0, True + + segments = _merge_labeled_samples(samples) + + start_time = None + if video["start_time"]: + try: + start_time = datetime.strptime(video["start_time"], "%Y-%m-%d %H:%M:%S") + except ValueError: + pass # 只精确到天的粗粒度值，无法算出绝对时间戳，timestamp_abs留空 + + for seg in segments: + timestamp_abs = (start_time + timedelta(seconds=seg["start"])).isoformat() if start_time else None + insert_event({ + "video_id": video["id"], + "event_type": "activity", + "label": seg["label"], + "start_offset_sec": seg["start"], + "end_offset_sec": seg["end"], + "timestamp_abs": timestamp_abs, + "confidence": None, + "track_id": None, + "thumbnail_path": None, + "extra_json": json.dumps({"raw_texts": seg["raw_texts"]}, ensure_ascii=False), + "created_at": datetime.now().isoformat(), + }) + + update_video_status(video["id"], "activity_done") + return len(segments), False + + +def process_all(limit=None): + videos = get_all_videos(status="fall_done") + if limit: + videos = videos[:limit] + + processed = 0 + total_segments = 0 + failed = 0 + for video in videos: + segments, is_failed = process_video(video) + total_segments += segments + if is_failed: + failed += 1 + processed += 1 + + return {"processed": processed, "total_segments": total_segments, "failed": failed} + + +if __name__ == "__main__": + print("===== 开始活动语义打标 =====") + result = process_all() + print(f"→ 处理 {result['processed']} 个视频，共标注 {result['total_segments']} 段活动") + print("===== 打标完成 =====") diff --git a/analysis/backfill_start_time.py b/analysis/backfill_start_time.py new file mode 100644 index 0000000..fac04ed --- /dev/null +++ b/analysis/backfill_start_time.py @@ -0,0 +1,48 @@ +"""把 OSD 时间戳水印识别出的精确到秒的时间，回填到 videos.start_time。 + +video_scanner 登记时 start_time 只精确到"天"（因为文件名时间戳是下载时刻而非 +录制时刻）。这里用 timestamp_reader 重新读取每个视频首帧的画面水印，识别成功 +则覆盖为精确到秒的时间；识别失败（分辨率未标定/置信度不足）则保留原有粗粒度 +值不动，不强行覆盖成错误数据。 +""" +from analysis.config import PROJECT_ROOT +from analysis.db import get_all_videos, update_start_time +from analysis.timestamp_reader import read_timestamp + + +# 720p 方案实测通过率仅约19%（模板库覆盖的时段太窄，泛化差），投入产出比太低， +# 暂不处理，720p视频保留 video_scanner 登记时"只精确到天"的粗粒度 start_time。 +BACKFILL_RESOLUTIONS = {(2560, 1440)} + + +def process_all(limit=None): + videos = [ + v for v in get_all_videos() + if v["status"] != "corrupt" and (v["width"], v["height"]) in BACKFILL_RESOLUTIONS + ] + if limit: + videos = videos[:limit] + + updated = 0 + fallback = 0 + for video in videos: + file_path = PROJECT_ROOT / video["file_path"] + precise_start_time = read_timestamp(file_path, capture_date=video["capture_date"]) + if precise_start_time is None: + fallback += 1 + continue + update_start_time(video["id"], precise_start_time) + updated += 1 + + return {"total": len(videos), "updated": updated, "fallback": fallback} + + +if __name__ == "__main__": + import sys + + limit = int(sys.argv[1]) if len(sys.argv) > 1 else None + print("===== 开始回填精确时间戳 =====") + result = process_all(limit=limit) + print(f"→ 共处理 {result['total']} 条，成功回填 {result['updated']} 条，" + f"识别失败保留粗粒度值 {result['fallback']} 条") + print("===== 回填完成 =====") diff --git a/analysis/cleanup_orphans.py b/analysis/cleanup_orphans.py new file mode 100644 index 0000000..e1318fc --- /dev/null +++ b/analysis/cleanup_orphans.py @@ -0,0 +1,90 @@ +"""清理数据库中文件已不存在的孤儿记录。 + +video_scanner 入库后，如果录像文件被手动删除或被 dedup_cleanup 清理， +数据库里的 videos/events 记录不会自动消失，导致： +- analyze.py 对不存在的文件跑分析，全部失败或产生空结果 +- webapp 点击播放 404 +- 统计数据虚高 + +本脚本扫描 videos 表，删除 file_path 不存在的记录及其关联 events， +并同步清理孤儿缩略图文件。执行前请确保已备份 analysis.db。 +""" +import os +import sqlite3 + +from analysis.config import PROJECT_ROOT, THUMBNAILS_DIR, DB_PATH + + +def cleanup_orphans(dry_run=True): + conn = sqlite3.connect(str(DB_PATH)) + conn.row_factory = sqlite3.Row + + videos = conn.execute("SELECT id, file_path, status FROM videos").fetchall() + to_delete = [] + for v in videos: + full_path = PROJECT_ROOT / v["file_path"] + if not full_path.exists(): + to_delete.append(v) + + print(f"videos 总数: {len(videos)}") + print(f"文件不存在的孤儿记录: {len(to_delete)}") + + # 按状态统计 + by_status = {} + for v in to_delete: + by_status[v["status"]] = by_status.get(v["status"], 0) + 1 + for status, cnt in sorted(by_status.items(), key=lambda x: -x[1]): + print(f" {status}: {cnt}") + + if dry_run: + print("\n[dry-run] 未执行任何删除。加 --execute 实际执行。") + conn.close() + return + + # 删除关联 events 和缩略图 + deleted_events = 0 + deleted_thumbs = 0 + for v in to_delete: + events = conn.execute( + "SELECT id, thumbnail_path FROM events WHERE video_id = ?", (v["id"],) + ).fetchall() + for e in events: + if e["thumbnail_path"]: + thumb_path = PROJECT_ROOT / e["thumbnail_path"] + if thumb_path.exists(): + thumb_path.unlink() + deleted_thumbs += 1 + conn.execute("DELETE FROM events WHERE video_id = ?", (v["id"],)) + deleted_events += len(events) + + # 删除 video 记录 + video_ids = [v["id"] for v in to_delete] + if video_ids: + placeholders = ",".join("?" * len(video_ids)) + conn.execute(f"DELETE FROM videos WHERE id IN ({placeholders})", video_ids) + + conn.commit() + + # 清理孤儿缩略图（events 已删但文件还在的） + event_thumbs = set( + r[0] for r in conn.execute("SELECT thumbnail_path FROM events WHERE thumbnail_path IS NOT NULL") + ) + orphan_thumbs = 0 + if THUMBNAILS_DIR.exists(): + for f in THUMBNAILS_DIR.glob("*.jpg"): + rel = str(f.relative_to(PROJECT_ROOT)) + if rel not in event_thumbs: + f.unlink() + orphan_thumbs += 1 + + conn.close() + + print(f"\n已删除 videos 记录: {len(to_delete)}") + print(f"已删除 events 记录: {deleted_events}") + print(f"已清理缩略图文件: {deleted_thumbs + orphan_thumbs}") + + +if __name__ == "__main__": + import sys + dry_run = "--execute" not in sys.argv + cleanup_orphans(dry_run=dry_run) diff --git a/analysis/config.py b/analysis/config.py new file mode 100644 index 0000000..4d9c525 --- /dev/null +++ b/analysis/config.py @@ -0,0 +1,65 @@ +"""本地视频分析模块的集中配置：路径、阈值、模型名、词表等所有可调参数。""" +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parent.parent + +# 录像来源目录：recordings/{设备号}/{YYYYMMDD}/*.mp4 +RECORDINGS_DIR = PROJECT_ROOT / "recordings" + +# 分析结果落地位置 +DB_PATH = PROJECT_ROOT / "analysis.db" +THUMBNAILS_DIR = PROJECT_ROOT / "thumbnails" + +# 网页播放按需转码缓存（源文件mp4容器HEVC轨道sample entry编码标签缺失， +# 浏览器无法直接播放，第一次点击时转码为标准H264并缓存在此，避免重复转码） +TRANSCODED_DIR = PROJECT_ROOT / "transcoded" + +# 活动语义打标的限定词表（供 VLM 分类 prompt 和 events.label 取值域共用） +ACTIVITY_LABELS = ["走动", "进食", "坐下", "站立", "离开", "其他"] + +# Ollama 本地模型 +OLLAMA_MODEL = "qwen2.5vl:7b" + +# 活动窗口检测（MOG2 背景建模）参数 +# 帧差法对"人物占画面小、动作幅度小"的场景（如老人在客厅整理东西）容易漏检， +# 改用 MOG2 自适应背景建模：背景随时间缓慢更新，前景掩码对小幅度/慢速运动更敏感。 +MOTION_RESIZE_WIDTH = 320 # 缩放到此宽度再做背景建模，加速计算 +MOTION_HISTORY = 120 # MOG2 背景模型参考的历史帧数 +MOTION_VAR_THRESHOLD = 16 # 像素值与背景模型的马氏距离阈值，越小越敏感（越容易判定为前景） +MOTION_WARMUP_FRAMES = 15 # 背景模型尚未收敛的前N个采样帧不计入判断，避免开场误判为"全员活动" +MOTION_SHORT_CLIP_SEC = 2.0 # 短于此时长的视频（约占全量一半，中位时长仅0.56秒）帧数不够MOG2预热， + # 触发即录的告警片段本身就是一次完整活动，直接整段视为活动窗口，不跑MOG2 +MOTION_PIXEL_RATIO = 0.015 # 一帧里前景像素（不含阴影）占比超过此值才算"有活动" +MOTION_MERGE_GAP_SEC = 1.5 # 两段活动间隔小于此秒数则合并为一段 +MOTION_MIN_DURATION_SEC = 0.5 # 短于此秒数的活动段视为噪声，丢弃 +MOTION_SAMPLE_STRIDE = 1 # 每隔几帧采样一次（1=逐帧，长视频可调大提速） + +# 人物出现次数统计（YOLO11n 检测 + ByteTrack 跟踪）参数 +# 只在 motion_filter 找出的活动窗口内跑，空场景直接跳过，省算力。 +PERSON_MODEL_NAME = "yolo11n.pt" # ultralytics 官方COCO预训练权重，首次运行自动下载 +PERSON_DEVICE = "mps" # M系列芯片走Metal加速 +PERSON_CONF_THRESHOLD = 0.4 # 检测置信度阈值，低于此值的框直接丢弃 +PERSON_SAMPLE_STRIDE = 2 # 每隔几帧检测一次（视频多为15fps，隔帧检测足够跟踪连续性） +PERSON_MIN_TRACK_DURATION_SEC = 0.3 # 持续时间短于此值的track视为误检噪声，丢弃 + +# 摔倒检测（MediaPipe Pose 关键点 + 规则引擎）参数 +# 只在 person_tracker 已确认"有人"的时间段内跑姿态估计，空场景不跑。 +# 安全信号宁可多报不漏报：判定摔倒 = 短时间内从"直立"骤变为"横躺"，且横躺状态持续一段时间 +# （躯干角度变水平这个单一条件区分不了"摔倒"和"弯腰/坐下"，必须叠加"骤变"+"持续"两个条件） +FALL_MODEL_PATH = PROJECT_ROOT / "analysis" / "models" / "pose_landmarker_lite.task" +FALL_SAMPLE_STRIDE = 2 # 每隔几帧跑一次姿态估计 +FALL_MIN_VISIBILITY = 0.5 # 关键点可见度低于此值不参与姿态计算，视为遮挡/未检出 +FALL_ASPECT_RATIO_THRESH = 1.3 # 人体关键点bbox 宽/高 超过此值才可能是"横躺" +FALL_TORSO_ANGLE_UPRIGHT = 35 # 躯干（肩中点→髋中点连线）与垂直方向夹角(度)，低于此值判定"直立" +FALL_TORSO_ANGLE_PRONE = 55 # 高于此值判定"趋于水平/横躺" +FALL_TRANSITION_MAX_SEC = 1.5 # "直立"变"横躺"必须在此时间内完成，排除缓慢弯腰/坐下/趟下 +FALL_SUSTAIN_MIN_SEC = 1.0 # "横躺"状态至少持续这么久才确认为摔倒，排除弯腰瞬间的误判 +FALL_GAP_TOLERANCE_SEC = 5.0 # 一次疑似摔倒事件期间，允许多久检测不到"横躺"样本仍视为 + # 事件在继续（真实摔倒后姿态/角度常导致关键点长时间漏检， + # 不代表已经爬起来）；超过此时长且中途也没等到"直立"信号， + # 才判定事件跟丢并结算 + +# 活动语义打标（本地 Ollama + Qwen2.5-VL 限定词表分类）参数 +# 只在 motion_filter 找出的活动窗口内抽帧，VLM推理比传统CV慢得多，不能逐帧跑。 +ACTIVITY_SAMPLE_INTERVAL_SEC = 2.0 # 每隔多少秒抽一帧送VLM分类 +ACTIVITY_MERGE_GAP_SEC = 3.0 # 连续同标签采样点之间允许的最大间隔，超过则视为新的一段活动 diff --git a/analysis/db.py b/analysis/db.py new file mode 100644 index 0000000..af7d2b3 --- /dev/null +++ b/analysis/db.py @@ -0,0 +1,222 @@ +"""analysis.db（SQLite）建表 + CRUD，唯一直接写 SQL 的地方。""" +import sqlite3 +from contextlib import contextmanager + +from analysis.config import DB_PATH + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS videos ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + device_serial TEXT NOT NULL, + capture_date TEXT NOT NULL, + file_path TEXT NOT NULL UNIQUE, + file_name TEXT NOT NULL, + file_size_bytes INTEGER, + duration_sec REAL, + fps REAL, + width INTEGER, + height INTEGER, + start_time TEXT, + status TEXT NOT NULL DEFAULT 'pending', + motion_windows_json TEXT, + error_message TEXT, + analyzed_at TEXT, + file_md5 TEXT, + duplicate_of INTEGER REFERENCES videos(id) +); + +CREATE TABLE IF NOT EXISTS events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + video_id INTEGER NOT NULL REFERENCES videos(id), + event_type TEXT NOT NULL, + label TEXT, + start_offset_sec REAL, + end_offset_sec REAL, + timestamp_abs TEXT, + confidence REAL, + track_id INTEGER, + thumbnail_path TEXT, + extra_json TEXT, + created_at TEXT +); + +CREATE INDEX IF NOT EXISTS idx_events_type_time ON events(event_type, timestamp_abs); +""" + + +@contextmanager +def get_connection(): + conn = sqlite3.connect(DB_PATH, timeout=30) + conn.row_factory = sqlite3.Row + # WAL 模式允许读写并发（analyze.py 写 + webapp 读不互锁）， + # busy_timeout 避免 "database is locked" 直接报错 + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=30000") + conn.execute("PRAGMA foreign_keys=ON") + try: + yield conn + conn.commit() + finally: + conn.close() + + +def init_db(): + with get_connection() as conn: + conn.executescript(SCHEMA) + _migrate_add_dedup_columns(conn) + + +def _migrate_add_dedup_columns(conn): + """老库没有 file_md5/duplicate_of 字段，CREATE TABLE IF NOT EXISTS 对已存在的表不生效， + 这里用 ALTER TABLE 补齐（新建库这两列已经在 SCHEMA 里，ALTER 会因列已存在而报错，捕获忽略）。 + """ + existing_columns = {row["name"] for row in conn.execute("PRAGMA table_info(videos)")} + if "file_md5" not in existing_columns: + conn.execute("ALTER TABLE videos ADD COLUMN file_md5 TEXT") + if "duplicate_of" not in existing_columns: + conn.execute("ALTER TABLE videos ADD COLUMN duplicate_of INTEGER REFERENCES videos(id)") + conn.execute("CREATE INDEX IF NOT EXISTS idx_videos_md5 ON videos(file_md5)") + + +def video_exists(file_path): + with get_connection() as conn: + row = conn.execute( + "SELECT 1 FROM videos WHERE file_path = ?", (str(file_path),) + ).fetchone() + return row is not None + + +def insert_video(video): + """插入一条 video 记录，file_path 已存在则忽略（保证重复扫描幂等）。返回该记录的 id。""" + with get_connection() as conn: + conn.execute( + """ + INSERT OR IGNORE INTO videos ( + device_serial, capture_date, file_path, file_name, + file_size_bytes, duration_sec, fps, width, height, + start_time, status, error_message, file_md5, duplicate_of + ) VALUES ( + :device_serial, :capture_date, :file_path, :file_name, + :file_size_bytes, :duration_sec, :fps, :width, :height, + :start_time, :status, :error_message, :file_md5, :duplicate_of + ) + """, + video, + ) + row = conn.execute( + "SELECT id FROM videos WHERE file_path = ?", (video["file_path"],) + ).fetchone() + return row["id"] + + +def get_video_id_by_md5(file_md5): + with get_connection() as conn: + row = conn.execute( + "SELECT id FROM videos WHERE file_md5 = ?", (file_md5,) + ).fetchone() + return row["id"] if row else None + + +def get_videos_missing_md5(): + with get_connection() as conn: + rows = conn.execute( + "SELECT id, file_path FROM videos WHERE file_md5 IS NULL" + ).fetchall() + return [dict(r) for r in rows] + + +def update_video_md5(video_id, file_md5): + with get_connection() as conn: + conn.execute( + "UPDATE videos SET file_md5 = ? WHERE id = ?", (file_md5, video_id) + ) + + +def count_videos(): + with get_connection() as conn: + row = conn.execute("SELECT COUNT(*) AS c FROM videos").fetchone() + return row["c"] + + +def get_all_videos(status=None): + with get_connection() as conn: + if status is None: + rows = conn.execute("SELECT * FROM videos ORDER BY capture_date, file_name").fetchall() + else: + rows = conn.execute( + "SELECT * FROM videos WHERE status = ? ORDER BY capture_date, file_name", (status,) + ).fetchall() + return [dict(r) for r in rows] + + +def update_video_status(video_id, status, error_message=None): + with get_connection() as conn: + conn.execute( + "UPDATE videos SET status = ?, error_message = ? WHERE id = ?", + (status, error_message, video_id), + ) + + +def update_start_time(video_id, start_time): + with get_connection() as conn: + conn.execute( + "UPDATE videos SET start_time = ? WHERE id = ?", + (start_time, video_id), + ) + + +def update_motion_windows(video_id, motion_windows_json, status, analyzed_at): + with get_connection() as conn: + conn.execute( + "UPDATE videos SET motion_windows_json = ?, status = ?, analyzed_at = ? WHERE id = ?", + (motion_windows_json, status, analyzed_at, video_id), + ) + + +def insert_event(event): + with get_connection() as conn: + conn.execute( + """ + INSERT INTO events ( + video_id, event_type, label, start_offset_sec, end_offset_sec, + timestamp_abs, confidence, track_id, thumbnail_path, extra_json, created_at + ) VALUES ( + :video_id, :event_type, :label, :start_offset_sec, :end_offset_sec, + :timestamp_abs, :confidence, :track_id, :thumbnail_path, :extra_json, :created_at + ) + """, + event, + ) + + +def get_video_by_id(video_id): + with get_connection() as conn: + row = conn.execute("SELECT * FROM videos WHERE id = ?", (video_id,)).fetchone() + return dict(row) if row else None + + +def update_event_thumbnail(event_id, thumbnail_path): + with get_connection() as conn: + conn.execute( + "UPDATE events SET thumbnail_path = ? WHERE id = ?", + (thumbnail_path, event_id), + ) + + +def update_event_extra(event_id, extra_json): + with get_connection() as conn: + conn.execute( + "UPDATE events SET extra_json = ? WHERE id = ?", + (extra_json, event_id), + ) + + +def get_events(event_type=None): + with get_connection() as conn: + if event_type is None: + rows = conn.execute("SELECT * FROM events ORDER BY timestamp_abs").fetchall() + else: + rows = conn.execute( + "SELECT * FROM events WHERE event_type = ? ORDER BY timestamp_abs", (event_type,) + ).fetchall() + return [dict(r) for r in rows] diff --git a/analysis/dedup_cleanup.py b/analysis/dedup_cleanup.py new file mode 100644 index 0000000..012dc30 --- /dev/null +++ b/analysis/dedup_cleanup.py @@ -0,0 +1,130 @@ +"""基于 dedup_finder 检测出的字节级重复文件，生成清理计划 / 执行实际清理。 + +处理原则：同一组 MD5 完全相同的文件里，只保留文件名（时间戳）最小的一份， +其余文件从磁盘删除；这些多余视频如果已经跑过 Phase3-5 分析，产生的 +person_presence/fall/activity 事件会让统计虚高，所以对应的 videos/events +数据库记录、缩略图、转码缓存也一并清理，避免留下孤儿数据。 + +默认只生成计划（不改动任何文件/数据库），确认无误后加 --execute 才真正执行， +且执行前还会要求在终端输入 yes 二次确认。 +""" +import argparse + +from analysis.config import PROJECT_ROOT, TRANSCODED_DIR +from analysis.db import get_connection +from analysis.dedup_finder import find_duplicates + + +def _video_id_for_path(conn, rel_path): + row = conn.execute( + "SELECT id FROM videos WHERE file_path = ?", (str(rel_path),) + ).fetchone() + return row["id"] if row else None + + +def build_plan(): + """返回删除计划列表，每项对应一个要删除的多余文件。""" + dup_groups = find_duplicates() + + plan = [] + with get_connection() as conn: + for paths in dup_groups.values(): + paths = sorted(paths, key=lambda p: p.name) + keep, remove = paths[0], paths[1:] + + for path in remove: + rel_path = path.relative_to(PROJECT_ROOT) + video_id = _video_id_for_path(conn, rel_path) + thumbnails = [] + event_count = 0 + if video_id is not None: + rows = conn.execute( + "SELECT thumbnail_path FROM events WHERE video_id = ?", (video_id,) + ).fetchall() + event_count = len(rows) + thumbnails = [r["thumbnail_path"] for r in rows if r["thumbnail_path"]] + + plan.append({ + "keep_path": str(keep.relative_to(PROJECT_ROOT)), + "remove_path": str(rel_path), + "video_id": video_id, + "event_count": event_count, + "thumbnails": thumbnails, + "size_bytes": path.stat().st_size, + }) + + return plan + + +def print_plan(plan): + total_bytes = sum(p["size_bytes"] for p in plan) + total_events = sum(p["event_count"] for p in plan) + orphan = [p for p in plan if p["video_id"] is None] + + print(f"计划删除 {len(plan)} 个重复文件，释放 {total_bytes / 1024 / 1024:.1f} MB") + print(f"将同步清理 {total_events} 条关联 events 记录") + if orphan: + print(f"其中 {len(orphan)} 个文件在 analysis.db 中还没有对应记录（尚未扫描过），只删文件不动数据库") + + preview = plan[:20] + for p in preview: + print(f" - {p['remove_path']} (保留 {p['keep_path']}, video_id={p['video_id']}, events={p['event_count']})") + if len(plan) > len(preview): + print(f" ...(其余 {len(plan) - len(preview)} 条略)") + + +def execute_plan(plan): + deleted_files = 0 + deleted_events = 0 + deleted_videos = 0 + + with get_connection() as conn: + for p in plan: + remove_path = PROJECT_ROOT / p["remove_path"] + if remove_path.exists(): + remove_path.unlink() + deleted_files += 1 + + for thumb in p["thumbnails"]: + thumb_path = PROJECT_ROOT / thumb + if thumb_path.exists(): + thumb_path.unlink() + + if p["video_id"] is not None: + cache_path = TRANSCODED_DIR / f"{p['video_id']}.mp4" + if cache_path.exists(): + cache_path.unlink() + + cur = conn.execute("DELETE FROM events WHERE video_id = ?", (p["video_id"],)) + deleted_events += cur.rowcount + conn.execute("DELETE FROM videos WHERE id = ?", (p["video_id"],)) + deleted_videos += 1 + + return { + "deleted_files": deleted_files, + "deleted_events": deleted_events, + "deleted_videos": deleted_videos, + } + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="重复视频清理：默认只打印计划，加 --execute 才真正执行") + parser.add_argument("--execute", action="store_true", help="实际执行删除") + args = parser.parse_args() + + plan = build_plan() + print_plan(plan) + + if not args.execute: + print("\n（当前只是计划预览，未做任何改动；确认无误后加 --execute 参数执行）") + else: + answer = input(f"\n确认删除以上 {len(plan)} 个文件及关联数据库记录？输入 yes 继续：") + if answer.strip().lower() == "yes": + result = execute_plan(plan) + print( + f"\n完成：删除文件 {result['deleted_files']} 个，" + f"删除 videos 记录 {result['deleted_videos']} 条，" + f"删除 events 记录 {result['deleted_events']} 条" + ) + else: + print("已取消") diff --git a/analysis/dedup_finder.py b/analysis/dedup_finder.py new file mode 100644 index 0000000..cc514fe --- /dev/null +++ b/analysis/dedup_finder.py @@ -0,0 +1,67 @@ +"""检测 recordings/ 目录下内容完全相同（字节级重复）的录像文件。 + +根因：automate.py 的 ADB 自动化下载在滚动/点击时序上有已知问题，会把同一张 +缩略图卡片的下载重复触发，导致同一段录像被存成多份文件名不同的拷贝。 + +先按文件大小分组（成本低，能筛掉绝大多数内容不同的文件），只对同大小的文件 +再计算 MD5 确认是否真的内容相同，避免对全部文件逐个算 MD5 的开销。 + +只生成报告，不删除任何文件、不改动数据库。 +""" +from collections import defaultdict + +from analysis.config import PROJECT_ROOT, RECORDINGS_DIR +from analysis.hashing import compute_md5 + +REPORT_PATH = PROJECT_ROOT / "dedup_report.txt" + + +def find_duplicates(): + """返回 {md5: [文件路径, ...]}，只包含重复数≥2的分组。""" + by_size = defaultdict(list) + for path in RECORDINGS_DIR.rglob("*.mp4"): + by_size[path.stat().st_size].append(path) + + by_md5 = defaultdict(list) + for size, paths in by_size.items(): + if len(paths) < 2: + continue + for path in paths: + by_md5[compute_md5(path)].append(path) + + return {md5: paths for md5, paths in by_md5.items() if len(paths) > 1} + + +def build_report(): + dup_groups = find_duplicates() + + lines = [] + total_extra_files = 0 + total_wasted_bytes = 0 + for md5, paths in sorted(dup_groups.items(), key=lambda kv: -len(kv[1])): + paths = sorted(paths, key=lambda p: p.name) + size = paths[0].stat().st_size + extra = len(paths) - 1 + total_extra_files += extra + total_wasted_bytes += extra * size + + lines.append(f"[{md5}] {len(paths)} 份重复，单份 {size / 1024:.1f} KB") + for p in paths: + lines.append(f" {p.relative_to(PROJECT_ROOT)}") + + summary = ( + f"重复分组数：{len(dup_groups)}\n" + f"多余拷贝数（可释放的文件数）：{total_extra_files}\n" + f"可释放空间：{total_wasted_bytes / 1024 / 1024:.1f} MB\n" + ) + + return summary + "\n" + "\n".join(lines) + + +if __name__ == "__main__": + print("===== 开始检测重复视频 =====") + report = build_report() + REPORT_PATH.write_text(report, encoding="utf-8") + print(report.split("\n\n")[0]) + print(f"→ 详细清单已写入 {REPORT_PATH.relative_to(PROJECT_ROOT)}") + print("===== 检测完成 =====") diff --git a/analysis/fall_detector.py b/analysis/fall_detector.py new file mode 100644 index 0000000..056596e --- /dev/null +++ b/analysis/fall_detector.py @@ -0,0 +1,266 @@ +"""在 person_tracker 已确认"有人"的时间段内跑 MediaPipe Pose 关键点估计， +用确定性规则引擎判断是否摔倒，写入 events 表（event_type='fall'）。 + +不用VLM做摔倒检测：这是安全信号，需要可解释、可调阈值、宁可多报不漏报的确定性 +规则，而不是模型的"感觉"。规则：短时间内从"直立"骤变为"横躺"，且"横躺"状态 +持续一段时间——单独判断"是否躺平"分不清摔倒和弯腰/坐下，必须叠加"骤变"和 +"持续"两个时间条件。 +""" +import json +import math +from datetime import datetime, timedelta + +import cv2 +import mediapipe as mp +import numpy as np +from mediapipe.tasks import python as mp_python +from mediapipe.tasks.python import vision + +from analysis.config import ( + FALL_MODEL_PATH, + FALL_SAMPLE_STRIDE, + FALL_MIN_VISIBILITY, + FALL_ASPECT_RATIO_THRESH, + FALL_TORSO_ANGLE_UPRIGHT, + FALL_TORSO_ANGLE_PRONE, + FALL_TRANSITION_MAX_SEC, + FALL_SUSTAIN_MIN_SEC, + FALL_GAP_TOLERANCE_SEC, + PROJECT_ROOT, +) +from analysis.db import get_all_videos, get_events, insert_event, update_video_status + +LEFT_SHOULDER, RIGHT_SHOULDER = 11, 12 +LEFT_HIP, RIGHT_HIP = 23, 24 + + +def _make_landmarker(): + base_options = mp_python.BaseOptions(model_asset_path=str(FALL_MODEL_PATH)) + options = vision.PoseLandmarkerOptions( + base_options=base_options, + running_mode=vision.RunningMode.VIDEO, + num_poses=1, + ) + return vision.PoseLandmarker.create_from_options(options) + + +def _classify_pose(landmarks, width, height): + """返回 'upright' / 'prone' / 'unknown'（关键点不足或角度介于两者之间）。""" + visible = [lm for lm in landmarks if lm.visibility >= FALL_MIN_VISIBILITY] + if len(visible) < 4: + return "unknown" + + xs = [lm.x * width for lm in visible] + ys = [lm.y * height for lm in visible] + bbox_w = max(xs) - min(xs) + bbox_h = max(ys) - min(ys) + if bbox_h <= 0: + return "unknown" + aspect_ratio = bbox_w / bbox_h + + shoulders = [landmarks[i] for i in (LEFT_SHOULDER, RIGHT_SHOULDER) if landmarks[i].visibility >= FALL_MIN_VISIBILITY] + hips = [landmarks[i] for i in (LEFT_HIP, RIGHT_HIP) if landmarks[i].visibility >= FALL_MIN_VISIBILITY] + if not shoulders or not hips: + return "unknown" + + shoulder_mid = (sum(lm.x for lm in shoulders) / len(shoulders) * width, + sum(lm.y for lm in shoulders) / len(shoulders) * height) + hip_mid = (sum(lm.x for lm in hips) / len(hips) * width, + sum(lm.y for lm in hips) / len(hips) * height) + dx = abs(hip_mid[0] - shoulder_mid[0]) + dy = abs(hip_mid[1] - shoulder_mid[1]) + torso_angle = math.degrees(math.atan2(dx, dy + 1e-6)) # 0°=垂直直立，90°=水平躺平 + + if aspect_ratio >= FALL_ASPECT_RATIO_THRESH and torso_angle >= FALL_TORSO_ANGLE_PRONE: + return "prone" + if aspect_ratio < 1.0 and torso_angle <= FALL_TORSO_ANGLE_UPRIGHT: + return "upright" + return "unknown" + + +def _extract_pose_states(landmarker, cap, fps, width, height, start_sec, end_sec): + """在给定时间段内采样跑姿态估计，返回 [(t, state), ...]。""" + start_frame = int(start_sec * fps) + end_frame = int(end_sec * fps) + cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame) + + states = [] + frame_idx = start_frame + while frame_idx <= end_frame: + ret, frame = cap.read() + if not ret: + break + + if (frame_idx - start_frame) % FALL_SAMPLE_STRIDE == 0: + t = frame_idx / fps + rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) + mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=np.ascontiguousarray(rgb)) + result = landmarker.detect_for_video(mp_image, int(t * 1000)) + if result.pose_landmarks: + state = _classify_pose(result.pose_landmarks[0], width, height) + else: + state = "unknown" + states.append((t, state)) + + frame_idx += 1 + + return states + + +def _find_fall_windows(states): + """从时序姿态状态里找出"直立→骤变→横躺并持续"的片段，返回 [(start_sec, end_sec), ...]。 + + 用"疑似摔倒事件"（episode）建模：只有真正观察到"直立"样本才会结束当前事件的计时； + 事件进行期间即使长时间检测不到姿态（真实摔倒后人体角度/遮挡经常导致关键点大面积 + 漏检，不代表已经爬起来），也继续算同一次事件，直到确认恢复直立，或者长时间 + （超过 FALL_GAP_TOLERANCE_SEC）连"横躺"样本都等不到才判定事件已经结束/跟丢。 + """ + windows = [] + last_upright_t = None + episode_start = None + episode_last_prone = None + + def close_episode(): + nonlocal episode_start, episode_last_prone + if episode_start is not None and (episode_last_prone - episode_start) >= FALL_SUSTAIN_MIN_SEC: + windows.append((episode_start, episode_last_prone)) + episode_start = None + episode_last_prone = None + + for t, state in states: + if state == "upright": + close_episode() + last_upright_t = t + elif state == "prone": + if episode_start is None: + # 只有紧接在"直立"之后、且间隔够短，才算一次骤变 + if last_upright_t is not None and (t - last_upright_t) <= FALL_TRANSITION_MAX_SEC: + episode_start = t + episode_last_prone = t + # 否则忽略：不是从直立状态骤变而来（比如视频开头就是横躺，或者是缓慢躺下） + elif (t - episode_last_prone) <= FALL_GAP_TOLERANCE_SEC: + episode_last_prone = t + else: + # 距离上一次确认"横躺"太久，且中途也没等到"直立"信号，视为跟丢，结算当前事件 + close_episode() + # state == "unknown" 不改变状态机，只是跳过这一帧（容忍短暂遮挡/漏检） + + close_episode() + return windows + + +def _merge_windows(events): + """合并人物track的时间区间（有重叠或紧邻就合并成一段）。 + 同一个人有时会被 ByteTrack 拆成多个track_id（跟踪中途丢失又重新编号）， + 这些track的时间区间往往高度重叠——如果按track各自跑一遍姿态估计， + 同一次摔倒会被重复检测、重复插入，所以要先合并再处理，保证同一段画面只处理一次。 + """ + spans = sorted((e["start_offset_sec"], e["end_offset_sec"]) for e in events) + merged = [] + for start, end in spans: + if merged and start <= merged[-1][1]: + merged[-1] = (merged[-1][0], max(merged[-1][1], end)) + else: + merged.append((start, end)) + return merged + + +def process_video(video, person_events): + if not person_events: + update_video_status(video["id"], "fall_done") + return 0 + + cap = cv2.VideoCapture(str(PROJECT_ROOT / video["file_path"])) + if not cap.isOpened(): + update_video_status(video["id"], "fall_done") + return 0 + + fps = video["fps"] or 15.0 + width, height = video["width"], video["height"] + + start_time = None + if video["start_time"]: + try: + start_time = datetime.strptime(video["start_time"], "%Y-%m-%d %H:%M:%S") + except ValueError: + pass + + fall_count = 0 + # landmarker 实例按需复用：VIDEO 模式要求时间戳在同一实例内严格递增，只有当下一个 + # 事件的起始时间会比当前实例已处理到的时间更早（多人同时/交替出现导致事件时间重叠） + # 时才换新实例。之前"每个事件都新建一个实例"的写法在批量跑千级视频时会持续新建 + # GPU/Metal 上下文，资源迟迟未完全释放，最终在某个看似无关的帧读取操作上引发底层崩溃 + landmarker = _make_landmarker() + last_t = -1.0 + try: + # 先合并所有track的时间区间再处理：同一个人有时会被拆成多个重叠的track_id， + # 按原始track各自跑一遍会导致同一次摔倒被重复检测、重复插入 + for merged_start, merged_end in _merge_windows(person_events): + start_frame_t = int(merged_start * fps) / fps + if start_frame_t <= last_t: + landmarker.close() + landmarker = _make_landmarker() + + states = _extract_pose_states( + landmarker, cap, fps, width, height, + merged_start, merged_end, + ) + if states: + last_t = states[-1][0] + + for start_sec, end_sec in _find_fall_windows(states): + timestamp_abs = (start_time + timedelta(seconds=start_sec)).isoformat() if start_time else None + insert_event({ + "video_id": video["id"], + "event_type": "fall", + "label": None, + "start_offset_sec": start_sec, + "end_offset_sec": end_sec, + "timestamp_abs": timestamp_abs, + "confidence": None, + "track_id": None, + "thumbnail_path": None, + "extra_json": None, + "created_at": datetime.now().isoformat(), + }) + fall_count += 1 + except Exception as e: + # 兜底：单条视频处理异常不应中断整个批次，标记失败并继续下一条。 + # 异常本身可能就是底层解码/GPU上下文已经出问题，close()也可能再报错，不能让它盖过原始异常 + try: + landmarker.close() + except Exception: + pass + cap.release() + update_video_status(video["id"], "fall_failed", error_message=str(e)) + return fall_count + + landmarker.close() + cap.release() + update_video_status(video["id"], "fall_done") + return fall_count + + +def process_all(limit=None): + videos = get_all_videos(status="person_done") + if limit: + videos = videos[:limit] + + events_by_video = {} + for e in get_events("person_presence"): + events_by_video.setdefault(e["video_id"], []).append(e) + + processed = 0 + total_falls = 0 + for video in videos: + total_falls += process_video(video, events_by_video.get(video["id"], [])) + processed += 1 + + return {"processed": processed, "total_falls": total_falls} + + +if __name__ == "__main__": + print("===== 开始检测摔倒事件 =====") + result = process_all() + print(f"→ 处理 {result['processed']} 个视频，检测到 {result['total_falls']} 次疑似摔倒") + print("===== 检测完成 =====") diff --git a/analysis/frida_spawn.py b/analysis/frida_spawn.py new file mode 100644 index 0000000..e3d1c5b --- /dev/null +++ b/analysis/frida_spawn.py @@ -0,0 +1,16 @@ +#!/usr/bin/env python3 +"""spawn 萤石 app 并注入脱壳脚本，保持挂载。""" +import frida, sys, time + +dev = frida.get_usb_device(timeout=10) +pid = dev.spawn(["com.videogo"]) +print(f"spawned pid={pid}", flush=True) +session = dev.attach(pid) +with open("analysis/unpin.js") as f: + src = f.read() +script = session.create_script(src) +script.on("message", lambda m, d: print("[msg]", m.get("payload", m), flush=True)) +script.load() +dev.resume(pid) +print("resumed, 保持挂载 1800s，请在模拟器里手动操作", flush=True) +time.sleep(1800) diff --git a/analysis/hashing.py b/analysis/hashing.py new file mode 100644 index 0000000..9b63761 --- /dev/null +++ b/analysis/hashing.py @@ -0,0 +1,10 @@ +"""通用文件 MD5 计算：供入库查重（video_scanner）和重复检测报告（dedup_finder）共用。""" +import hashlib + + +def compute_md5(path, chunk_size=1024 * 1024): + h = hashlib.md5() + with open(path, "rb") as f: + while chunk := f.read(chunk_size): + h.update(chunk) + return h.hexdigest() diff --git a/analysis/motion_filter.py b/analysis/motion_filter.py new file mode 100644 index 0000000..aed6520 --- /dev/null +++ b/analysis/motion_filter.py @@ -0,0 +1,145 @@ +"""MOG2背景建模检测每个视频里"有活动"的时间窗口，写回 videos.motion_windows_json。 + +长视频不能逐帧跑重模型，所以先用轻量背景建模预筛出活动窗口，后续 person_tracker / +fall_detector 只在窗口内跑，空场景直接跳过。 + +之前用固定阈值帧差法（相邻两帧像素差），对"人物占画面小、动作幅度小"的场景 +（如老人在客厅整理东西，移动缓慢、遮挡画面比例低）容易漏检。改用 MOG2 +自适应背景建模：背景是随时间缓慢更新的统计模型而非"上一帧"，前景掩码对 +慢速/小幅度运动更敏感，且能自动适应光照的缓慢变化。 +""" +import json +from datetime import datetime + +import cv2 + +from analysis.config import ( + MOTION_RESIZE_WIDTH, + MOTION_HISTORY, + MOTION_VAR_THRESHOLD, + MOTION_WARMUP_FRAMES, + MOTION_SHORT_CLIP_SEC, + MOTION_PIXEL_RATIO, + MOTION_MERGE_GAP_SEC, + MOTION_MIN_DURATION_SEC, + MOTION_SAMPLE_STRIDE, + PROJECT_ROOT, +) +from analysis.db import get_all_videos, update_motion_windows + + +def detect_motion_windows(file_path, duration_sec=None): + """返回该视频里的活动时间窗口列表 [[start_sec, end_sec], ...]。 + + duration_sec 由调用方传入（video_scanner用ffprobe测得，比cv2的帧数估算更准）； + 不传时退化为用 cv2 的帧数/帧率估算，供独立调用/测试使用。 + """ + cap = cv2.VideoCapture(str(file_path)) + if not cap.isOpened(): + return [] + + fps = cap.get(cv2.CAP_PROP_FPS) or 15.0 + + if duration_sec is None: + frame_count = cap.get(cv2.CAP_PROP_FRAME_COUNT) + duration_sec = frame_count / fps if frame_count > 0 else None + + # 触发即录的极短告警片段，帧数不够 MOG2 预热收敛，直接整段视为活动窗口 + if duration_sec is not None and duration_sec < MOTION_SHORT_CLIP_SEC: + cap.release() + return [[0.0, duration_sec]] + + subtractor = cv2.createBackgroundSubtractorMOG2( + history=MOTION_HISTORY, varThreshold=MOTION_VAR_THRESHOLD, detectShadows=True + ) + motion_flags = [] + frame_idx = 0 + sample_idx = 0 + + while True: + ret, frame = cap.read() + if not ret: + break + if frame_idx % MOTION_SAMPLE_STRIDE != 0: + frame_idx += 1 + continue + + h, w = frame.shape[:2] + scale = MOTION_RESIZE_WIDTH / w + small = cv2.resize(frame, (MOTION_RESIZE_WIDTH, max(1, int(h * scale)))) + small = cv2.GaussianBlur(small, (5, 5), 0) + + fg_mask = subtractor.apply(small) + # MOG2 用 127 标记阴影、255 标记前景，阴影不算真实活动，只统计255 + foreground_ratio = (fg_mask == 255).sum() / fg_mask.size + + if sample_idx >= MOTION_WARMUP_FRAMES: + motion_flags.append((frame_idx / fps, foreground_ratio > MOTION_PIXEL_RATIO)) + + frame_idx += 1 + sample_idx += 1 + + cap.release() + return merge_motion_windows(motion_flags) + + +def merge_motion_windows(motion_flags): + windows = [] + window_start = None + window_last_motion = None + + for t, is_motion in motion_flags: + if is_motion: + if window_start is None: + window_start = t + window_last_motion = t + elif window_start is not None and (t - window_last_motion) > MOTION_MERGE_GAP_SEC: + windows.append([window_start, window_last_motion]) + window_start = None + window_last_motion = None + + if window_start is not None: + windows.append([window_start, window_last_motion]) + + return [w for w in windows if (w[1] - w[0]) >= MOTION_MIN_DURATION_SEC] + + +def process_all(limit=None): + videos = get_all_videos(status="pending") + if limit: + videos = videos[:limit] + + processed = 0 + no_motion = 0 + failed = 0 + for video in videos: + try: + file_path = PROJECT_ROOT / video["file_path"] + if not file_path.exists(): + update_video_status(video["id"], "motion_failed", error_message="文件不存在") + failed += 1 + processed += 1 + continue + windows = detect_motion_windows(file_path, duration_sec=video["duration_sec"]) + update_motion_windows( + video["id"], + json.dumps(windows), + status="motion_done", + analyzed_at=datetime.now().isoformat(), + ) + processed += 1 + if not windows: + no_motion += 1 + except Exception as e: + update_video_status(video["id"], "motion_failed", error_message=str(e)) + failed += 1 + processed += 1 + + return {"processed": processed, "no_motion": no_motion, "failed": failed} + + +if __name__ == "__main__": + print("===== 开始检测活动窗口 =====") + result = process_all() + print(f"→ 处理 {result['processed']} 个视频，其中 {result['no_motion']} 个未检测到活动") + print("===== 检测完成 =====") diff --git a/analysis/ollama_client.py b/analysis/ollama_client.py new file mode 100644 index 0000000..26ec343 --- /dev/null +++ b/analysis/ollama_client.py @@ -0,0 +1,86 @@ +"""封装本地 Ollama 视觉模型调用：给定一帧图片，从限定词表中分类出一个活动标签。 + +只做限定词表分类，不做开放式生成——分类结果直接落地为 events.label，取值域 +必须严格落在词表内。VLM 输出格式不总是稳定（可能带标点/解释性文字）， +所以要做防御性解析兜底，解析不出合法标签就归为"其他"，不让脏数据污染打标结果。 +""" +import time + +import cv2 +import ollama + +from analysis.config import ACTIVITY_LABELS, OLLAMA_MODEL + +_FALLBACK_LABEL = "其他" +_MAX_RETRIES = 3 +_RETRY_DELAY = 5.0 # 秒，Ollama 偶尔忙/模型加载中，退避重试 + + +def _build_prompt(): + labels = "、".join(ACTIVITY_LABELS) + return ( + "这是家中监控摄像头拍到的一帧画面。请判断画面里的人此刻最可能在做什么，" + f"只能从下面这些词里选一个作为回答，不要输出任何其他文字或标点：{labels}" + ) + + +def _parse_label(raw_text): + stripped = raw_text.strip() + if stripped in ACTIVITY_LABELS: + return stripped + for label in ACTIVITY_LABELS: + if label in raw_text: + return label + return _FALLBACK_LABEL + + +_MAX_FRAME_WIDTH = 640 # 摄像头原始分辨率最高2560x1440，直接送入会因视觉token过多超出上下文， + # 而"走动/进食/坐下"这类粗粒度分类不需要高分辨率细节，缩小后再送入 + + +def _resize_for_vlm(frame_bgr): + height, width = frame_bgr.shape[:2] + if width <= _MAX_FRAME_WIDTH: + return frame_bgr + scale = _MAX_FRAME_WIDTH / width + return cv2.resize(frame_bgr, (_MAX_FRAME_WIDTH, int(height * scale))) + + +def check_ollama_health(): + """检查 Ollama 服务是否可用、目标模型是否已拉取。返回 (ok, message)。""" + try: + models = ollama.list() + model_names = [m.get("name", "") for m in models.get("models", [])] + if not any(OLLAMA_MODEL in name for name in model_names): + return False, f"模型 {OLLAMA_MODEL} 未拉取，请先运行: ollama pull {OLLAMA_MODEL}" + return True, "ok" + except Exception as e: + return False, f"Ollama 服务不可用: {e}" + + +def classify_frame(frame_bgr): + """frame_bgr: OpenCV BGR格式的numpy帧。返回 (标签, 原始回复文本)。""" + ok, buf = cv2.imencode(".jpg", _resize_for_vlm(frame_bgr)) + if not ok: + return _FALLBACK_LABEL, "" + + last_error = None + for attempt in range(_MAX_RETRIES): + try: + response = ollama.chat( + model=OLLAMA_MODEL, + messages=[{ + "role": "user", + "content": _build_prompt(), + "images": [buf.tobytes()], + }], + options={"num_ctx": 2048}, # 单帧分类不需要8192上下文，省显存提速 + ) + raw_text = response["message"]["content"].strip() + return _parse_label(raw_text), raw_text + except Exception as e: + last_error = e + if attempt < _MAX_RETRIES - 1: + time.sleep(_RETRY_DELAY) + # 全部重试失败，归为"其他"而不是抛异常中断整个批次 + return _FALLBACK_LABEL, f"[ollama_error] {last_error}" diff --git a/analysis/person_tracker.py b/analysis/person_tracker.py new file mode 100644 index 0000000..53b4503 --- /dev/null +++ b/analysis/person_tracker.py @@ -0,0 +1,158 @@ +"""在 motion_filter 找出的活动窗口内跑 YOLO11n 检测 + ByteTrack 跟踪， +统计人物出现次数，写入 events 表（event_type='person_presence'）。 + +一个 track（同一个人从出现到消失的连续跟踪片段）对应一条 person_presence +事件。只在活动窗口内跑，空场景（motion_windows_json 为空）直接跳过，不跑模型。 +""" +import json +from datetime import datetime, timedelta + +import cv2 + +from analysis.config import ( + PERSON_MODEL_NAME, + PERSON_DEVICE, + PERSON_CONF_THRESHOLD, + PERSON_SAMPLE_STRIDE, + PERSON_MIN_TRACK_DURATION_SEC, + PROJECT_ROOT, +) +from analysis.db import get_all_videos, insert_event, update_video_status + +_model = None + + +def _get_model(): + """YOLO模型只在真正用到时才加载（避免只跑无活动视频时也触发权重下载/加载）。""" + global _model + if _model is None: + from ultralytics import YOLO + + _model = YOLO(PERSON_MODEL_NAME) + return _model + + +def _track_windows(file_path, windows, fps): + """在给定的活动窗口内跑检测+跟踪，返回 [(track_id, start_sec, end_sec, max_conf), ...]。""" + model = _get_model() + cap = cv2.VideoCapture(str(file_path)) + if not cap.isOpened(): + return [] + + spans = {} # (window_index, track_id) -> [min_t, max_t, max_conf] + + for window_index, (start_sec, end_sec) in enumerate(windows): + start_frame = int(start_sec * fps) + end_frame = int(end_sec * fps) + cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame) + + frame_idx = start_frame + first_call = True + while frame_idx <= end_frame: + ret, frame = cap.read() + if not ret: + break + + if (frame_idx - start_frame) % PERSON_SAMPLE_STRIDE == 0: + # persist=False 强制在每个窗口开头重置跟踪器状态，跟踪器内部的track_id计数器 + # 也会跟着从1重新开始——不同窗口间同一个数字的track_id并不代表同一个人， + # 所以下面的 spans key 必须带上 window_index，否则会把不同窗口的片段错误合并 + results = model.track( + frame, persist=not first_call, classes=[0], conf=PERSON_CONF_THRESHOLD, + tracker="bytetrack.yaml", device=PERSON_DEVICE, verbose=False, + ) + first_call = False + + boxes = results[0].boxes + if boxes is not None and boxes.id is not None: + t = frame_idx / fps + for track_id, conf in zip(boxes.id.tolist(), boxes.conf.tolist()): + key = (window_index, int(track_id)) + if key not in spans: + spans[key] = [t, t, conf] + else: + spans[key][1] = t + spans[key][2] = max(spans[key][2], conf) + + frame_idx += 1 + + cap.release() + return [ + (track_id, start, end, conf) + for (_window_index, track_id), (start, end, conf) in spans.items() + if (end - start) >= PERSON_MIN_TRACK_DURATION_SEC + ] + + +def process_video(video): + """返回 (track_count, failed)。failed=True 表示该视频处理异常已标记 person_failed。""" + try: + windows = json.loads(video["motion_windows_json"] or "[]") + except (json.JSONDecodeError, TypeError): + windows = [] + if not windows: + update_video_status(video["id"], "person_done") + return 0, False + + file_path = PROJECT_ROOT / video["file_path"] + if not file_path.exists(): + update_video_status(video["id"], "person_failed", error_message="文件不存在") + return 0, True + + try: + fps = video["fps"] or 15.0 + tracks = _track_windows(file_path, windows, fps) + except Exception as e: + update_video_status(video["id"], "person_failed", error_message=str(e)) + return 0, True + + start_time = None + if video["start_time"]: + try: + start_time = datetime.strptime(video["start_time"], "%Y-%m-%d %H:%M:%S") + except ValueError: + pass # 只精确到天的粗粒度值（"YYYY-MM-DD"），无法算出绝对时间戳，timestamp_abs留空 + + for track_id, start_sec, end_sec, conf in tracks: + timestamp_abs = (start_time + timedelta(seconds=start_sec)).isoformat() if start_time else None + insert_event({ + "video_id": video["id"], + "event_type": "person_presence", + "label": None, + "start_offset_sec": start_sec, + "end_offset_sec": end_sec, + "timestamp_abs": timestamp_abs, + "confidence": conf, + "track_id": track_id, + "thumbnail_path": None, + "extra_json": None, + "created_at": datetime.now().isoformat(), + }) + + update_video_status(video["id"], "person_done") + return len(tracks), False + + +def process_all(limit=None): + videos = get_all_videos(status="motion_done") + if limit: + videos = videos[:limit] + + processed = 0 + total_tracks = 0 + failed = 0 + for video in videos: + tracks, is_failed = process_video(video) + total_tracks += tracks + if is_failed: + failed += 1 + processed += 1 + + return {"processed": processed, "total_tracks": total_tracks, "failed": failed} + + +if __name__ == "__main__": + print("===== 开始统计人物出现次数 =====") + result = process_all() + print(f"→ 处理 {result['processed']} 个视频，共识别出 {result['total_tracks']} 段人物出现") + print("===== 统计完成 =====") diff --git a/analysis/templates/1440p/digit_templates.npz b/analysis/templates/1440p/digit_templates.npz new file mode 100644 index 0000000..920fe73 Binary files /dev/null and b/analysis/templates/1440p/digit_templates.npz differ diff --git a/analysis/templates/720p/digit_templates.npz b/analysis/templates/720p/digit_templates.npz new file mode 100644 index 0000000..584ecf3 Binary files /dev/null and b/analysis/templates/720p/digit_templates.npz differ diff --git a/analysis/thumbnail.py b/analysis/thumbnail.py new file mode 100644 index 0000000..970be60 --- /dev/null +++ b/analysis/thumbnail.py @@ -0,0 +1,103 @@ +"""用 ffmpeg 在每条 events 记录的 start_offset_sec 处抽一帧，生成缩略图。 + +供网页事件列表展示预览图，点击缩略图再跳转播放对应视频的完整时间点。 + +失败重试上限：单个事件连续失败超过 MAX_ATTEMPTS 次后，在 extra_json 里标记 +thumbnail_failed，后续运行直接跳过，避免对损坏文件无限重试。 +""" +import json +import subprocess + +from analysis.config import PROJECT_ROOT, THUMBNAILS_DIR +from analysis.db import get_events, get_video_by_id, update_event_thumbnail, update_event_extra + +MAX_ATTEMPTS = 3 + + +def _extract_frame(video_path, offset_sec, output_path): + # -ss 放在 -i 之后用精确seek：放在 -i 之前的快速seek在临近视频结尾处 + # 会因缺少解码所需的参考帧而失败（HEVC "Could not find ref with POC"）。 + # 录像多为几秒到几十秒的短片，精确seek的性能代价可忽略。 + result = subprocess.run( + [ + "ffmpeg", "-i", str(video_path), "-ss", str(offset_sec), + "-frames:v", "1", "-q:v", "3", "-y", str(output_path), + ], + capture_output=True, timeout=30, + ) + return result.returncode == 0 and output_path.exists() + + +def _get_attempts(event): + try: + extra = json.loads(event["extra_json"] or "{}") + return extra.get("thumbnail_attempts", 0) + except (json.JSONDecodeError, TypeError): + return 0 + + +def _increment_attempts(event): + try: + extra = json.loads(event["extra_json"] or "{}") + except (json.JSONDecodeError, TypeError): + extra = {} + extra["thumbnail_attempts"] = extra.get("thumbnail_attempts", 0) + 1 + if extra["thumbnail_attempts"] >= MAX_ATTEMPTS: + extra["thumbnail_failed"] = True + update_event_extra(event["id"], json.dumps(extra, ensure_ascii=False)) + + +def process_all(limit=None): + THUMBNAILS_DIR.mkdir(exist_ok=True) + events = [] + for e in get_events(): + if e["thumbnail_path"]: + continue + try: + extra = json.loads(e["extra_json"] or "{}") + if extra.get("thumbnail_failed"): + continue + except (json.JSONDecodeError, TypeError): + pass + events.append(e) + + if limit: + events = events[:limit] + + video_cache = {} + processed = 0 + failed = 0 + for event in events: + video_id = event["video_id"] + if video_id not in video_cache: + video_cache[video_id] = get_video_by_id(video_id) + video = video_cache[video_id] + if video is None: + _increment_attempts(event) + failed += 1 + continue + + video_path = PROJECT_ROOT / video["file_path"] + if not video_path.exists(): + _increment_attempts(event) + failed += 1 + continue + + output_path = THUMBNAILS_DIR / f"{event['id']}.jpg" + offset_sec = event["start_offset_sec"] or 0.0 + + if _extract_frame(video_path, offset_sec, output_path): + update_event_thumbnail(event["id"], str(output_path.relative_to(PROJECT_ROOT))) + processed += 1 + else: + _increment_attempts(event) + failed += 1 + + return {"processed": processed, "failed": failed} + + +if __name__ == "__main__": + print("===== 开始生成缩略图 =====") + result = process_all() + print(f"→ 生成 {result['processed']} 张缩略图，失败 {result['failed']} 条") + print("===== 生成完成 =====") diff --git a/analysis/timestamp_reader.py b/analysis/timestamp_reader.py new file mode 100644 index 0000000..c791720 --- /dev/null +++ b/analysis/timestamp_reader.py @@ -0,0 +1,158 @@ +"""从摄像头画面自带的OSD时间戳水印读取精确到秒的录制时间。 + +原理：画面固定位置烧录 "YYYY-MM-DD HH:MM:SS" 字样。由于点阵字体渲染时字符间 +不保证有空白列（会粘连），不用连通域/空白列分割单个字符，而是先定位"日期组" +"时间组"两个文本块的整体边界（这两组之间的空格间隙远大于组内字符间隙），再用 +已知的固定字符数（日期10个字符，时间8个字符）做等宽硬切割。 + +日期组、时间组分别独立做 Otsu 二值化（而非整行统一阈值）：两组背景光照条件 +可能不同，统一阈值会让偏暗的一组文字被过度清除。 + +数字识别用多模板 IOU（交并比）相似度匹配，而非 cv2.matchTemplate 的 +TM_CCOEFF_NORMED —— 二值图像中大量近空白/全黑的 cell 两两比较时，归一化相关 +系数会退化出虚假的高分。模板库按分辨率持久化在 analysis/templates/<分辨率>/ +digit_templates.npz，收集自多个已知真值样本中同一数字在所有出现位置的实例。 +""" +from datetime import datetime, timedelta + +import cv2 +import numpy as np + +from analysis.config import PROJECT_ROOT + +TEMPLATES_DIR = PROJECT_ROOT / "analysis" / "templates" + +PATTERN = ("digit digit digit digit dash digit digit dash digit digit " + "digit digit colon digit digit colon digit digit").split() + +# 每种分辨率的裁剪坐标 + 日期组/时间组边界（相对裁剪区域的局部坐标）+ cell 尺寸。 +# 边界均由列投影法（找字符间隙）在多个样本上标定校正，不是拍脑袋的估计值。 +RESOLUTION_PROFILES = { + (2560, 1440): { + "crop_y": (1372, 1402), + "crop_x": (1750, 2560), + "date_group": (108, 373), # 10字符: Y Y Y Y - M M - D D + "time_group": (407, 613), # 8字符: H H : M M : S S + "cell_size": (20, 30), # (width, height)，传给 cv2.resize + }, + (1280, 720): { + "crop_y": (682, 705), + "crop_x": (870, 1280), + "date_group": (60, 192), + "time_group": (208, 312), + "cell_size": (20, 30), + }, +} + +# 识别置信度低于此值时判定为不可信，调用方应回退到"只精确到天"的粗粒度时间 +MIN_CONFIDENCE = 0.5 + +_template_cache = {} + + +def _load_templates(profile_key): + """按分辨率加载模板库，进程内缓存一次。""" + if profile_key not in _template_cache: + npz_path = TEMPLATES_DIR / f"{profile_key[1]}p" / "digit_templates.npz" + data = np.load(npz_path) + _template_cache[profile_key] = {digit: data[digit] for digit in data.files} + return _template_cache[profile_key] + + +def _cells_from_group(gray, group, n, cell_size): + s, e = group + sub = gray[:, s:e] + _, thresh = cv2.threshold(sub, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU) + w = (e - s) / n + cells = [] + for i in range(n): + x0 = int(round(i * w)) + x1 = int(round((i + 1) * w)) + cells.append(cv2.resize(thresh[:, x0:x1], cell_size, interpolation=cv2.INTER_AREA)) + return cells + + +def _iou(a, b): + a = a > 0 + b = b > 0 + inter = np.logical_and(a, b).sum() + union = np.logical_or(a, b).sum() + return inter / union if union > 0 else 0.0 + + +def _match_digit(cell, templates): + best_ch, best_score = None, -1 + for digit, tmpls in templates.items(): + for tmpl in tmpls: + score = _iou(cell, tmpl) + if score > best_score: + best_score, best_ch = score, digit + return best_ch, best_score + + +def _decode(gray, profile, templates): + cells = (_cells_from_group(gray, profile["date_group"], 10, profile["cell_size"]) + + _cells_from_group(gray, profile["time_group"], 8, profile["cell_size"])) + + out = [] + scores = [] + for cell, kind in zip(cells, PATTERN): + if kind == "digit": + ch, score = _match_digit(cell, templates) + out.append(ch) + scores.append(score) + elif kind == "dash": + out.append("-") + elif kind == "colon": + out.append(":") + + s = "".join(out) + # out 顺序 = 日期组10个cell(已含2个'-') + 时间组8个cell(已含2个':') + formatted = f"{s[0:10]} {s[10:18]}" + return formatted, (min(scores) if scores else 0.0) + + +def read_timestamp(file_path, capture_date=None): + """读取视频首帧的OSD时间戳水印，返回 "YYYY-MM-DD HH:MM:SS" 字符串。 + + 分辨率不在已标定范围内、读帧失败、识别置信度过低、或识别结果未通过合法性 + 校验时返回 None，调用方应保留原有"只精确到天"的粗粒度 start_time 作为兜底。 + + 单纯依赖 IOU 置信度阈值并不可靠：抽样发现即使分数达标，仍会出现"分钟81" + 这类不合法数值、或年份被识别错一位数（如2026错认成2006）但恰好凑成一个 + 合法日期的情况。因此额外做两层校验： + 1. 用 datetime 解析，过滤掉月/日/时/分/秒超出合法范围的结果 + 2. 若调用方提供 capture_date（文件所在日期文件夹，格式YYYYMMDD），解码出的 + 日期必须与之相差在1天以内（凌晨录制可能跨天），否则判定为误识别 + """ + cap = cv2.VideoCapture(str(file_path)) + ok, frame = cap.read() + cap.release() + if not ok: + return None + + height, width = frame.shape[:2] + profile = RESOLUTION_PROFILES.get((width, height)) + if profile is None: + return None + + y0, y1 = profile["crop_y"] + x0, x1 = profile["crop_x"] + gray = cv2.cvtColor(frame[y0:y1, x0:x1], cv2.COLOR_BGR2GRAY) + + templates = _load_templates((width, height)) + formatted, min_score = _decode(gray, profile, templates) + if min_score < MIN_CONFIDENCE: + return None + + try: + parsed = datetime.strptime(formatted, "%Y-%m-%d %H:%M:%S") + except ValueError: + return None + + if capture_date is not None: + folder_date = datetime.strptime(capture_date, "%Y%m%d") + if abs((parsed.date() - folder_date.date())) > timedelta(days=1): + return None + + return formatted diff --git a/analysis/unpin.js b/analysis/unpin.js new file mode 100644 index 0000000..0553032 --- /dev/null +++ b/analysis/unpin.js @@ -0,0 +1,53 @@ +/* + * 通用 Android SSL Pinning 绕过脚本（Frida） + * 覆盖常见实现：系统 TrustManagerImpl、OkHttp3 CertificatePinner、 + * SSLContext 自定义 TrustManager、Conscrypt。 + * 用途：抓包萤石云 app (com.videogo) 的 HTTPS 私有接口。 + */ +setTimeout(function () { + Java.perform(function () { + console.log('[unpin] 开始注入'); + + // 1. 系统 TrustManagerImpl.verifyChain —— 直接放行 + try { + var TMI = Java.use('com.android.org.conscrypt.TrustManagerImpl'); + TMI.verifyChain.implementation = function (untrustedChain, trustAnchorChain, host, clientAuth, ocspData, tlsSctData) { + console.log('[unpin] TrustManagerImpl.verifyChain 放行 host=' + host); + return untrustedChain; + }; + } catch (e) { console.log('[unpin] TrustManagerImpl 跳过: ' + e); } + + // 2. OkHttp3 CertificatePinner.check —— 直接返回 + try { + var CP = Java.use('okhttp3.CertificatePinner'); + CP.check.overload('java.lang.String', 'java.util.List').implementation = function (a, b) { + console.log('[unpin] okhttp3 CertificatePinner.check 放行 ' + a); + return; + }; + } catch (e) { console.log('[unpin] okhttp3 CertificatePinner 跳过: ' + e); } + + // 3. 自定义 X509TrustManager —— 替换成全信任 + try { + var X509TM = Java.use('javax.net.ssl.X509TrustManager'); + var SSLContext = Java.use('javax.net.ssl.SSLContext'); + var TrustManager = Java.registerClass({ + name: 'com.ez.FakeTM', + implements: [X509TM], + methods: { + checkClientTrusted: function () {}, + checkServerTrusted: function () {}, + getAcceptedIssuers: function () { return []; } + } + }); + var TMs = [TrustManager.$new()]; + var initOverload = SSLContext.init.overload( + '[Ljavax.net.ssl.KeyManager;', '[Ljavax.net.ssl.TrustManager;', 'java.security.SecureRandom'); + initOverload.implementation = function (km, tm, sr) { + console.log('[unpin] SSLContext.init 替换 TrustManager'); + initOverload.call(this, km, TMs, sr); + }; + } catch (e) { console.log('[unpin] SSLContext 跳过: ' + e); } + + console.log('[unpin] 注入完成'); + }); +}, 0); diff --git a/analysis/video_scanner.py b/analysis/video_scanner.py new file mode 100644 index 0000000..1960757 --- /dev/null +++ b/analysis/video_scanner.py @@ -0,0 +1,167 @@ +"""扫描 recordings/{设备号}/{YYYYMMDD}/*.mp4，登记进 videos 表。 + +start_time 只精确到"天"：文件名时间戳实测证实记录的是下载/导出时刻而非真实录制 +时刻（容器是 MPEG-PS，没有 creation_time 元数据可交叉验证），因此放弃追求绝对 +精确的录制时间，start_time 直接取自文件所在日期文件夹（capture_date）。 + +automate.py 的 ADB 自动化下载在滚动/点击时序上有已知问题，会把同一段录像重复 +下载成多份文件名不同的拷贝（详见 dedup_finder.py）。这里在入库时按 MD5 查重： +新文件如果和已入库的某个视频内容完全相同，标记为 duplicate 状态，不再进入 +Phase2~5 的分析流程，避免继续产生虚高的事件统计。 +""" +import json +import subprocess +from datetime import datetime + +from analysis.config import RECORDINGS_DIR, PROJECT_ROOT +from analysis.db import ( + insert_video, init_db, count_videos, video_exists, get_video_id_by_md5, + get_videos_missing_md5, update_video_md5, +) +from analysis.hashing import compute_md5 + + +def backfill_md5(): + """给这个字段上线之前已经入库、还没有 file_md5 的老记录补算一次， + 否则新文件查重时比对不到旧文件，查重形同虚设。""" + for video in get_videos_missing_md5(): + file_path = PROJECT_ROOT / video["file_path"] + if not file_path.exists(): + continue # 文件已被删除（例如清理过的重复文件），跳过 + update_video_md5(video["id"], compute_md5(file_path)) + + +def probe_video(file_path): + """用 ffprobe 取时长/帧率/分辨率。文件损坏或探测失败时返回 None。""" + try: + out = subprocess.run( + ["ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", str(file_path)], + capture_output=True, text=True, timeout=30, + ) + data = json.loads(out.stdout) + except (subprocess.TimeoutExpired, json.JSONDecodeError, OSError): + return None + + fmt = data.get("format") + if not fmt or "duration" not in fmt: + return None + + video_stream = next((s for s in data.get("streams", []) if s.get("codec_type") == "video"), None) + if not video_stream: + return None + + fps = None + if video_stream.get("r_frame_rate"): + num, _, den = video_stream["r_frame_rate"].partition("/") + if num and den and int(den) != 0: + fps = int(num) / int(den) + + return { + "duration_sec": float(fmt["duration"]), + "fps": fps, + "width": video_stream.get("width"), + "height": video_stream.get("height"), + } + + +def capture_date_to_start_time(capture_date): + return f"{capture_date[0:4]}-{capture_date[4:6]}-{capture_date[6:8]}" + + +def scan_all(): + """扫描全部设备目录下的全部录像文件，登记入库。返回 {已处理, 新增损坏, 新增重复} 统计。""" + init_db() + backfill_md5() + processed = 0 + corrupt = 0 + duplicates = 0 + + if not RECORDINGS_DIR.exists(): + return {"processed": 0, "corrupt": 0, "duplicates": 0} + + for device_dir in sorted(RECORDINGS_DIR.iterdir()): + if not device_dir.is_dir(): + continue + device_serial = device_dir.name + + for date_dir in sorted(device_dir.iterdir()): + if not date_dir.is_dir(): + continue + capture_date = date_dir.name + + for file_path in sorted(date_dir.glob("*.mp4")): + rel_path = str(file_path.relative_to(PROJECT_ROOT)) + if video_exists(rel_path): + continue # 已登记过，跳过重复的探测/哈希计算 + + file_size_bytes = file_path.stat().st_size + file_md5 = compute_md5(file_path) + duplicate_of = get_video_id_by_md5(file_md5) + + if duplicate_of is not None: + # 与已入库视频内容完全相同，标记为 duplicate，不进入后续分析流程 + video = { + "device_serial": device_serial, + "capture_date": capture_date, + "file_path": rel_path, + "file_name": file_path.name, + "file_size_bytes": file_size_bytes, + "duration_sec": None, + "fps": None, + "width": None, + "height": None, + "start_time": capture_date_to_start_time(capture_date), + "status": "duplicate", + "error_message": f"与 video_id={duplicate_of} 内容完全相同（MD5一致），跳过分析", + "file_md5": file_md5, + "duplicate_of": duplicate_of, + } + insert_video(video) + duplicates += 1 + processed += 1 + continue + + probe = probe_video(file_path) + + if probe is None: + status = "corrupt" + error_message = "ffprobe 探测失败或文件损坏（可能是 0 字节空文件）" + duration_sec = fps = width = height = None + corrupt += 1 + else: + status = "pending" + error_message = None + duration_sec = probe["duration_sec"] + fps = probe["fps"] + width = probe["width"] + height = probe["height"] + + video = { + "device_serial": device_serial, + "capture_date": capture_date, + "file_path": rel_path, + "file_name": file_path.name, + "file_size_bytes": file_size_bytes, + "duration_sec": duration_sec, + "fps": fps, + "width": width, + "height": height, + "start_time": capture_date_to_start_time(capture_date), + "status": status, + "error_message": error_message, + "file_md5": file_md5, + "duplicate_of": None, + } + insert_video(video) + processed += 1 + + return {"processed": processed, "corrupt": corrupt, "duplicates": duplicates} + + +if __name__ == "__main__": + print("===== 开始扫描录像文件 =====") + result = scan_all() + print(f"→ 本次处理 {result['processed']} 个文件，其中损坏/空文件 {result['corrupt']} 个，" + f"重复文件 {result['duplicates']} 个（已跳过分析）") + print(f"→ 数据库当前共登记 {count_videos()} 条记录") + print("===== 扫描完成 =====") diff --git a/analyze.py b/analyze.py new file mode 100644 index 0000000..7c28193 --- /dev/null +++ b/analyze.py @@ -0,0 +1,72 @@ +"""本地视频分析主入口：串联 扫描→时间戳回填→活动窗口→人物统计→摔倒检测→活动打标→缩略图 全流程。 + +手动/事后触发，独立于 automate.py 的下载流程，不影响现有下载逻辑。 +流水线按 videos.status 状态机推进，重复运行是幂等的：每个阶段只处理"上一阶段 +已完成、本阶段还没跑过"的视频，天然支持"每天下载完后跑一次，只处理当天新增"。 +""" +import argparse + +from analysis import activity_tagger, backfill_start_time, fall_detector, motion_filter, person_tracker, thumbnail, video_scanner +from analysis.db import get_all_videos, update_video_status + +STAGES = ["scan", "backfill", "motion", "person", "fall", "activity", "thumbnail"] + +# 每个阶段完成后视频进入的状态，以及 --force 时应该回退到的"上一阶段完成"状态 +_STAGE_STATUS = { + "motion": ("pending", "motion_done"), + "person": ("motion_done", "person_done"), + "fall": ("person_done", "fall_done"), + "activity": ("fall_done", "activity_done"), +} + + +def _force_reset(stage): + """--force 时，把已经跑过该阶段的视频状态回退，让流水线重新处理一遍。""" + if stage not in _STAGE_STATUS: + return + reset_to, done_status = _STAGE_STATUS[stage] + for video in get_all_videos(status=done_status): + update_video_status(video["id"], reset_to) + + +def run_stage(stage, limit=None, force=False): + if force: + _force_reset(stage) + + if stage == "scan": + result = video_scanner.scan_all() + print(f"[scan] 处理 {result['processed']} 个文件，损坏 {result['corrupt']} 个，重复 {result['duplicates']} 个") + elif stage == "backfill": + result = backfill_start_time.process_all(limit=limit) + print(f"[backfill] 处理 {result['total']} 条，成功回填精确时间 {result['updated']} 条，识别失败保留粗粒度 {result['fallback']} 条") + elif stage == "motion": + result = motion_filter.process_all(limit=limit) + print(f"[motion] 处理 {result['processed']} 个视频，{result['no_motion']} 个无活动，失败 {result.get('failed', 0)} 个") + elif stage == "person": + result = person_tracker.process_all(limit=limit) + print(f"[person] 处理 {result['processed']} 个视频，共 {result['total_tracks']} 段人物出现，失败 {result.get('failed', 0)} 个") + elif stage == "fall": + result = fall_detector.process_all(limit=limit) + print(f"[fall] 处理 {result['processed']} 个视频，检测到 {result['total_falls']} 次疑似摔倒") + elif stage == "activity": + result = activity_tagger.process_all(limit=limit) + print(f"[activity] 处理 {result['processed']} 个视频，标注 {result['total_segments']} 段活动，失败 {result.get('failed', 0)} 个") + elif stage == "thumbnail": + result = thumbnail.process_all(limit=limit) + print(f"[thumbnail] 生成 {result['processed']} 张缩略图，失败 {result['failed']} 条") + + +def main(): + parser = argparse.ArgumentParser(description="本地视频分析主流程") + parser.add_argument("--only-stage", choices=STAGES, help="只跑指定阶段，不跑完整流程") + parser.add_argument("--force", action="store_true", help="强制重新跑该阶段（把已完成的视频状态回退）") + parser.add_argument("--limit", type=int, default=None, help="限制本次处理的视频数量，用于小批量测试") + args = parser.parse_args() + + stages = [args.only_stage] if args.only_stage else STAGES + for stage in stages: + run_stage(stage, limit=args.limit, force=args.force) + + +if __name__ == "__main__": + main() diff --git a/automate.py b/automate.py index bff0a6c..f3c309e 100644 --- a/automate.py +++ b/automate.py @@ -7,8 +7,6 @@ import subprocess, time, os, shutil, struct, zlib from datetime import datetime, timedelta from pathlib import Path -PYTHON = "/usr/local/bin/python3.12" - ADB = os.path.expanduser("~/Library/Android/sdk/platform-tools/adb") DEVICE = "emulator-5554" OUTPUT_DIR = Path("./recordings") @@ -435,8 +433,9 @@ def tap_and_check_date_changed(arrow_coord, timeout=12, interval=1.0): return False -def navigate_to_yesterday_batch(): - """导航到昨天的 SD 卡批量下载模式。 +def navigate_to_yesterday_batch(days_back=1): + """导航到"今天往前第 days_back 天"的 SD 卡批量下载模式。 + days_back=0 是今天，1 是昨天，2 是前天……（默认 1 保持原行为）。 关键节点用 tap_text 通过 UI dump 定位，避免坐标错位时误触首页底部 Tab（萤石/对话/商城/社区/我的，"商城"中心 x=540 跟录像 Tab 撞车）。 """ @@ -485,11 +484,12 @@ def navigate_to_yesterday_batch(): else: print(" ⚠ 连续 8 次翻页仍在变化，按当前位置继续") - print("→ 从今日退一天到昨天") - if not tap_and_check_date_changed(COORD["prev_day_arrow"]): - print(" ! 首次未检测到日期变化，重试一次") + print(f"→ 从今日往前退 {days_back} 天") + for step in range(days_back): if not tap_and_check_date_changed(COORD["prev_day_arrow"]): - raise RuntimeError("点击'上一天'箭头后日期未变化，无法切换到昨天") + print(f" ! 第 {step+1} 次点'上一天'未检测到日期变化，重试一次") + if not tap_and_check_date_changed(COORD["prev_day_arrow"]): + raise RuntimeError(f"点击'上一天'箭头后日期未变化，无法退到第 {days_back} 天") # 等日期切换后的录像列表稳定加载 time.sleep(3) @@ -777,15 +777,24 @@ def tap_all_thumbnails(baseline, max_rounds=300, poll_interval=4, max_wait_per_r def _median(vals): return sorted(vals)[len(vals) // 2] if vals else -1 - scrolled = not (pre_rows and post_rows - and abs(_median(pre_rows) - _median(post_rows)) < 40) + def _moved(a, b): + return bool(a and b and abs(_median(a) - _median(b)) >= 40) - if not scrolled: - # 到底了（滚不动）。此时当前屏可能还有被"每轮只处理前 3 行"限制漏下的 - # 底部未下载卡片（它们本该靠继续滚动移到前 3 行处理，但到底滚不动了）。 - # 清扫：对当前屏【所有】检出行判打勾，分批（每批≤9 不超并发）点未下载的、 + if not _moved(pre_rows, post_rows): + # 疑似到底，但一次没滚动也可能是偶发滚动手势失效 / 下载中大文件卡顿吸收了 + # 手势（实测 26、27 号就因单次误判、提前结束、漏下几百个）。再滚一次确认： + # 只有连续两次都没推进才判真到底。 + time.sleep(2) + adb("shell", "input", "swipe", "540", "1820", "540", "1414", "400") + time.sleep(1.5) + post2_rows = detect_card_rows(screenshot("/tmp/_postscroll2.png")) + if _moved(pre_rows, post2_rows): + print(f" 轮 {r+1}: 首次滚动疑似卡顿未生效，二次滚动已推进，继续") + continue + # 连续两次滚动行位置都未变，确认到底。此时当前屏可能还有被"每轮只处理前 + # 3 行"限制漏下的底部未下载卡片，清扫：对当前屏所有检出行分批点未下载的、 # 等打勾，直到当前屏没有未下载卡片，再结束循环。 - print(f" 轮 {r+1}: 滚动后行位置未变，已到达列表底部，清扫当前屏剩余未下载卡片") + print(f" 轮 {r+1}: 连续两次滚动都未推进，确认到达列表底部，清扫当前屏剩余") for sweep in range(6): # 最多 6 批，防意外死循环 ensure_not_anr() time.sleep(THUMB_LOAD_WAIT) @@ -860,11 +869,14 @@ def wait_for_downloads(baseline_files, timeout=600): return list(all_files - baseline_files) -def pull_files(remote_files): - """把下载好的文件 adb pull 到本地""" - yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d") - # 本地目录按录像日期（昨天）命名，方便归档 - local_dir = OUTPUT_DIR / "BH5455783" / yesterday +def pull_files(remote_files, date_str=None): + """把下载好的文件 adb pull 到本地。 + date_str 为录像日期（YYYYMMDD），归档到 recordings/BH5455783/{date_str}/。 + 不传则默认昨天（保持原行为）。""" + if date_str is None: + date_str = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d") + # 本地目录按录像日期命名，方便归档 + local_dir = OUTPUT_DIR / "BH5455783" / date_str local_dir.mkdir(parents=True, exist_ok=True) print(f"→ 拉取 {len(remote_files)} 个文件到 {local_dir}") @@ -922,5 +934,58 @@ def sync(): print("===== 同步完成 =====") +def clear_device_recordings(): + """清空模拟器上萤石下载的录像文件 + 相册里的录像副本，释放存储空间。 + 只删录像（*_BH* 下载文件、YS 相册 *.mp4），不碰 App 的日志/播放器资源等其它文件。 + 多天连续下载时必须每天清，否则录像（含相册双份）会很快撑满模拟器 10G 存储。""" + today = datetime.now().strftime("%Y%m%d") + remote_dir = f"/sdcard/Android/data/{PKG}/files/{today}" + adb("shell", f"find {remote_dir} -name '*_BH*' -type f -delete") + adb("shell", "find /sdcard/DCIM/YS -iname '*.mp4' -type f -delete") + + +def sync_recent_days(n_days=7): + """下载最近 n_days 天的录像（今天、昨天……往前 n_days-1 天），按录像日期归档。 + + 每天独立重启 App 导航（比"退出批量下载页再切日期"稳妥）。关键：每天下载【前】 + 先清空模拟器上的录像+相册，让这一天从零开始——这样：①模拟器不会被多天累积撑满 + （10G 存储，含相册双份，几天就满）；②下完后设备上就是这一天的全部录像，直接 + 全量拉取到 recordings/{录像日期}/，不用再靠前后差集区分。拉完再清一次即时释放。 + 某天若无录像（点下载弹不出菜单）则跳过。 + """ + print(f"===== 开始下载最近 {n_days} 天录像 =====") + for d in range(n_days): + rec_date = (datetime.now() - timedelta(days=d)).strftime("%Y%m%d") + print(f"\n===== [{d+1}/{n_days}] 录像日期 {rec_date}（今天往前第 {d} 天）=====") + try: + # 清空本地该天旧目录：重下的文件名是新的下载时刻，和旧文件名不同， + # 不清会产生"内容相同、文件名不同"的重复副本。SD 卡是真源，重下可恢复。 + day_dir = OUTPUT_DIR / "BH5455783" / rec_date + if day_dir.exists(): + shutil.rmtree(day_dir) + print(f" 已清空本地旧目录 {day_dir}（覆盖式重下）") + launch_app() + clear_device_recordings() # 下载前清空，保证有空间、且设备只剩这天的 + navigate_to_yesterday_batch(days_back=d) + tap_all_thumbnails(set()) + wait_for_downloads(set()) + day_files = get_all_remote_files() # 这一天下载的全部录像 + if day_files: + pull_files(list(day_files), date_str=rec_date) + else: + print(f"→ {rec_date} 没有下载到文件") + clear_device_recordings() # 拉完即时清空，释放空间给下一天 + except Exception as e: + print(f"⚠ {rec_date} 处理失败，跳过：{e}") + continue + print(f"\n===== 最近 {n_days} 天下载完成 =====") + + if __name__ == "__main__": - sync() + import sys + # 用法：python automate.py → 只下昨天（原行为） + # python automate.py 7 → 下最近 7 天 + if len(sys.argv) > 1: + sync_recent_days(int(sys.argv[1])) + else: + sync() diff --git a/requirements.lock b/requirements.lock new file mode 100644 index 0000000..09d230b --- /dev/null +++ b/requirements.lock @@ -0,0 +1,70 @@ +absl-py==2.5.0 +annotated-types==0.8.0 +anyio==4.14.2 +blinker==1.9.0 +certifi==2026.7.22 +cffi==2.1.0 +charset-normalizer==3.4.9 +click==8.4.2 +colorama==0.4.6 +contourpy==1.3.3 +cycler==0.12.1 +filelock==3.32.0 +Flask==3.1.3 +flatbuffers==25.12.19 +fonttools==4.63.0 +frida==16.6.6 +frida-tools==12.5.1 +fsspec==2026.6.0 +h11==0.16.0 +httpcore==1.0.9 +httpx==0.28.1 +idna==3.18 +itsdangerous==2.2.0 +Jinja2==3.1.6 +kiwisolver==1.5.0 +lap==0.5.13 +MarkupSafe==3.0.3 +matplotlib==3.11.1 +mediapipe==0.10.35 +mpmath==1.3.0 +networkx==3.6.1 +numpy==1.26.4 +nvidia-ml-py==13.610.43 +ollama==0.6.2 +opencv-contrib-python==4.11.0.86 +opencv-python==4.10.0.84 +packaging==26.2 +paho-mqtt==2.1.0 +pillow==12.3.0 +polars==1.43.0 +polars-runtime-32==1.43.0 +prompt_toolkit==3.0.52 +psutil==7.2.2 +pycparser==3.0 +pycryptodome==3.23.0 +pydantic==2.13.4 +pydantic_core==2.46.4 +pyezvizapi==1.0.5.0 +Pygments==2.20.0 +pyparsing==3.3.2 +pytesseract==0.3.13 +python-dateutil==2.9.0.post0 +PyYAML==6.0.3 +requests==2.34.2 +setuptools==83.0.0 +six==1.17.0 +sounddevice==0.5.5 +sympy==1.14.0 +torch==2.13.0 +torchvision==0.28.0 +tqdm==4.69.1 +typing-inspection==0.4.2 +typing_extensions==4.16.0 +ultralytics==8.4.105 +ultralytics-thop==2.0.20 +urllib3==2.7.0 +wcwidth==0.8.2 +websockets==13.1 +Werkzeug==3.1.8 +xmltodict==1.0.4 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..124385f --- /dev/null +++ b/requirements.txt @@ -0,0 +1,10 @@ +opencv-python==4.10.0.84 +numpy>=1.26,<2.0 +ultralytics>=8.3 +lap>=0.5.12 +mediapipe>=0.10.14 +ollama>=0.4 +pytesseract>=0.3 +Flask>=3.0 +Werkzeug>=3.0 +tqdm>=4.66 diff --git a/sync.py b/sync.py index d8c1e49..cb718a0 100644 --- a/sync.py +++ b/sync.py @@ -1,9 +1,12 @@ #!/usr/bin/env python3 """ -萤石CB60 SD卡录像每日同步脚本 -每天凌晨运行，下载前24小时内的所有SD卡录像到本地目录。 +[已废弃，保留作为备选] 萤石CB60 SD卡录像每日同步脚本（开放平台 API 方案） + +当前主力下载方案是 automate.py（ADB 自动化模拟器内萤石 App 批量下载）。 +本脚本走萤石开放平台 API（open.ys7.com），需要先在开放平台关闭设备码流加密， +且 FLV 回放流拼接的录像不含精确时间戳，因此被 ADB 方案取代。保留仅作备选参考。 """ -import json, os, re, subprocess, logging +import json, os, re, shutil, subprocess, logging from datetime import datetime from pathlib import Path import requests @@ -15,8 +18,8 @@ OUTPUT_DIR = Path(OUTPUT_DIR) TOKEN_CACHE = Path(TOKEN_CACHE) LOG_FILE = Path("./sync.log") -# NAS上部署时改为 "ffmpeg"（Docker镜像内） -FFMPEG = "/opt/homebrew/bin/ffmpeg" +# ffmpeg 从 PATH 查找；NAS/Docker 部署时确保 ffmpeg 在 PATH 里即可，无需改代码 +FFMPEG = shutil.which("ffmpeg") or "ffmpeg" BASE = "https://open.ys7.com/api/lapp" diff --git a/webapp/app.py b/webapp/app.py new file mode 100644 index 0000000..3ee14ba --- /dev/null +++ b/webapp/app.py @@ -0,0 +1,106 @@ +"""本地事件展示网页：纯只读查询 analysis.db + 播放原始录像文件， +不写入任何数据，可以在 analyze.py 跑到一半时随时启动查看已完成的部分。 +""" +from flask import Flask, jsonify, render_template, request, send_file + +from analysis.config import PROJECT_ROOT +from analysis.db import get_connection +from webapp import transcode_cache + +app = Flask(__name__) + +EVENT_TYPES = ["person_presence", "fall", "activity"] + + +@app.route("/") +def index(): + return render_template("index.html") + + +@app.route("/api/events") +def api_events(): + event_type = request.args.get("event_type") + capture_date = request.args.get("date") # YYYYMMDD + + query = """ + SELECT e.id, e.video_id, e.event_type, e.label, e.start_offset_sec, + e.end_offset_sec, e.timestamp_abs, e.confidence, e.thumbnail_path, + v.capture_date, v.file_name + FROM events e JOIN videos v ON v.id = e.video_id + WHERE 1=1 + """ + params = [] + if event_type in EVENT_TYPES: + query += " AND e.event_type = ?" + params.append(event_type) + if capture_date: + query += " AND v.capture_date = ?" + params.append(capture_date) + query += " ORDER BY e.timestamp_abs DESC, e.id DESC LIMIT 500" + + with get_connection() as conn: + rows = conn.execute(query, params).fetchall() + + events = [ + { + "id": r["id"], + "video_id": r["video_id"], + "event_type": r["event_type"], + "label": r["label"], + "start_offset_sec": r["start_offset_sec"], + "end_offset_sec": r["end_offset_sec"], + "timestamp_abs": r["timestamp_abs"], + "confidence": r["confidence"], + "capture_date": r["capture_date"], + "file_name": r["file_name"], + "thumbnail_url": f"/media/thumb/{r['id']}" if r["thumbnail_path"] else None, + } + for r in rows + ] + return jsonify(events) + + +@app.route("/api/stats") +def api_stats(): + with get_connection() as conn: + rows = conn.execute( + "SELECT event_type, COUNT(*) AS c FROM events GROUP BY event_type" + ).fetchall() + return jsonify({r["event_type"]: r["c"] for r in rows}) + + +@app.route("/media//stream") +def media_stream(video_id): + with get_connection() as conn: + row = conn.execute("SELECT file_path FROM videos WHERE id = ?", (video_id,)).fetchone() + if row is None: + return "视频不存在", 404 + + try: + play_path = transcode_cache.ensure_transcoded(video_id, PROJECT_ROOT / row["file_path"]) + except RuntimeError as e: + return f"转码失败：{e}", 500 + return send_file(play_path, conditional=True) + + +@app.route("/api/cache/stats") +def api_cache_stats(): + return jsonify(transcode_cache.cache_stats()) + + +@app.route("/api/cache/clear", methods=["POST"]) +def api_cache_clear(): + return jsonify(transcode_cache.clear_cache()) + + +@app.route("/media/thumb/") +def media_thumb(event_id): + with get_connection() as conn: + row = conn.execute("SELECT thumbnail_path FROM events WHERE id = ?", (event_id,)).fetchone() + if row is None or not row["thumbnail_path"]: + return "缩略图不存在", 404 + return send_file(PROJECT_ROOT / row["thumbnail_path"], conditional=True) + + +if __name__ == "__main__": + app.run(host="127.0.0.1", port=5050, debug=False) diff --git a/webapp/static/app.js b/webapp/static/app.js new file mode 100644 index 0000000..3823bc4 --- /dev/null +++ b/webapp/static/app.js @@ -0,0 +1,108 @@ +const eventTypeFilter = document.getElementById("event-type-filter"); +const dateFilter = document.getElementById("date-filter"); +const eventList = document.getElementById("event-list"); +const player = document.getElementById("player"); +const playerInfo = document.getElementById("player-info"); +const statsEl = document.getElementById("stats"); +const cacheInfoEl = document.getElementById("cache-info"); +const clearCacheBtn = document.getElementById("clear-cache-btn"); + +const EVENT_TYPE_LABELS = { + person_presence: "人物出现", + fall: "摔倒", + activity: "活动", +}; + +function formatEventLabel(event) { + const typeName = EVENT_TYPE_LABELS[event.event_type] || event.event_type; + return event.label ? `${typeName} · ${event.label}` : typeName; +} + +async function loadStats() { + const res = await fetch("/api/stats"); + const stats = await res.json(); + const parts = Object.entries(stats).map( + ([type, count]) => `${EVENT_TYPE_LABELS[type] || type}: ${count}` + ); + statsEl.textContent = parts.join(" | "); +} + +function formatBytes(bytes) { + if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)} KB`; + if (bytes < 1024 * 1024 * 1024) return `${(bytes / 1024 / 1024).toFixed(1)} MB`; + return `${(bytes / 1024 / 1024 / 1024).toFixed(2)} GB`; +} + +async function loadCacheStats() { + const res = await fetch("/api/cache/stats"); + const stats = await res.json(); + cacheInfoEl.textContent = `转码缓存：${stats.file_count} 个文件，共占用 ${formatBytes(stats.total_size_bytes)}`; +} + +clearCacheBtn.addEventListener("click", async () => { + if (!confirm("确认清空所有转码缓存吗？清空后再次播放这些视频需要重新转码。")) return; + const res = await fetch("/api/cache/clear", { method: "POST" }); + const result = await res.json(); + alert(`已清理 ${result.deleted} 个文件，释放 ${formatBytes(result.freed_bytes)} 空间`); + loadCacheStats(); +}); + +async function loadEvents() { + const params = new URLSearchParams(); + if (eventTypeFilter.value) params.set("event_type", eventTypeFilter.value); + if (dateFilter.value) params.set("date", dateFilter.value.replaceAll("-", "")); + + const res = await fetch(`/api/events?${params.toString()}`); + const events = await res.json(); + renderEventList(events); +} + +function renderEventList(events) { + eventList.innerHTML = ""; + for (const event of events) { + const item = document.createElement("div"); + item.className = "event-item"; + + const thumb = document.createElement("img"); + thumb.className = "event-thumb"; + thumb.alt = formatEventLabel(event); + if (event.thumbnail_url) thumb.src = event.thumbnail_url; + + const info = document.createElement("div"); + info.className = "event-info"; + info.innerHTML = ` +
${formatEventLabel(event)}
+
${event.timestamp_abs || event.capture_date}
+ `; + + item.appendChild(thumb); + item.appendChild(info); + item.addEventListener("click", () => playEvent(event)); + eventList.appendChild(item); + } +} + +function playEvent(event) { + const isSameVideo = player.dataset.videoId === String(event.video_id); + player.dataset.videoId = event.video_id; + + const seekAndPlay = () => { + player.currentTime = event.start_offset_sec || 0; + player.play(); + }; + + if (isSameVideo) { + seekAndPlay(); + } else { + player.src = `/media/${event.video_id}/stream`; + player.addEventListener("loadedmetadata", () => { seekAndPlay(); loadCacheStats(); }, { once: true }); + } + playerInfo.textContent = `${event.file_name} · ${formatEventLabel(event)}`; +} + +eventTypeFilter.addEventListener("change", loadEvents); +dateFilter.addEventListener("change", loadEvents); + +loadStats(); +loadCacheStats(); +loadEvents(); diff --git a/webapp/static/style.css b/webapp/static/style.css new file mode 100644 index 0000000..8874570 --- /dev/null +++ b/webapp/static/style.css @@ -0,0 +1,96 @@ +body { + font-family: -apple-system, "PingFang SC", sans-serif; + margin: 20px; + color: #222; +} + +.filters { + margin-bottom: 16px; + display: flex; + gap: 16px; + align-items: center; +} + +#stats { + color: #666; + font-size: 0.9em; +} + +.cache-panel { + margin-bottom: 16px; + display: flex; + gap: 12px; + align-items: center; + color: #666; + font-size: 0.9em; +} + +.cache-panel button { + cursor: pointer; +} + +.main { + display: flex; + gap: 20px; +} + +.player-panel { + flex: 1; +} + +#player { + width: 100%; + max-width: 640px; + background: #000; +} + +#player-info { + margin-top: 8px; + color: #555; + font-size: 0.9em; +} + +.event-list { + flex: 1; + max-height: 80vh; + overflow-y: auto; + display: flex; + flex-direction: column; + gap: 8px; +} + +.event-item { + display: flex; + gap: 10px; + cursor: pointer; + padding: 6px; + border-radius: 6px; +} + +.event-item:hover { + background: #f0f0f0; +} + +.event-thumb { + width: 120px; + height: 68px; + object-fit: cover; + background: #ddd; + border-radius: 4px; + flex-shrink: 0; +} + +.event-info { + display: flex; + flex-direction: column; + justify-content: center; +} + +.event-type { + font-weight: 600; +} + +.event-time { + color: #666; + font-size: 0.9em; +} diff --git a/webapp/templates/index.html b/webapp/templates/index.html new file mode 100644 index 0000000..3abc732 --- /dev/null +++ b/webapp/templates/index.html @@ -0,0 +1,41 @@ + + + + + 录像事件回放 + + + +

录像事件回放

+ +
+ + + +
+ +
+ + +
+ +
+
+ +
+
+
+
+ + + + diff --git a/webapp/transcode_cache.py b/webapp/transcode_cache.py new file mode 100644 index 0000000..0d6c4d2 --- /dev/null +++ b/webapp/transcode_cache.py @@ -0,0 +1,96 @@ +"""网页播放的按需转码缓存：源文件mp4容器HEVC轨道sample entry编码标签缺失， +浏览器