yuming
b3608f534e
- 新增 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、截图、备份文件
122 lines
4.1 KiB
Python
122 lines
4.1 KiB
Python
"""在 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("===== 打标完成 =====")
|