99 lines
4.0 KiB
Python
99 lines
4.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
from datetime import datetime, date
|
||
from pathlib import Path
|
||
from statistics import mean
|
||
|
||
|
||
class FishHistoryManager:
|
||
"""维护每日鱼类活动档案，并在指定时间生成报告。"""
|
||
|
||
def __init__(self, report_dir: str = "reports", report_hour: int = 20):
|
||
self.report_dir = Path(report_dir)
|
||
self.report_dir.mkdir(parents=True, exist_ok=True)
|
||
self.report_hour = report_hour
|
||
self.records = {} # fish_id -> metrics
|
||
self.last_report_date: date | None = None
|
||
|
||
def update(self, track_stats, ts: datetime):
|
||
"""用当前帧的跟踪统计更新档案。"""
|
||
for stats in track_stats:
|
||
fid = stats["id"]
|
||
rec = self.records.setdefault(fid, {
|
||
"frames": 0,
|
||
"distance": 0.0,
|
||
"bottom_frames": 0,
|
||
"slow_frames": 0,
|
||
"speed_samples": [],
|
||
"first_seen": ts,
|
||
"last_seen": ts,
|
||
})
|
||
rec["frames"] += 1
|
||
rec["distance"] += float(stats.get("distance_delta", 0) or 0)
|
||
if stats.get("is_bottom"):
|
||
rec["bottom_frames"] += 1
|
||
if stats.get("is_slow"):
|
||
rec["slow_frames"] += 1
|
||
speed = stats.get("speed")
|
||
if speed is not None:
|
||
rec["speed_samples"].append(float(speed))
|
||
rec["last_seen"] = ts
|
||
|
||
def _analyze_record(self, fid, rec):
|
||
frames = rec.get("frames", 1)
|
||
bottom_ratio = rec.get("bottom_frames", 0) / frames
|
||
slow_ratio = rec.get("slow_frames", 0) / frames
|
||
avg_speed = mean(rec["speed_samples"]) if rec["speed_samples"] else 0
|
||
distance = rec.get("distance", 0)
|
||
|
||
issues = []
|
||
if avg_speed < 10 or slow_ratio > 0.5:
|
||
issues.append("活动偏低")
|
||
if bottom_ratio > 0.4:
|
||
issues.append("长时间停留底部")
|
||
|
||
health = "疑似异常" if issues else "正常"
|
||
return {
|
||
"fish_id": fid,
|
||
"frames": frames,
|
||
"distance": distance,
|
||
"avg_speed": avg_speed,
|
||
"bottom_ratio": bottom_ratio,
|
||
"slow_ratio": slow_ratio,
|
||
"health": health,
|
||
"issues": issues,
|
||
"active_period": f"{rec['first_seen'].strftime('%H:%M:%S')} - {rec['last_seen'].strftime('%H:%M:%S')}",
|
||
}
|
||
|
||
def generate_report(self, report_date: date | None = None):
|
||
"""生成报告文本并写入文件，返回文件路径。"""
|
||
report_date = report_date or date.today()
|
||
analyses = [self._analyze_record(fid, rec) for fid, rec in self.records.items()]
|
||
analyses.sort(key=lambda x: x["fish_id"])
|
||
|
||
lines = [f"鱼类日度报告 - {report_date.strftime('%Y-%m-%d')}"]
|
||
if not analyses:
|
||
lines.append("今日无检测数据。")
|
||
else:
|
||
for item in analyses:
|
||
issues_text = "；".join(item["issues"]) if item["issues"] else "无"
|
||
lines.append(
|
||
f"- 鱼 {item['fish_id']}: 状态={item['health']}，平均速度={item['avg_speed']:.1f}px/s，"
|
||
f"底部占比={item['bottom_ratio']:.2f}，慢速占比={item['slow_ratio']:.2f}，"
|
||
f"累计位移={item['distance']:.1f}px，时间段={item['active_period']}，问题={issues_text}"
|
||
)
|
||
|
||
report_path = self.report_dir / f"fish_report_{report_date.strftime('%Y%m%d')}.txt"
|
||
report_path.write_text("\n".join(lines), encoding="utf-8")
|
||
self.last_report_date = report_date
|
||
# 每日报告后可选择清空记录，当前继续保留，次日将覆盖分析。
|
||
return report_path
|
||
|
||
def maybe_generate_report(self, now: datetime):
|
||
"""在到达指定时间且今日未生成报告时生成报告。"""
|
||
if now.hour >= self.report_hour:
|
||
today = now.date()
|
||
if self.last_report_date != today:
|
||
return self.generate_report(today)
|
||
return None
|
||
|