Compare commits

...

16 Commits

Author SHA1 Message Date
yuming 548ab3b2f1 feat: 多趟循环下载直到无新增唯一录像,补齐单趟遗漏
- tap_all_thumbnails 返回本趟点击的未打勾卡片数
- 新增 tap_all_thumbnails_multipass:同一 App 会话内反复回顶重扫、
  只补漏点卡,用"设备唯一内容(md5)增量=0"判完整——对蓝勾约4.5%
  假阴性、死录像均免疫("零点击"判据因假阴性永远达不到,已弃用)
- sync_recent_days 接入多趟循环,替代单趟 tap+wait
- 配套:cell_has_card 增亮度判据识别灰色占位卡;到底重试加耐心(6次/7s)

实测 09-19:清理存储后单趟 120→281,合并归档达 285 完整;
多趟复验第 1 趟 21 次假阴性补点、新增唯一 0,正确判完整并 1 趟退出。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-09-23 18:00:52 +08:00
yuming 8010aab3a0 fix: 到底后补扫最终屏,修列表尾部最后一两行漏下
确认到底后那几次重试滚动的微小移动会把最底一两行完整露出来,
旧逻辑一 break 就结束、最终屏从未被清扫(实测 06:19~06:30 两行漏下)。
修复:确认到底后 break 前对最终屏再 clear_current_screen 补扫一次。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-09-21 12:31:01 +08:00
yuming 09246db013 fix: 批量下载可靠性修复(点击滚动解耦+动态行检测+边缘行/底部漏下+收尾崩溃)
- clear_current_screen 点击与滚动解耦:当前屏全部下完才前进,结构上不漏
- detect_card_rows 动态行检测 + navigate/screenshot/tap_until_activity 稳定性修复
- 滚动步长 1900→1000 缩为 1800→1150,加大重叠,修中段边缘行漏
- 到底判定 MIN_SCROLL_PX 250→80 + 连续3次强力重试,修底部一整段误判到底漏下
- wait_for_downloads 入口归一 set,修 set-list TypeError 收尾崩溃

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-09-20 21:50:24 +08:00
yuming 8a934c8813 fix: 每轮点3个+等待180秒,避免队列堆积
红点检测不准(失败任务红点持续存在),改为:
- 每轮只点3个(而不是6个),给队列留余量
- 等待时间从120秒增加到180秒
- 超时后直接滚动,靠下游MD5去重兜底
2026-09-18 22:55:12 +08:00
yuming dd4d5571e2 fix: 点击前检测下载队列红点,队列忙时等待消化
之前不检测队列状态,队列满了还继续点击新卡片,
新点击被拒绝永远不打勾,空等6分钟后滚动走导致漏下载。

现在每轮截图后先检测右上角红点:
- 有红点=队列忙,等待最多5分钟让队列消化
- 红点消失后再点击新卡片
2026-09-18 22:38:11 +08:00
yuming 68847167dd fix: 打勾计数改用动态行检测,修复固定步长漏检
之前用固定步长204采样y坐标,与实际行位置不匹配,
只检测到5个角标(实际16个),导致等待逻辑永远超时。
改为每次截图后重新 detect_card_rows 检测实际行位置。
2026-09-18 22:04:43 +08:00
yuming 3dd66529a2 fix: 增大滚动距离406px→1200px,避免误判到底 2026-09-18 21:55:48 +08:00
yuming 011eb51cfb fix: 打勾检测改用全局蓝色角标计数,不受列表滚动漂移影响
根因:等待期间列表轻微滚动(约50px),用点击时的旧行坐标检测打勾
位置偏移,导致明明已打勾却检测不到0/6,误判为下载失败。

修复:
- 新增 count_checked_badges():扫描整屏蓝色角标总数,不依赖行坐标
- 等待逻辑改为:点击前记基线→等待期间统计当前总数→
  增量>=点击数即完成
- 重试逻辑同步改用全局计数
2026-09-18 21:44:17 +08:00
yuming 317bd3020d fix: 修复等待逻辑,避免未完成就滚动导致漏下载
问题:旧逻辑等30秒就超时,重试点击一次后再等30秒就滚动。
大文件30秒下不完,队列满时重试点击无效,导致卡片漏下载。

