alarm_plan.py 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515
  1. from typing import List, Tuple, Optional
  2. from datetime import datetime, time, date
  3. from threading import Thread, Lock
  4. from enum import Enum
  5. import uuid
  6. import json
  7. import traceback
  8. from datetime import datetime, timezone, timedelta
  9. from common.sys_comm import (
  10. LOGDBG, LOGINFO, LOGWARN, LOGERR,
  11. get_utc_time_ms, get_utc_time_s, get_bj_time_ms, get_bj_time_s,
  12. utc_to_bj_ms, bj_to_utc_ms, utc_to_bj_s, bj_to_utc_s
  13. )
  14. from core.time_plan import TimePlan
  15. from core.event_type import EventType, event_desc_map
  16. import core.g_LAS as g_las
  17. from device.dev_mng import (
  18. Device,
  19. dev_map_push, dev_map_pop, dev_map_find, dev_map_delete
  20. )
  21. from db.db_process import (
  22. db_req_que, DBRequest
  23. )
  24. import db.db_sqls as sqls
  25. import mqtt.mqtt_send as mqtt_send
  26. from device.dev_mng import g_dev_map, g_dev_map_lock
  27. class EventAttr_Base:
  28. def __init__(self):
  29. return
  30. # 事件属性 事件事件
  31. class EventAttr_StayDetection(EventAttr_Base):
  32. def __init__(self):
  33. self.enter_ts_: int = -1 # 进入时间(s)
  34. self.leave_ts_: int = -1 # 离开时间(s)
  35. self.stay_time_: int = -1 # 停留时长(s)
  36. return
  37. def reset(self):
  38. self.enter_ts_ = -1
  39. self.leave_ts_ = -1
  40. self.stay_time_ = -1
  41. # 事件属性 滞留事件
  42. class EventAttr_RetentionDetection(EventAttr_Base):
  43. def __init__(self):
  44. self.enter_ts_: int = -1 # 进入时间(s)
  45. self.leave_ts_: int = -1 # 离开时间(s)
  46. self.stay_time_: int = -1 # 停留时长(s)
  47. return
  48. def reset(self):
  49. self.enter_ts_ = -1
  50. self.leave_ts_ = -1
  51. self.stay_time_ = -1
  52. # 事件属性 如厕事件
  53. class EventAttr_ToiletingDetection(EventAttr_Base):
  54. def __init__(self):
  55. self.enter_ts_: int = -1 # 进入时间(ms)
  56. self.leave_ts_: int = -1 # 离开时间(ms)
  57. self.stay_time_: int = -1 # 停留时长(ms)
  58. return
  59. def reset(self):
  60. self.enter_ts_ = -1
  61. self.leave_ts_ = -1
  62. self.stay_time_ = -1
  63. # 事件属性 如厕频次统计
  64. class EventAttr_ToiletingFrequency(EventAttr_Base):
  65. def __init__(self):
  66. self.enter_ts_: int = -1 # 进入时间(ms)
  67. self.leave_ts_: int = -1 # 离开时间(ms)
  68. self.stay_time_: int = -1 # 停留时长(ms)
  69. return
  70. def reset(self):
  71. self.enter_ts_ = -1
  72. self.leave_ts_ = -1
  73. self.stay_time_ = -1
  74. class AlarmPlan:
  75. def __init__(self,
  76. plan_uuid: str,
  77. name: str,
  78. dev_id: str,
  79. enable: bool,
  80. time_plan: TimePlan,
  81. rect: list,
  82. event_type: int,
  83. threshold_time: int,
  84. merge_time: int
  85. ):
  86. self.lock_ = Lock()
  87. self.plan_uuid_ = plan_uuid # 计划id
  88. self.name_ = name # 计划名称
  89. self.dev_id_ = dev_id # 设备id
  90. self.enable_ = enable # 是否启用
  91. self.time_plan_ = time_plan # 时间计划
  92. # 维护状态(根据TimePlanu判断)
  93. self.status_ = 0 # 0未激活,1激活,-1过期
  94. self.status_update_ts_ = -1 # 状态更新时间,初始值为-1
  95. # 事件属性表
  96. self.event_attr_map = {
  97. EventType.STAY_DETECTION.value: EventAttr_StayDetection,
  98. EventType.RETENTION_DETECTION.value: EventAttr_RetentionDetection,
  99. EventType.TOILETING_DETECTION.value: EventAttr_ToiletingDetection,
  100. EventType.TOILETING_FREQUENCY.value: EventAttr_ToiletingFrequency
  101. }
  102. # 事件触发参数
  103. self.rect_ = rect # 检测区域 [left, top, width, height]
  104. self.threshold_time_ = threshold_time # 触发时间阈值
  105. self.merge_time_ = merge_time # 归并时间窗口
  106. self.event_type_ = event_type # 事件类型
  107. self.event_attr_ = self.init_event_attr() # 事件属性
  108. if self.event_attr_ is None:
  109. raise ValueError(f"Invalid event_type: {event_type}")
  110. def execute(self):
  111. if self.status_ != 1:
  112. return
  113. g_las.g_event_dispatcher.dispatch(self.event_type_, self)
  114. # 更新激活状态
  115. def update_status(self, now: Optional[datetime] = None) -> None:
  116. now = now or datetime.now()
  117. old_status = self.status_
  118. if not self.enable_:
  119. self.status_ = 0
  120. else:
  121. now_fmt = now.strftime("%Y-%m-%d")
  122. # 过期
  123. if now_fmt > self.time_plan_.stop_date_:
  124. self.status_ = -1
  125. elif now_fmt < self.time_plan_.start_date_:
  126. self.status_ = 0
  127. elif self.time_plan_.is_active_now(now):
  128. self.status_ = 1
  129. else:
  130. self.status_ = 0
  131. if self.status_ != old_status:
  132. self.status_update_ts = int(now.timestamp())
  133. LOGINFO(f"[Status] plan {self.plan_uuid_} status_ changed {old_status} -> {self.status_}")
  134. def is_point_in_rect(self, x:float, y:float, rect:list) -> bool:
  135. rx, ry, rw, rh = rect
  136. x_min = min(rx, rx + rw)
  137. x_max = max(rx, rx + rw)
  138. y_min = min(ry, ry - rh)
  139. y_max = max(ry, ry - rh)
  140. bRet: bool = x_min <= x <= x_max and y_min <= y <= y_max
  141. return bRet
  142. # 查找最近 t 秒内,最后一个落在 rect 内的 target_point 的 rtd_unit
  143. def find_latest_rtd_in_region(self, device: Device, rect: list, now: int=None, t: int=1):
  144. now_s = now if now else get_utc_time_s()
  145. rtd_que_copy = device.get_rtd_que_copy()
  146. with self.lock_:
  147. for rtd_unit in reversed(rtd_que_copy): # 倒序扫描
  148. ts_s = int(rtd_unit.get("timestamp", 0))
  149. if now_s - ts_s > t:
  150. break # 已经超过 t 秒,可以直接结束
  151. # 检查点是否在区域内
  152. for pt in rtd_unit.get("target_point", []):
  153. if len(pt) >= 2:
  154. x, y = pt[0], pt[1]
  155. if self.is_point_in_rect(x, y, rect):
  156. return rtd_unit
  157. return None
  158. # 初始化事件属性
  159. def init_event_attr(self):
  160. event_cls = self.event_attr_map.get(self.event_type_)
  161. if event_cls is None:
  162. return None
  163. event_attr = event_cls()
  164. return event_attr
  165. # 停留事件
  166. def handle_stay_detection(self):
  167. try:
  168. dev_id = self.dev_id_
  169. device:Device = dev_map_find(dev_id)
  170. if not device:
  171. return
  172. now = get_utc_time_s()
  173. # 查找最新的落在检测区域的目标
  174. rtd_unit = self.find_latest_rtd_in_region(device, self.rect_, now, 3)
  175. if rtd_unit:
  176. timestamp = rtd_unit["timestamp"]
  177. pose = rtd_unit["pose"]
  178. target_point = rtd_unit["target_point"]
  179. if self.event_attr_.enter_ts_ == -1:
  180. self.event_attr_.enter_ts_ = timestamp
  181. else:
  182. self.event_attr_.leave_ts_ = timestamp
  183. if self.event_attr_.enter_ts_ == -1 or self.event_attr_.leave_ts_ == -1:
  184. return
  185. # 归并时间内,不认为事件结束
  186. if now - self.event_attr_.leave_ts_ < self.merge_time_:
  187. return
  188. self.event_attr_.stay_time_ = self.event_attr_.leave_ts_ - self.event_attr_.enter_ts_
  189. stay_time =self.event_attr_.stay_time_
  190. # 时间小于触发时间阈值,忽略并重置
  191. if stay_time < self.threshold_time_ :
  192. self.event_attr_.reset()
  193. LOGINFO(f"less than threshold_time, alarm_plan: {self.plan_uuid_}, event_type: {self.event_type_}")
  194. return
  195. # 构造事件
  196. # 入库
  197. info = {
  198. "enter_time": utc_to_bj_s(self.event_attr_.enter_ts_),
  199. "leave_time": utc_to_bj_s(self.event_attr_.enter_ts_),
  200. "stay_time": stay_time
  201. }
  202. event_uuid = str(uuid.uuid4())
  203. params = {
  204. "dev_id": dev_id,
  205. "uuid": event_uuid,
  206. "plan_uuid": self.plan_uuid_,
  207. "event_type": event_desc_map[self.event_type_],
  208. "info": json.dumps(info),
  209. "is_handle": 0,
  210. "is_deleted": 0
  211. }
  212. db_req_que.put(DBRequest(sql=sqls.sql_insert_events, params=params, callback=None))
  213. # 通知
  214. mqtt_send.alarm_event(dev_id, event_uuid, self.plan_uuid_, event_desc_map[self.event_type_], info, "events")
  215. LOGDBG(f"new event: {event_desc_map[self.event_type_]}, stay_time: {stay_time}")
  216. self.event_attr_.reset()
  217. except json.JSONDecodeError as e:
  218. tb_info = traceback.extract_tb(e.__traceback__)
  219. for frame in tb_info:
  220. LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
  221. except Exception as e:
  222. tb_info = traceback.extract_tb(e.__traceback__)
  223. for frame in tb_info:
  224. LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
  225. # 滞留事件
  226. def handle_retention_detection(self):
  227. try:
  228. dev_id = self.dev_id_
  229. device:Device = dev_map_find(dev_id)
  230. if not device:
  231. return
  232. now = get_utc_time_s()
  233. # 查找最新的落在检测区域的目标
  234. rtd_unit = self.find_latest_rtd_in_region(device, self.rect_, now, 3)
  235. if rtd_unit:
  236. timestamp = rtd_unit["timestamp"]
  237. pose = rtd_unit["pose"]
  238. target_point = rtd_unit["target_point"]
  239. if self.event_attr_.enter_ts_ == -1:
  240. self.event_attr_.enter_ts_ = timestamp
  241. else:
  242. self.event_attr_.leave_ts_ = timestamp
  243. if self.event_attr_.enter_ts_ == -1 or self.event_attr_.leave_ts_ == -1:
  244. return
  245. # 归并时间内,不认为事件结束
  246. if now - self.event_attr_.leave_ts_ < self.merge_time_:
  247. return
  248. self.event_attr_.stay_time_ = self.event_attr_.leave_ts_ - self.event_attr_.enter_ts_
  249. stay_time =self.event_attr_.stay_time_
  250. # 时间小于触发时间阈值,忽略并重置
  251. if stay_time < self.threshold_time_ :
  252. self.event_attr_.reset()
  253. LOGINFO(f"less than threshold_time, alarm_plan: {self.plan_uuid_}, event_type: {self.event_type_}")
  254. return
  255. # 构造事件
  256. # 入库
  257. info = {
  258. "enter_time": utc_to_bj_s(self.event_attr_.enter_ts_),
  259. "leave_time": utc_to_bj_s(self.event_attr_.enter_ts_),
  260. "stay_time": stay_time
  261. }
  262. event_uuid = str(uuid.uuid4())
  263. params = {
  264. "dev_id": dev_id,
  265. "uuid": event_uuid,
  266. "plan_uuid": self.plan_uuid_,
  267. "event_type": event_desc_map[self.event_type_],
  268. "info": json.dumps(info),
  269. "is_handle": 0,
  270. "is_deleted": 0
  271. }
  272. db_req_que.put(DBRequest(sql=sqls.sql_insert_events, params=params, callback=None))
  273. # 通知
  274. mqtt_send.alarm_event(dev_id, event_uuid, self.plan_uuid_, event_desc_map[self.event_type_], info, "events")
  275. LOGDBG(f"new event: {event_desc_map[self.event_type_]}, stay_time: {stay_time}")
  276. self.event_attr_.reset()
  277. except json.JSONDecodeError as e:
  278. tb_info = traceback.extract_tb(e.__traceback__)
  279. for frame in tb_info:
  280. LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
  281. except Exception as e:
  282. tb_info = traceback.extract_tb(e.__traceback__)
  283. for frame in tb_info:
  284. LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
  285. # 如厕事件
  286. def handle_toileting_detection(self):
  287. try:
  288. dev_id = self.dev_id_
  289. device:Device = dev_map_find(dev_id)
  290. if not device:
  291. return
  292. now = get_utc_time_s()
  293. # 查找最新的落在检测区域的目标
  294. rtd_unit = self.find_latest_rtd_in_region(device, self.rect_, now, 3)
  295. if rtd_unit:
  296. timestamp = rtd_unit["timestamp"]
  297. pose = rtd_unit["pose"]
  298. target_point = rtd_unit["target_point"]
  299. if self.event_attr_.enter_ts_ == -1:
  300. self.event_attr_.enter_ts_ = timestamp
  301. else:
  302. self.event_attr_.leave_ts_ = timestamp
  303. if self.event_attr_.enter_ts_ == -1 or self.event_attr_.leave_ts_ == -1:
  304. return
  305. # 归并时间内,不认为事件结束
  306. if now - self.event_attr_.leave_ts_ < self.merge_time_:
  307. return
  308. self.event_attr_.stay_time_ = self.event_attr_.leave_ts_ - self.event_attr_.enter_ts_
  309. stay_time =self.event_attr_.stay_time_
  310. # 时间小于触发时间阈值,忽略并重置
  311. if stay_time < self.threshold_time_ :
  312. self.event_attr_.reset()
  313. LOGINFO(f"less than threshold_time, alarm_plan: {self.plan_uuid_}, event_type: {self.event_type_}")
  314. return
  315. # 构造事件
  316. # 入库
  317. info = {
  318. "enter_time": utc_to_bj_s(self.event_attr_.enter_ts_),
  319. "leave_time": utc_to_bj_s(self.event_attr_.enter_ts_),
  320. "stay_time": stay_time
  321. }
  322. event_uuid = str(uuid.uuid4())
  323. params = {
  324. "dev_id": dev_id,
  325. "uuid": event_uuid,
  326. "plan_uuid": self.plan_uuid_,
  327. "event_type": event_desc_map[self.event_type_],
  328. "info": json.dumps(info),
  329. "is_handle": 0,
  330. "is_deleted": 0
  331. }
  332. db_req_que.put(DBRequest(sql=sqls.sql_insert_events, params=params, callback=None))
  333. # 通知
  334. mqtt_send.alarm_event(dev_id, event_uuid, self.plan_uuid_, event_desc_map[self.event_type_], info, "events")
  335. LOGDBG(f"new event: {event_desc_map[self.event_type_]}, stay_time: {stay_time}")
  336. self.event_attr_.reset()
  337. except json.JSONDecodeError as e:
  338. tb_info = traceback.extract_tb(e.__traceback__)
  339. for frame in tb_info:
  340. LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
  341. except Exception as e:
  342. tb_info = traceback.extract_tb(e.__traceback__)
  343. for frame in tb_info:
  344. LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
  345. # 如厕频次统计
  346. def handle_toileting_frequency(self):
  347. try:
  348. dev_id = self.dev_id_
  349. device:Device = dev_map_find(dev_id)
  350. if not device:
  351. return
  352. now = get_utc_time_s()
  353. # 查找最新的落在检测区域的目标
  354. rtd_unit = self.find_latest_rtd_in_region(device, self.rect_, now, 3)
  355. if rtd_unit:
  356. timestamp = rtd_unit["timestamp"]
  357. pose = rtd_unit["pose"]
  358. target_point = rtd_unit["target_point"]
  359. if self.event_attr_.enter_ts_ == -1:
  360. self.event_attr_.enter_ts_ = timestamp
  361. else:
  362. self.event_attr_.leave_ts_ = timestamp
  363. if self.event_attr_.enter_ts_ == -1 or self.event_attr_.leave_ts_ == -1:
  364. return
  365. # 归并时间内,不认为事件结束
  366. if now - self.event_attr_.leave_ts_ < self.merge_time_:
  367. return
  368. self.event_attr_.stay_time_ = self.event_attr_.leave_ts_ - self.event_attr_.enter_ts_
  369. stay_time =self.event_attr_.stay_time_
  370. # 时间小于触发时间阈值,忽略并重置
  371. if stay_time < self.threshold_time_ :
  372. self.event_attr_.reset()
  373. LOGINFO(f"less than threshold_time, alarm_plan: {self.plan_uuid_}, event_type: {self.event_type_}")
  374. return
  375. # 构造事件
  376. # 入库
  377. info = {
  378. "enter_time": utc_to_bj_s(self.event_attr_.enter_ts_),
  379. "leave_time": utc_to_bj_s(self.event_attr_.enter_ts_),
  380. "stay_time": stay_time
  381. }
  382. event_uuid = str(uuid.uuid4())
  383. params = {
  384. "dev_id": dev_id,
  385. "uuid": event_uuid,
  386. "plan_uuid": self.plan_uuid_,
  387. "event_type": event_desc_map[self.event_type_],
  388. "info": json.dumps(info),
  389. "is_handle": 0,
  390. "is_deleted": 0
  391. }
  392. db_req_que.put(DBRequest(sql=sqls.sql_insert_events, params=params, callback=None))
  393. # 通知
  394. mqtt_send.alarm_event(dev_id, event_uuid, self.plan_uuid_, event_desc_map[self.event_type_], info, "events")
  395. LOGDBG(f"new event: {event_desc_map[self.event_type_]}, stay_time: {stay_time}")
  396. self.event_attr_.reset()
  397. except json.JSONDecodeError as e:
  398. tb_info = traceback.extract_tb(e.__traceback__)
  399. for frame in tb_info:
  400. LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
  401. except Exception as e:
  402. tb_info = traceback.extract_tb(e.__traceback__)
  403. for frame in tb_info:
  404. LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
  405. # 夜间如厕频次统计
  406. def handle_night_toileting_frequency(self):
  407. return
  408. # 如厕频次异常
  409. def handle_toileting_frequency_abnormal(self):
  410. return
  411. # 起夜异常
  412. def handle_night_toileting_frequency_abnormal(self):
  413. return
  414. # 卫生间频次统计
  415. def handle_bathroom_stay_frequency(self):
  416. return
  417. # 异常消失
  418. def handle_target_missing(self):
  419. return