XW-AIOT/Security_AI_integrated
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复 ROI 区域内人员离开十几分钟未触发告警的问题。
Some checks failed
Python Test / test (push) Has been cancelled
Details

Browse Source 
1. 仍在 confirm_sec 滑动窗口内（未完成确认）
2. threshold_sec 阈值设置过长（需检查数据库实际配置值）
3. 新算法未被正确调用
This commit is contained in:
16337
2026-01-22 12:22:26 +08:00
parent cc4f33c0fd
commit 20f295a491
3 changed files with 31 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

97
inference/rules/algorithms.py
View File

@@ -13,18 +13,15 @@ from sort import Sort
class LeavePostAlgorithm:
STATE_INIT = "INIT_STATE"
STATE_ON_DUTY_CONFIRMING = "ON_DUTY_CONFIRMING"
STATE_ON_DUTY = "ON_DUTY"
STATE_OFF_DUTY_CONFIRMING = "OFF_DUTY_CONFIRMING"
STATE_OFF_DUTY_COUNTDOWN = "OFF_DUTY_COUNTDOWN"
STATE_NON_WORK_TIME = "NON_WORK_TIME"
def __init__(
self,
threshold_sec: int = 360,
confirm_sec: int = 30,
return_sec: int = 5,
threshold_sec: int = 300,
confirm_sec: int = 10,
return_sec: int = 30,
working_hours: Optional[List[Dict]] = None,
roi_polygon: Optional[List[Tuple[float, float]]] = None,
):
@@ -37,14 +34,11 @@ class LeavePostAlgorithm:
self.alert_cooldowns: Dict[str, datetime] = {}
self.cooldown_seconds = 300
self.state: str = self.STATE_INIT
self.state: str = self.STATE_ON_DUTY
self.state_start_time: Optional[datetime] = None
self.initial_state_start_time: Optional[datetime] = None
self.has_ever_seen_person: bool = False
self.on_duty_window = deque()
self.alarm_sent: bool = False
self.last_person_seen_time: Optional[datetime] = None
self.on_duty_start_time: Optional[datetime] = None
def is_in_working_hours(self, dt: Optional[datetime] = None) -> bool:
if not self.working_hours:
@@ -63,7 +57,7 @@ class LeavePostAlgorithm:
def is_point_in_roi(self, x: float, y: float) -> bool:
if not self.roi_polygon or len(self.roi_polygon) < 3:
return False
return True
from shapely.geometry import Point, Polygon
point = Point(x, y)
polygon = Polygon(self.roi_polygon)
@@ -96,87 +90,49 @@ class LeavePostAlgorithm:
self.state = self.STATE_NON_WORK_TIME
self.on_duty_start_time = None
self.last_person_seen_time = None
self.initial_state_start_time = None
self.has_ever_seen_person = False
self.on_duty_window.clear()
self.alarm_sent = False
roi_has_person = False
else:
if self.state == self.STATE_NON_WORK_TIME:
self.state = self.STATE_INIT
self.initial_state_start_time = current_time
self.has_ever_seen_person = False
self.state = self.STATE_ON_DUTY
self.on_duty_window.clear()
self.alarm_sent = False
if self.state == self.STATE_INIT:
self.initial_state_start_time = current_time
self.has_ever_seen_person = False
self.alarm_sent = False
if roi_has_person:
self.state = self.STATE_ON_DUTY_CONFIRMING
self.state_start_time = current_time
self.on_duty_window.clear()
self.on_duty_window.append((current_time, True))
self.has_ever_seen_person = True
else:
elapsed = (current_time - self.initial_state_start_time).total_seconds()
if elapsed >= self.threshold_sec:
self.state = self.STATE_OFF_DUTY_COUNTDOWN
self.state_start_time = current_time
self.alarm_sent = False
elif self.state == self.STATE_ON_DUTY_CONFIRMING:
if self.state == self.STATE_ON_DUTY:
self.on_duty_window.append((current_time, roi_has_person))
while self.on_duty_window and (current_time - self.on_duty_window[0][0]).total_seconds() > self.confirm_sec:
self.on_duty_window.popleft()
hit_ratio = sum(1 for t, detected in self.on_duty_window if detected) / max(len(self.on_duty_window), 1)
if hit_ratio >= 0.75 and (current_time - self.state_start_time).total_seconds() >= self.confirm_sec:
self.state = self.STATE_ON_DUTY
self.on_duty_start_time = None
self.last_person_seen_time = current_time
elif not roi_has_person and (current_time - self.state_start_time).total_seconds() > self.confirm_sec:
self.state = self.STATE_INIT
self.initial_state_start_time = current_time
elif self.state == self.STATE_ON_DUTY:
if roi_has_person:
self.last_person_seen_time = current_time
else:
self.state = self.STATE_OFF_DUTY_CONFIRMING
self.state_start_time = current_time
elif self.state == self.STATE_OFF_DUTY_CONFIRMING:
if roi_has_person:
self.state = self.STATE_ON_DUTY
self.state_start_time = current_time
self.last_person_seen_time = current_time
elif (current_time - self.state_start_time).total_seconds() >= self.return_sec:
if not roi_has_person and hit_ratio == 0:
self.state = self.STATE_OFF_DUTY_COUNTDOWN
self.state_start_time = current_time
self.alarm_sent = False
elif hit_ratio >= 0.75 and (current_time - self.on_duty_window[0][0]).total_seconds() >= self.confirm_sec:
self.last_person_seen_time = current_time
elif self.state == self.STATE_OFF_DUTY_COUNTDOWN:
elapsed = (current_time - self.state_start_time).total_seconds()
if roi_has_person:
self.state = self.STATE_ON_DUTY_CONFIRMING
self.state_start_time = current_time
self.state = self.STATE_ON_DUTY
self.on_duty_window.clear()
self.on_duty_window.append((current_time, True))
elif (current_time - self.state_start_time).total_seconds() >= self.threshold_sec:
self.alarm_sent = False
elif elapsed >= self.threshold_sec:
if not self.alarm_sent:
cooldown_key = f"{camera_id}"
if cooldown_key not in self.alert_cooldowns or (
current_time - self.alert_cooldowns[cooldown_key]
).total_seconds() > self.cooldown_seconds:
bbox = self.get_latest_bbox(tracks)
elapsed_minutes = int((current_time - self.state_start_time).total_seconds() / 60)
elapsed_minutes = int(elapsed / 60)
alerts.append({
"track_id": camera_id,
"bbox": bbox,
"off_duty_duration": (current_time - self.state_start_time).total_seconds(),
"off_duty_duration": elapsed,
"alert_type": "leave_post",
"message": f"离岗超过 {elapsed_minutes} 分钟",
})
@@ -199,21 +155,18 @@ class LeavePostAlgorithm:
return []
def reset(self):
self.state = self.STATE_INIT
self.state = self.STATE_ON_DUTY
self.state_start_time = None
self.initial_state_start_time = None
self.has_ever_seen_person = False
self.on_duty_window.clear()
self.alarm_sent = False
self.last_person_seen_time = None
self.on_duty_start_time = None
self.alert_cooldowns.clear()
def get_state(self, track_id: str) -> Optional[Dict[str, Any]]:
return {
"state": self.state,
"alarm_sent": self.alarm_sent,
"has_ever_seen_person": self.has_ever_seen_person,
"last_person_seen_time": self.last_person_seen_time,
}
@@ -293,9 +246,9 @@ class AlgorithmManager:
self.default_params = {
"leave_post": {
"threshold_sec": 360,
"confirm_sec": 30,
"return_sec": 5,
"threshold_sec": 300,
"confirm_sec": 10,
"return_sec": 30,
},
"intrusion": {
"check_interval_sec": 1.0,
@@ -322,9 +275,9 @@ class AlgorithmManager:
if algorithm_type == "leave_post":
self.algorithms[roi_id]["leave_post"] = LeavePostAlgorithm(
threshold_sec=algo_params.get("threshold_sec", 360),
confirm_sec=algo_params.get("confirm_sec", 30),
return_sec=algo_params.get("return_sec", 5),
threshold_sec=algo_params.get("threshold_sec", 300),
confirm_sec=algo_params.get("confirm_sec", 10),
return_sec=algo_params.get("return_sec", 30),
working_hours=self.working_hours,
)
elif algorithm_type == "intrusion":