修复:
- max_wait_per_round 从30秒增加到120秒
- 等待逻辑改为三轮:先等120s→再等120s→重试点击→再等120s
- 总最长等待6分钟,足够6个小文件全部下载完
- 队列满时不立即重试点击,先让正在下载的完成腾出位置
2026-09-18 21:22:44 +08:00
yuming 2e01d9e774 fix: 每轮点击数从9降到6,匹配萤石并发上限避免超发空转 2026-09-18 20:58:55 +08:00
yuming 028fcbe990 fix: 修复批量下载漏检测稀疏行和底部半行
根因:
1. std 横向采样区 x=350-1050 不覆盖左列卡片(中心x=201),
   导致只有左列有卡片的稀疏行 std 被拉低到27,低于阈值28被误判为间隙
2. 扫描范围 ROW_Y1=2300 太靠上,底部半行被边缘过滤
3. 每轮只处理前3行(full_rows[:3]),后续行即使检测到也被跳过

修复:
- ROW_STD_X0 350→50, ROW_STD_W 700→980(覆盖全部3列)
- ROW_STD_LOW 28→18(分组标签std=0不会误判)
- ROW_Y1 2300→2380(扩大底部扫描范围)
- 处理所有检测到的行,未打勾格子最多取前9个避免超并发
2026-09-18 18:05:33 +08:00
yuming 913b0c9ca7 test: 新增automate.py Pillow图像函数回归测试(9个用例)
覆盖 detect_card_rows 行检测、count_blue_pixels 蓝勾检测、
sample_rgb_from_file 像素采样、std_profile 标准差剖面、
date_label_hash/screenshot_hash 哈希稳定性。全部通过。
2026-09-18 16:13:22 +08:00
yuming 80e3815b2f fix: 用Pillow替代ImageMagick,修复批量下载漏下载的4个逻辑漏洞
漏下载根因:
1. ImageMagick(magick)未安装,detect_card_rows/std_profile/count_blue_pixels
   全部依赖magick命令,命令不存在时抛FileNotFoundError,sync_recent_days
   捕获异常后直接跳过当天 → 整天漏下载
2. detect_card_rows两次返回空就误判'全部已打勾'直接滚动 → 漏掉当前屏所有卡片
3. 到底清扫时检测不到行就误判'清扫完成'提前结束 → 漏掉底部卡片
4. 点击超时未打勾的卡片不重试 → 点击落在间隙/标签上的卡片永远漏下载
5. sync_recent_days某天异常直接跳过不重试 → 偶发ANR导致整天漏下载

修复:
- 全部magick调用(sample_pixel_rgb/count_blue_pixels/std_profile/date_label_hash/
  screenshot_hash)替换为Pillow实现,消除系统级外部依赖
