404 lines
18 KiB
Python
404 lines
18 KiB
Python
from datetime import datetime
|
|
import json
|
|
import asyncio
|
|
from typing import Dict
|
|
from fastapi import HTTPException
|
|
from openai import OpenAI
|
|
from .config import (
|
|
CAMERA_DB_MAPPING,
|
|
BEHAVIOR_CATEGORIES,
|
|
ABNORMAL_BEHAVIORS,
|
|
REDIS_CACHE_CONFIG,
|
|
ai_client
|
|
)
|
|
|
|
async def get_camera_data_by_date(camera_id: str, date: str, redis_client, is_face: bool = False):
|
|
"""获取摄像头某天的所有数据"""
|
|
try:
|
|
if camera_id not in CAMERA_DB_MAPPING:
|
|
raise HTTPException(status_code=400, detail="Invalid camera ID")
|
|
|
|
# 使用新的键格式进行模式匹配
|
|
pattern = f"{'face_' if is_face else ''}{camera_id}_{date}_*"
|
|
all_keys = redis_client.keys(pattern)
|
|
|
|
if not all_keys:
|
|
return {"message": "No data found", "data": None}
|
|
|
|
# 获取所有键的数据并解析
|
|
all_data = {}
|
|
for key in all_keys:
|
|
data = redis_client.get(key)
|
|
if data:
|
|
all_data[key] = json.loads(data)
|
|
|
|
return {
|
|
"message": "success",
|
|
"data": all_data,
|
|
"total_records": len(all_data)
|
|
}
|
|
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
def background_generate_report(date_no_hyphen: str, redis_connections):
|
|
"""后台生成报告的函数"""
|
|
try:
|
|
print(f"\n=== 后台任务开始生成报告 {date_no_hyphen} ===")
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
report_redis = redis_connections["report"]
|
|
task_key = f"task_status_{date_no_hyphen}"
|
|
report_redis.setex(
|
|
task_key,
|
|
REDIS_CACHE_CONFIG["task_status_expiry"],
|
|
json.dumps({"status": "running"})
|
|
)
|
|
|
|
print("开始调用 generate_daily_report...")
|
|
report = loop.run_until_complete(generate_daily_report(date_no_hyphen, redis_connections))
|
|
print(f"报告生成结果: {report.get('message', 'unknown')}")
|
|
|
|
if report.get("message") != "no_data":
|
|
print("报告生成成功，准备保存到Redis...")
|
|
report_key = f"report_{date_no_hyphen}"
|
|
report_redis.setex(
|
|
report_key,
|
|
REDIS_CACHE_CONFIG["report_expiry"],
|
|
json.dumps(report)
|
|
)
|
|
print(f"报告已保存到Redis，键名: {report_key}")
|
|
report_redis.setex(
|
|
task_key,
|
|
REDIS_CACHE_CONFIG["task_status_expiry"],
|
|
json.dumps({"status": "completed"})
|
|
)
|
|
else:
|
|
report_redis.setex(
|
|
task_key,
|
|
REDIS_CACHE_CONFIG["task_status_expiry"],
|
|
json.dumps({"status": "completed"})
|
|
)
|
|
|
|
print("任务状态已更新为completed")
|
|
|
|
except Exception as e:
|
|
print(f"报告生成失败: {str(e)}")
|
|
report_redis.setex(
|
|
task_key,
|
|
REDIS_CACHE_CONFIG["task_status_expiry"],
|
|
json.dumps({
|
|
"status": "failed",
|
|
"error": str(e)
|
|
})
|
|
)
|
|
print("任务状态已更新为failed")
|
|
finally:
|
|
loop.close()
|
|
print(f"=== 后台任务结束 {date_no_hyphen} ===\n")
|
|
|
|
async def generate_daily_report(date: str, redis_connections) -> Dict:
|
|
"""生成每日分析报告"""
|
|
print(f"\n=== 开始生成日报 {date} ===")
|
|
|
|
# 初始化数据收集结构
|
|
data_collection = {
|
|
"date": date,
|
|
"total_events": 0,
|
|
"abnormal_events": 0,
|
|
"camera_num": set(),
|
|
"activity_areas": {}, # 活动区域统计
|
|
"behavior_distribution": {}, # 行为分布
|
|
"hourly_stats": {}, # 每小时统计
|
|
"category_stats": { # 各类别行为统计
|
|
category: {
|
|
"count": 0,
|
|
"behaviors": {}
|
|
} for category in BEHAVIOR_CATEGORIES.keys()
|
|
},
|
|
"abnormal_stats": {
|
|
"behaviors": [],
|
|
"times": [],
|
|
"locations": []
|
|
}
|
|
}
|
|
|
|
# 初始化摄像头小时统计
|
|
camera_hourly_counts = {
|
|
camera_id: {f"{hour:02d}": 0 for hour in range(24)}
|
|
for camera_id in CAMERA_DB_MAPPING.keys()
|
|
if camera_id != "report"
|
|
}
|
|
|
|
# 遍历所有摄像头数据
|
|
has_any_data = False
|
|
total_cameras = len([cam for cam in CAMERA_DB_MAPPING.keys() if cam != "report"])
|
|
processed_cameras = 0
|
|
|
|
print(f"开始处理 {total_cameras} 个摄像头的数据")
|
|
|
|
# 数据收集和预处理
|
|
for camera_id, redis_client in redis_connections.items():
|
|
if camera_id == "report":
|
|
continue
|
|
|
|
processed_cameras += 1
|
|
print(f"\n处理摄像头 {camera_id} ({processed_cameras}/{total_cameras})")
|
|
|
|
camera_event_count = 0
|
|
|
|
for hour in range(24):
|
|
hour_str = f"{hour:02d}"
|
|
pattern = f"{camera_id}_{date}_{hour_str}*"
|
|
hour_keys = redis_client.keys(pattern)
|
|
|
|
for key in hour_keys:
|
|
hour_data = redis_client.get(key)
|
|
if hour_data:
|
|
has_any_data = True
|
|
hour_json = json.loads(hour_data)
|
|
|
|
for video_file, video_data in hour_json.items():
|
|
if "video_analysis" in video_data:
|
|
analysis = video_data["video_analysis"]["qwen-7B"]["extracted_info"]
|
|
|
|
behaviors = analysis.get("actions", [])
|
|
camera_event_count += len(behaviors)
|
|
data_collection["total_events"] += len(behaviors)
|
|
|
|
environment = analysis.get("environment", "")
|
|
if environment:
|
|
if isinstance(environment, list):
|
|
for env in environment:
|
|
if isinstance(env, str):
|
|
data_collection["activity_areas"][env] = \
|
|
data_collection["activity_areas"].get(env, 0) + 1
|
|
elif isinstance(environment, str):
|
|
data_collection["activity_areas"][environment] = \
|
|
data_collection["activity_areas"].get(environment, 0) + 1
|
|
|
|
if hour_str not in data_collection["hourly_stats"]:
|
|
data_collection["hourly_stats"][hour_str] = {
|
|
"event_count": 0,
|
|
"categories": {cat: 0 for cat in BEHAVIOR_CATEGORIES.keys()}
|
|
}
|
|
|
|
data_collection["hourly_stats"][hour_str]["event_count"] += len(behaviors)
|
|
camera_hourly_counts[camera_id][hour_str] += len(behaviors)
|
|
|
|
for behavior in behaviors:
|
|
data_collection["behavior_distribution"][behavior] = \
|
|
data_collection["behavior_distribution"].get(behavior, 0) + 1
|
|
|
|
behavior_categorized = False
|
|
for category, keywords in BEHAVIOR_CATEGORIES.items():
|
|
if any(keyword in behavior for keyword in keywords):
|
|
data_collection["category_stats"][category]["count"] += 1
|
|
|
|
if behavior not in data_collection["category_stats"][category]["behaviors"]:
|
|
data_collection["category_stats"][category]["behaviors"][behavior] = {
|
|
"count": 0,
|
|
"occurrences": {}
|
|
}
|
|
|
|
occurrence_key = f"{camera_id}_{hour_str}"
|
|
if occurrence_key not in data_collection["category_stats"][category]["behaviors"][behavior]["occurrences"]:
|
|
data_collection["category_stats"][category]["behaviors"][behavior]["count"] += 1
|
|
data_collection["category_stats"][category]["behaviors"][behavior]["occurrences"][occurrence_key] = {
|
|
"time": f"{hour_str}:00",
|
|
"camera": camera_id
|
|
}
|
|
|
|
data_collection["hourly_stats"][hour_str]["categories"][category] += 1
|
|
behavior_categorized = True
|
|
break
|
|
|
|
if not behavior_categorized:
|
|
data_collection["category_stats"]["其他"]["count"] += 1
|
|
if behavior not in data_collection["category_stats"]["其他"]["behaviors"]:
|
|
data_collection["category_stats"]["其他"]["behaviors"][behavior] = {
|
|
"count": 0,
|
|
"occurrences": {}
|
|
}
|
|
|
|
occurrence_key = f"{camera_id}_{hour_str}"
|
|
if occurrence_key not in data_collection["category_stats"]["其他"]["behaviors"][behavior]["occurrences"]:
|
|
data_collection["category_stats"]["其他"]["behaviors"][behavior]["count"] += 1
|
|
data_collection["category_stats"]["其他"]["behaviors"][behavior]["occurrences"][occurrence_key] = {
|
|
"time": f"{hour_str}:00",
|
|
"camera": camera_id
|
|
}
|
|
|
|
data_collection["hourly_stats"][hour_str]["categories"]["其他"] += 1
|
|
|
|
if any(abnormal in behavior for abnormal in ABNORMAL_BEHAVIORS):
|
|
occurrence_key = f"{camera_id}_{hour_str}"
|
|
abnormal_key = f"{behavior}_{occurrence_key}"
|
|
if abnormal_key not in data_collection["abnormal_stats"]["behaviors"]:
|
|
data_collection["abnormal_events"] += 1
|
|
data_collection["abnormal_stats"]["behaviors"].append({
|
|
"behavior": behavior,
|
|
"time": f"{hour_str}:00",
|
|
"camera": camera_id
|
|
})
|
|
else:
|
|
print(f" - {hour_str}时 无数据")
|
|
|
|
print(f"摄像头 {camera_id} 总计: {camera_event_count} 个事件")
|
|
if camera_event_count > 0:
|
|
data_collection["camera_num"].add(camera_id)
|
|
|
|
if len(data_collection["camera_num"]) == 0:
|
|
print("判定为无数据，返回")
|
|
return {
|
|
"message": "no_data",
|
|
"data": None,
|
|
"detail": "暂无数据"
|
|
}
|
|
|
|
data_collection["camera_num"] = list(data_collection["camera_num"])
|
|
|
|
sorted_hours = sorted(
|
|
data_collection["hourly_stats"].items(),
|
|
key=lambda x: x[1]["event_count"],
|
|
reverse=True
|
|
)
|
|
data_collection["peak_hours"] = [hour for hour, _ in sorted_hours[:3]]
|
|
|
|
preprocessed_data = {
|
|
"日期": data_collection["date"],
|
|
"摄像头数量": len(data_collection["camera_num"]),
|
|
"行为总数": data_collection["total_events"],
|
|
"异常行为数": data_collection["abnormal_events"],
|
|
"行为高峰时段": data_collection["peak_hours"],
|
|
"主要活动区域": data_collection["activity_areas"],
|
|
"行为类别统计": data_collection["category_stats"],
|
|
"异常行为统计": data_collection["abnormal_stats"],
|
|
"每小时行为统计": data_collection["hourly_stats"]
|
|
}
|
|
|
|
ai_analysis = await analyze_experiment_data(preprocessed_data)
|
|
|
|
final_report = {
|
|
"整体活动趋势": ai_analysis["整体活动趋势"],
|
|
"高峰时段分析": ai_analysis["高峰时段分析"],
|
|
"异常行为分析": ai_analysis["异常行为分析"],
|
|
"行为分析": ai_analysis["行为分析"],
|
|
"建议": ai_analysis["建议"]
|
|
}
|
|
|
|
final_report["hourly_distribution"] = []
|
|
for camera_id in camera_hourly_counts:
|
|
total_events = sum(camera_hourly_counts[camera_id].values())
|
|
if total_events > 0:
|
|
camera_data = {
|
|
"camera_id": camera_id,
|
|
"data": []
|
|
}
|
|
for hour in range(24):
|
|
hour_str = f"{hour:02d}"
|
|
hour_data = {
|
|
"hour": f"{hour_str}:00",
|
|
"count": camera_hourly_counts[camera_id][hour_str],
|
|
"categories": data_collection["hourly_stats"].get(hour_str, {}).get("categories", {})
|
|
}
|
|
camera_data["data"].append(hour_data)
|
|
final_report["hourly_distribution"].append(camera_data)
|
|
|
|
return final_report
|
|
|
|
async def analyze_experiment_data(report_info):
|
|
"""使用AI分析实验数据"""
|
|
system_prompt = """
|
|
You are an AI assistant tasked with analyzing data.
|
|
Generate a comprehensive analysis report in JSON format.
|
|
The JSON structure must strictly follow the provided template.
|
|
"""
|
|
|
|
user_prompt = f"""Analyze the preprocessed data based on the following information:
|
|
Preprocessed data: {json.dumps(report_info, ensure_ascii=False)}
|
|
|
|
Generate a JSON response with the following structure:
|
|
|
|
{{
|
|
"整体活动趋势": {{
|
|
"日期": "{report_info['日期']}",
|
|
"摄像头数量": {report_info['摄像头数量']},
|
|
"行为总数": {report_info['行为总数']},
|
|
"异常行为数": {report_info['异常行为数']},
|
|
"行为高峰时段": {json.dumps(report_info['行为高峰时段'], ensure_ascii=False)},
|
|
"主要活动区域": {json.dumps(report_info['主要活动区域'], ensure_ascii=False)}
|
|
}},
|
|
"高峰时段分析": {{
|
|
"高峰时段": "分析行为高峰时段",
|
|
"高峰时段行为": "分析高峰时段主要行为",
|
|
"活动规律": "分析活动规律"
|
|
}},
|
|
"异常行为分析": {{
|
|
"异常行为": "分析异常行为类型",
|
|
"异常行为次数": "分析异常行为频率",
|
|
"异常行为出现时间": "分析异常行为时间分布",
|
|
"异常行为地点": "分析监测到异常行为的摄像头"
|
|
}},
|
|
"行为分析": {{
|
|
"基础动作": {{
|
|
"站立行为": "分析站立相关行为",
|
|
"行走行为": "分析行走相关行为",
|
|
"坐卧行为": "分析坐卧相关行为",
|
|
"其他基础动作": "分析其他基础动作"
|
|
}},
|
|
"日常生活": {{
|
|
"饮食情况": "分析饮食相关行为",
|
|
"休息情况": "分析休息相关行为",
|
|
"医疗情况": "分析医疗相关行为"
|
|
}},
|
|
"社交活动": {{
|
|
"交际情况": "分析交际相关行为",
|
|
"娱乐情况": "分析娱乐相关行为",
|
|
"情感表达": "分析情感表达相关行为"
|
|
}},
|
|
"工作学习": {{
|
|
"学习情况": "分析学习相关行为",
|
|
"工作情况": "分析学习相关行为",
|
|
"创作活动": "分析创作相关行为"
|
|
}},
|
|
"运动娱乐": {{
|
|
"运动情况": "分析运动相关行为",
|
|
"运动时长": "分析运动持续时间",
|
|
"运动强度": "分析运动强度"
|
|
}},
|
|
"其他行为": {{
|
|
"出现时间":"分析其他行为出现时间",
|
|
"出现次数":"分析其他行为出现次数"
|
|
}}
|
|
}},
|
|
"建议": {{
|
|
"生活作息": ["建议1", "建议2"],
|
|
"活动安排": ["建议1", "建议2"],
|
|
"安全防护": ["建议1", "建议2"],
|
|
"健康建议": ["建议1", "建议2"]
|
|
}}
|
|
}}
|
|
"""
|
|
|
|
try:
|
|
response = ai_client.chat.completions.create(
|
|
model="deepseek-chat",
|
|
messages=[
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": user_prompt}
|
|
],
|
|
max_tokens=4096,
|
|
temperature=0.7,
|
|
response_format={'type': 'json_object'}
|
|
)
|
|
|
|
return json.loads(response.choices[0].message.content)
|
|
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail="AI分析服务暂时不可用，请稍后重试"
|
|
)
|