- detect_card_rows返回空时重试3次(逐次加长等待),仍为空则跳过本轮不盲点
- 到底清扫时检测不到行重试3次,仍为空才判底部无内容
- 点击超时未打勾的卡片重试点击一次,再等一轮
- sync_recent_days每天最多重试1次(共2次尝试)
2026-09-18 16:06:50 +08:00
yuming 55361a8f58 perf: webapp事件接口加分页、motion默认隔帧采样、新增20个单元测试
- /api/events 支持 limit/offset 分页参数,默认500上限2000
- MOTION_SAMPLE_STRIDE 默认从1改为2,长视频提速一倍
- 新增 tests/test_core.py:覆盖MD5、活动窗口合并、摔倒检测、活动标签合并、Ollama解析等20个纯函数测试
2026-09-18 15:55:09 +08:00
yuming b3608f534e feat: 新增视频分析流水线、Web界面、孤儿清理工具;修复健壮性问题
- 新增 analysis/ 模块:MOG2活动检测、YOLO人物跟踪、MediaPipe摔倒检测、Ollama活动打标、缩略图抽取
- 新增 webapp/:Flask事件浏览界面,按需HEVC→H264转码缓存
- 新增 analyze.py:7阶段状态机流水线(scan→backfill→motion→person→fall→activity→thumbnail)
- 新增 cleanup_orphans.py:清理文件已删除的数据库孤儿记录
- 修复:motion/person/activity三阶段加单视频异常兜底,避免批次中断
- 修复:ollama_client加健康检查和3次重试,num_ctx从8192降到2048
- 修复:transcode_cache加fcntl文件锁,防止并发转码同一视频
- 修复:SQLite开WAL模式+busy_timeout,支持读写并发
- 修复:sync.py ffmpeg路径改为shutil.which查找,删除automate.py死代码PYTHON
- 修复:thumbnail失败加3次重试上限,避免无限重试损坏文件
- 新增:requirements.lock锁定版本、README文档
- 更新:.gitignore排除.playwright-mcp、截图、备份文件
2026-09-18 15:52:49 +08:00
yuming 203cca42a2 fix: 动态定位卡片行根治萤石批量下载重复问题
根因:批量下载列表是自绘 UI,原用写死的网格坐标 THUMB_Y 定位卡片;但列表
每隔一段插入"整点分组标签"行,滚动跨过标签会使所有卡片整行错位(实测漂移
达 174px),导致点击落空、打勾检测采样错位、已下载卡片被误判重下,重复率约 30%。

修复与优化:
- 改用亮度标准差(std)垂直剖面每屏动态实测卡片行真实中心,替代写死的 THUMB_Y
- 每轮最多点前 3 行(9 格),避免超萤石约 10 个的并发下载上限
- 到底判定改用"卡片行有没有整体平移",抗底部灰占位抖动(原像素哈希会一直空滚)
- 到底时清扫当前屏剩余未下载卡片,补下被限 3 行漏掉的底部几个
- 截图改 exec-out 一步式 + 复用检测结果 + 缩短固定等待,单轮 ~12s 降到 ~7s

实测重复率 30% → 0%,端到端跑通(启动→下载→到底清扫→拉取到本地→同步完成)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-27 15:31:40 +08:00
33 changed files with 3907 additions and 167 deletions
+24 -1
View File
@@ -1,11 +1,21 @@
# 敏感配置和缓存 # 敏感配置和缓存
config.py /config.py
token_cache.json token_cache.json
sync.log sync.log
# 下载的录像(每天几百 MB) # 下载的录像(每天几百 MB)
recordings/ recordings/
# 本地视频分析产出(运行时生成)
analysis.db
analysis.db.bak.*
thumbnails/
transcoded/
# 模型权重(体积大,首次运行自动下载)
analysis/models/
yolo11n.pt
# Python # Python
__pycache__/ __pycache__/
*.pyc *.pyc
@@ -20,3 +30,16 @@ __pycache__/
# 临时 # 临时
*.tmp *.tmp
/tmp/ /tmp/
# 重复视频检测报告(运行时生成)
dedup_report.txt
# Playwright MCP 运行日志/快照
.playwright-mcp/
# 开发截图
webapp-check.png
webapp-index.png
# Frida
*.frida
+170
View File
@@ -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 <repo> && 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` 清理多余文件。
+121
View File
@@ -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("===== 打标完成 =====")
+48
View File
@@ -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("===== 回填完成 =====")
+90
View File
@@ -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)
+65
View File
@@ -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 = 2 # 每隔几帧采样一次(2=隔帧,长视频提速且不影响活动窗口检测精度)
# 人物出现次数统计(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 # 连续同标签采样点之间允许的最大间隔,超过则视为新的一段活动
+222
View File
@@ -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]
+130
View File
@@ -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("已取消")
+67
View File
@@ -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("===== 检测完成 =====")
+266
View File
@@ -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("===== 检测完成 =====")
+16
View File
@@ -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)
+10
View File
@@ -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()
+145
View File
@@ -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("===== 检测完成 =====")
+86
View File
@@ -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}"
+158
View File
@@ -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("===== 统计完成 =====")
Binary file not shown.
Binary file not shown.
+103
View File
@@ -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("===== 生成完成 =====")
+158
View File
@@ -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
+53
View File
@@ -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);
+167
View File
@@ -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("===== 扫描完成 =====")
+72
View File
@@ -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()
+910 -161
View File
File diff suppressed because it is too large Load Diff
+70
View File
@@ -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
+10
View File
@@ -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
+8 -5
View File
@@ -1,9 +1,12 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
萤石CB60 SD卡录像每日同步脚本 [已废弃,保留作为备选] 萤石CB60 SD卡录像每日同步脚本(开放平台 API 方案)
每天凌晨运行,下载前24小时内的所有SD卡录像到本地目录。
当前主力下载方案是 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 datetime import datetime
from pathlib import Path from pathlib import Path
import requests import requests
@@ -15,8 +18,8 @@ OUTPUT_DIR = Path(OUTPUT_DIR)
TOKEN_CACHE = Path(TOKEN_CACHE) TOKEN_CACHE = Path(TOKEN_CACHE)
LOG_FILE = Path("./sync.log") LOG_FILE = Path("./sync.log")
# NAS上部署时改为 "ffmpeg"(Docker镜像内) # ffmpeg 从 PATH 查找;NAS/Docker 部署时确保 ffmpeg 在 PATH 里即可,无需改代码
FFMPEG = "/opt/homebrew/bin/ffmpeg" FFMPEG = shutil.which("ffmpeg") or "ffmpeg"
BASE = "https://open.ys7.com/api/lapp" BASE = "https://open.ys7.com/api/lapp"
+153
View File
@@ -0,0 +1,153 @@
"""automate.py 的 Pillow 图像分析函数回归测试。
覆盖:detect_card_rows 行检测、count_blue_pixels 蓝勾检测、
sample_rgb_from_file 像素采样、std_profile 标准差剖面、
date_label_hash / screenshot_hash 哈希稳定性。
这些测试用合成图片验证逻辑,不需要启动模拟器。
"""
import os
import random
import tempfile
import unittest
from PIL import Image, ImageDraw
# 把项目根目录加入 path
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import automate
class TestDetectCardRows(unittest.TestCase):
def test_detects_three_rows(self):
img = Image.new("RGB", (1080, 2400), (240, 240, 240))
random.seed(42)
for row_top in [1060, 1280, 1500]:
for y in range(row_top, row_top + 180):
for x in range(100, 980):
img.putpixel((x, y), (
random.randint(50, 200),
random.randint(50, 200),
random.randint(50, 200),
))
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
path = f.name
try:
img.save(path)
rows = automate.detect_card_rows(path)
self.assertEqual(len(rows), 3)
for expected, actual in zip([1150, 1370, 1590], rows):
self.assertLess(abs(expected - actual), 20)
finally:
os.unlink(path)
def test_empty_image_returns_no_rows(self):
img = Image.new("RGB", (1080, 2400), (200, 200, 200))
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
path = f.name
try:
img.save(path)
rows = automate.detect_card_rows(path)
self.assertEqual(rows, [])
finally:
os.unlink(path)
class TestCountBluePixels(unittest.TestCase):
def test_blue_badge_detected(self):
img = Image.new("RGB", (1080, 2400), (200, 200, 200))
draw = ImageDraw.Draw(img)
draw.rectangle([500, 1160, 580, 1240], fill=(100, 143, 252))
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
path = f.name
try:
img.save(path)
cnt = automate.count_blue_pixels(path, 540, 1200)
self.assertGreaterEqual(cnt, 500)
finally:
os.unlink(path)
def test_gray_area_not_detected(self):
img = Image.new("RGB", (1080, 2400), (200, 200, 200))
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
path = f.name
try:
img.save(path)
cnt = automate.count_blue_pixels(path, 200, 1200)
self.assertLess(cnt, 500)
finally:
os.unlink(path)
class TestSampleRgb(unittest.TestCase):
def test_sample_pixel(self):
img = Image.new("RGB", (100, 100), (0, 0, 0))
img.putpixel((50, 50), (255, 128, 64))
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
path = f.name
try:
img.save(path)
rgb = automate.sample_rgb_from_file(path, 50, 50)
self.assertEqual(rgb, (255, 128, 64))
finally:
os.unlink(path)
def test_out_of_bounds_returns_zero(self):
img = Image.new("RGB", (10, 10), (10, 20, 30))
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
path = f.name
try:
img.save(path)
rgb = automate.sample_rgb_from_file(path, 999, 999)
self.assertEqual(rgb, (0, 0, 0))
finally:
os.unlink(path)
class TestStdProfile(unittest.TestCase):
def test_textured_rows_have_high_std(self):
img = Image.new("RGB", (1080, 2400), (240, 240, 240))
random.seed(123)
for y in range(1100, 1300):
for x in range(350, 1050):
img.putpixel((x, y), (random.randint(0, 255),) * 3)
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
path = f.name
try:
img.save(path)
prof = automate.std_profile(path)
textured = sum(1 for _, s in prof if s >= 28)
self.assertGreater(textured, 100)
finally:
os.unlink(path)
class TestHashStability(unittest.TestCase):
def test_date_label_hash_stable(self):
img = Image.new("RGB", (1080, 2400), (100, 150, 200))
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
path = f.name
try:
img.save(path)
h1 = automate.date_label_hash(path)
h2 = automate.date_label_hash(path)
self.assertEqual(h1, h2)
finally:
os.unlink(path)
def test_screenshot_hash_stable(self):
img = Image.new("RGB", (1080, 2400), (100, 150, 200))
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
path = f.name
try:
img.save(path)
h1 = automate.screenshot_hash(path)
h2 = automate.screenshot_hash(path)
self.assertEqual(h1, h2)
finally:
os.unlink(path)
if __name__ == "__main__":
unittest.main()
+135
View File
@@ -0,0 +1,135 @@
"""基础单元测试:覆盖不依赖外部工具/模型的纯函数。
运行:.venv/bin/python -m pytest tests/ -v
或:.venv/bin/python tests/test_core.py
"""
import hashlib
import os
import tempfile
import unittest
from analysis.hashing import compute_md5
from analysis.motion_filter import merge_motion_windows
from analysis.fall_detector import _find_fall_windows
from analysis.activity_tagger import _merge_labeled_samples
from analysis.ollama_client import _parse_label
class TestHashing(unittest.TestCase):
def test_compute_md5_matches_hashlib(self):
with tempfile.NamedTemporaryFile(delete=False) as f:
f.write(b"hello world" * 1000)
path = f.name
try:
expected = hashlib.md5(b"hello world" * 1000).hexdigest()
self.assertEqual(compute_md5(path), expected)
finally:
os.unlink(path)
def test_compute_md5_empty_file(self):
with tempfile.NamedTemporaryFile(delete=False) as f:
path = f.name
try:
self.assertEqual(compute_md5(path), hashlib.md5(b"").hexdigest())
finally:
os.unlink(path)
class TestMergeMotionWindows(unittest.TestCase):
def test_single_window(self):
flags = [(0.0, True), (0.5, True), (1.0, True)]
self.assertEqual(merge_motion_windows(flags), [[0.0, 1.0]])
def test_merge_nearby_windows(self):
# 两段活动间隔 1.0s < MERGE_GAP 1.5s,应合并
flags = [(0.0, True), (0.5, True), (2.0, True), (2.5, True)]
result = merge_motion_windows(flags)
self.assertEqual(len(result), 1)
self.assertEqual(result[0], [0.0, 2.5])
def test_separate_windows(self):
# 第一段 0-0.5s 活动,中间 1.0-3.5s 无活动(间隔3s > MERGE_GAP 1.5s),第二段 4.0-4.5s 活动
flags = [(0.0, True), (0.5, True), (1.0, False), (2.0, False),
(3.0, False), (3.5, False), (4.0, True), (4.5, True)]
result = merge_motion_windows(flags)
self.assertEqual(len(result), 2)
def test_filter_short_windows(self):
# 0.3s < MIN_DURATION 0.5s,应被过滤
flags = [(0.0, True), (0.3, True)]
self.assertEqual(merge_motion_windows(flags), [])
def test_empty_input(self):
self.assertEqual(merge_motion_windows([]), [])
class TestFallDetection(unittest.TestCase):
def test_upright_to_prone_fast(self):
# 直立(0s) → 1秒后横躺(1s) → 持续到3s,应检测为摔倒
states = [(0.0, "upright"), (1.0, "prone"), (2.0, "prone"), (3.0, "prone")]
windows = _find_fall_windows(states)
self.assertEqual(len(windows), 1)
self.assertEqual(windows[0], (1.0, 3.0))
def test_slow_prone_not_fall(self):
# 直立(0s) → 3秒后才横躺(3s) > TRANSITION_MAX 1.5s,不是摔倒
states = [(0.0, "upright"), (3.0, "prone"), (4.0, "prone")]
self.assertEqual(_find_fall_windows(states), [])
def test_prone_without_prior_upright(self):
# 视频开头就是横躺,没有直立前置,不算摔倒
states = [(0.0, "prone"), (1.0, "prone"), (2.0, "prone")]
self.assertEqual(_find_fall_windows(states), [])
def test_prone_too_short(self):
# 横躺只持续 0.5s < SUSTAIN_MIN 1.0s,不算摔倒
states = [(0.0, "upright"), (0.5, "prone"), (1.0, "upright")]
self.assertEqual(_find_fall_windows(states), [])
def test_unknown_states_ignored(self):
states = [(0.0, "upright"), (0.5, "unknown"), (1.0, "prone"),
(2.0, "prone"), (3.0, "prone")]
windows = _find_fall_windows(states)
self.assertEqual(len(windows), 1)
class TestActivityMerge(unittest.TestCase):
def test_merge_same_label(self):
samples = [(0.0, "走动", "raw1"), (2.0, "走动", "raw2")]
segments = _merge_labeled_samples(samples)
self.assertEqual(len(segments), 1)
self.assertEqual(segments[0]["label"], "走动")
self.assertEqual(segments[0]["start"], 0.0)
self.assertEqual(segments[0]["end"], 2.0)
def test_different_labels_separate(self):
samples = [(0.0, "走动", "raw1"), (5.0, "坐下", "raw2")]
segments = _merge_labeled_samples(samples)
self.assertEqual(len(segments), 2)
def test_gap_too_large_separates(self):
# 间隔 5.0s > MERGE_GAP 3.0s,即使同标签也分开
samples = [(0.0, "走动", "raw1"), (5.0, "走动", "raw2")]
segments = _merge_labeled_samples(samples)
self.assertEqual(len(segments), 2)
class TestOllamaParse(unittest.TestCase):
def test_exact_label(self):
self.assertEqual(_parse_label("走动"), "走动")
def test_label_with_punctuation(self):
self.assertEqual(_parse_label("走动。"), "走动")
def test_label_in_sentence(self):
self.assertEqual(_parse_label("这个人在坐下休息"), "坐下")
def test_unknown_fallback(self):
self.assertEqual(_parse_label("xyz 不存在的标签"), "其他")
def test_empty_string(self):
self.assertEqual(_parse_label(""), "其他")
if __name__ == "__main__":
unittest.main()
+109
View File
@@ -0,0 +1,109 @@
"""本地事件展示网页:纯只读查询 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
limit = min(int(request.args.get("limit", 500)), 2000)
offset = max(int(request.args.get("offset", 0)), 0)
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 ? OFFSET ?"
params.extend([limit, offset])
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/<int:video_id>/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/<int:event_id>")
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)
+108
View File
@@ -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 = `
<div class="event-type">${formatEventLabel(event)}</div>
<div class="event-time">${event.timestamp_abs || event.capture_date}</div>
`;
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();
+96
View File
@@ -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;
}
+41
View File
@@ -0,0 +1,41 @@
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>录像事件回放</title>
<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
</head>
<body>
<h1>录像事件回放</h1>
<div class="filters">
<label>事件类型
<select id="event-type-filter">
<option value="">全部</option>
<option value="person_presence">人物出现</option>
<option value="fall">摔倒</option>
<option value="activity">活动</option>
</select>
</label>
<label>日期
<input type="date" id="date-filter">
</label>
<span id="stats"></span>
</div>
<div class="cache-panel">
<span id="cache-info"></span>
<button id="clear-cache-btn">清空转码缓存</button>
</div>
<div class="main">
<div class="player-panel">
<video id="player" controls></video>
<div id="player-info"></div>
</div>
<div id="event-list" class="event-list"></div>
</div>
<script src="{{ url_for('static', filename='app.js') }}"></script>
</body>
</html>
+96
View File
@@ -0,0 +1,96 @@
"""网页播放的按需转码缓存:源文件mp4容器HEVC轨道sample entry编码标签缺失,
浏览器<video>标签无法识别,第一次点击播放时用ffmpeg(VideoToolbox硬件加速)转码成
标准H264并缓存到 TRANSCODED_DIR,之后复用缓存不重复转码;不预先转码全部录像。
"""
import fcntl
import subprocess
from pathlib import Path
from analysis.config import TRANSCODED_DIR
def cached_path(video_id):
return TRANSCODED_DIR / f"{video_id}.mp4"
def _lock_path(video_id):
return TRANSCODED_DIR / f"{video_id}.lock"
def ensure_transcoded(video_id, source_path):
"""返回可直接播放的文件路径。已有缓存直接复用,否则先转码再缓存。
用 fcntl 文件锁防止两个请求同时转码同一个视频:先到的转码,后到的等锁释放后
直接读缓存。锁文件在转码完成后删除。
"""
output_path = cached_path(video_id)
if output_path.exists():
return output_path
TRANSCODED_DIR.mkdir(exist_ok=True)
lock_file = _lock_path(video_id)
lock_fd = open(lock_file, "w")
try:
# 阻塞等待锁,最多等 10 分钟(转码大文件可能慢)
fcntl.flock(lock_fd, fcntl.LOCK_EX)
# 拿到锁后再检查一次:可能前面的进程已经转码完成
if output_path.exists():
return output_path
tmp_path = output_path.with_suffix(".mp4.tmp")
result = subprocess.run(
[
"ffmpeg", "-i", str(source_path),
"-c:v", "h264_videotoolbox", "-b:v", "1500k",
"-c:a", "aac",
"-movflags", "+faststart",
"-f", "mp4", # 输出文件名带.tmp后缀,ffmpeg无法从扩展名猜出格式,需显式指定
"-y", str(tmp_path),
],
capture_output=True, timeout=600,
)
if result.returncode != 0 or not tmp_path.exists():
tmp_path.unlink(missing_ok=True)
raise RuntimeError(result.stderr.decode("utf-8", errors="ignore"))
tmp_path.rename(output_path) # 转码完成后原子改名,避免并发请求读到半成品文件
return output_path
finally:
fcntl.flock(lock_fd, fcntl.LOCK_UN)
lock_fd.close()
lock_file.unlink(missing_ok=True)
def cache_stats():
if not TRANSCODED_DIR.exists():
return {"file_count": 0, "total_size_bytes": 0, "files": []}
files = []
total = 0
for f in TRANSCODED_DIR.glob("*.mp4"):
size = f.stat().st_size
total += size
files.append({"video_id": int(f.stem), "size_bytes": size})
return {"file_count": len(files), "total_size_bytes": total, "files": files}
def clear_cache():
if not TRANSCODED_DIR.exists():
return {"deleted": 0, "freed_bytes": 0}
deleted = 0
freed = 0
for f in TRANSCODED_DIR.glob("*.mp4"):
freed += f.stat().st_size
f.unlink()
deleted += 1
# 清理残留的 .tmp 和 .lock 文件
for f in TRANSCODED_DIR.glob("*.tmp"):
f.unlink(missing_ok=True)
for f in TRANSCODED_DIR.glob("*.lock"):
f.unlink(missing_ok=True)
return {"deleted": deleted, "freed_bytes": freed}