12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103 |
- from typing import List, Tuple, Optional
- from datetime import datetime, time, date
- from threading import Thread, Lock
- from enum import Enum
- import uuid
- import json
- import traceback
- from datetime import datetime, timezone, timedelta
- from common.sys_comm import (
- LOGDBG, LOGINFO, LOGWARN, LOGERR,
- get_utc_time_ms, get_utc_time_s, get_bj_time_ms, get_bj_time_s,
- utc_to_bj_ms, bj_to_utc_ms, utc_to_bj_s, bj_to_utc_s
- )
- from core.time_plan import TimePlan
- from core.event_type import EventType, event_desc_map
- import core.g_LAS as g_las
- import core.alarm_plan_helper as helper
- from device.dev_mng import (
- Device,
- dev_map_push, dev_map_pop, dev_map_find, dev_map_delete
- )
- from db.db_process import (
- db_req_que, DBRequest
- )
- import db.db_sqls as sqls
- import mqtt.mqtt_send as mqtt_send
- from device.dev_mng import g_dev_map, g_dev_map_lock
- class EventAttr_Base:
- def __init__(self):
- return
- # 事件属性 事件事件
- class EventAttr_StayDetection(EventAttr_Base):
- def __init__(self, event_type):
- self.event_type_ = event_type
- self.enter_ts_: int = -1 # 进入时间(s)
- self.leave_ts_: int = -1 # 离开时间(s)
- self.stay_time_: int = -1 # 停留时长(s)
- return
- def reset(self):
- self.enter_ts_ = -1
- self.leave_ts_ = -1
- self.stay_time_ = -1
- # 事件属性 滞留事件
- class EventAttr_RetentionDetection(EventAttr_Base):
- def __init__(self, event_type):
- self.event_type_ = event_type
- self.enter_ts_: int = -1 # 进入时间(s)
- self.leave_ts_: int = -1 # 离开时间(s)
- self.stay_time_: int = -1 # 停留时长(s)
- return
- def reset(self):
- self.enter_ts_ = -1
- self.leave_ts_ = -1
- self.stay_time_ = -1
- # 事件属性 如厕事件
- class EventAttr_ToiletingDetection(EventAttr_Base):
- def __init__(self, event_type):
- self.event_type_ = event_type
- self.enter_ts_: int = -1 # 进入时间(ms)
- self.leave_ts_: int = -1 # 离开时间(ms)
- self.stay_time_: int = -1 # 停留时长(ms)
- return
- def reset(self):
- self.enter_ts_ = -1
- self.leave_ts_ = -1
- self.stay_time_ = -1
- # 事件属性 如厕频次统计
- class EventAttr_ToiletingFrequency(EventAttr_Base):
- def __init__(self, event_type):
- self.event_type_ = event_type
- self.count_: int = 0 # 统计次数
- self.event_list: list = []
- return
- def reset(self):
- self.count_ = 0
- self.event_list = []
- # 事件属性 夜间如厕频次统计
- class EventAttr_NightToiletingFrequency(EventAttr_Base):
- def __init__(self, event_type):
- self.event_type_ = event_type
- self.count_: int = 0 # 统计次数
- self.event_list: list = []
- return
- def reset(self):
- self.count_ = 0
- self.event_list = []
- # 事件属性 如厕频次异常
- class EventAttr_ToiletingFrequencyAbnormal(EventAttr_Base):
- def __init__(self, event_type):
- self.event_type_ = event_type
- self.count_: int = 0 # 统计次数
- self.threshold_count_: int = 0 # 异常阈值
- self.event_list: list = []
- return
- def reset(self):
- self.count_ = 0
- self.event_list = []
- # 事件属性 起夜异常
- class EventAttr_NightToiletingFrequencyAbnormal(EventAttr_Base):
- def __init__(self, event_type):
- self.event_type_ = event_type
- self.count_: int = 0 # 统计次数
- self.threshold_count_: int = 0 # 异常阈值
- self.event_list: list = []
- return
- def reset(self):
- self.count_ = 0
- self.event_list = []
- # 事件属性 卫生间频次统计
- class EventAttr_BathroomStayFrequency(EventAttr_Base):
- def __init__(self, event_type):
- self.event_type_ = event_type
- self.count_: int = 0 # 统计次数
- self.event_list: list = []
- return
- def reset(self):
- self.count_ = 0
- self.event_list = []
- # 事件属性 异常消失
- class EventAttr_TargetAbsence(EventAttr_Base):
- def __init__(self, event_type):
- self.event_type_ = event_type
- self.leave_ts_: int = -1 # 离开时间(ms)
- self.enter_ts_: int = -1 # 进入时间(ms)
- self.absence_time_: int = -1 # 消失时长(ms)
- self.time_threshold_: int = 300 # 触发消失时间阈值(ms)
- return
- def reset(self):
- self.leave_ts_ = -1
- self.enter_ts_ = -1
- self.absence_time_ = -1
- # 事件属性表
- event_attr_map = {
- EventType.STAY_DETECTION.value : EventAttr_StayDetection,
- EventType.RETENTION_DETECTION.value : EventAttr_RetentionDetection,
- EventType.TOILETING_DETECTION.value : EventAttr_ToiletingDetection,
- EventType.TOILETING_FREQUENCY.value : EventAttr_ToiletingFrequency,
- EventType.NIGHT_TOILETING_FREQUENCY.value : EventAttr_NightToiletingFrequency,
- EventType.TOILETING_FREQUENCY_ABNORMAL.value : EventAttr_ToiletingFrequencyAbnormal,
- EventType.NIGHT_TOILETING_FREQUENCY_ABNORMAL.value: EventAttr_NightToiletingFrequencyAbnormal,
- EventType.BATHROOM_STAY_FREQUENCY.value : EventAttr_BathroomStayFrequency,
- EventType.TARGET_ABSENCE.value : EventAttr_TargetAbsence,
- }
- class Cron:
- def __init__(self, h, m, s):
- self.h_ = None
- self.m_ = None
- self.s_ = None
- class AlarmPlan:
- def __init__(self,
- plan_uuid: str,
- name: str,
- dev_id: str,
- dev_name: str,
- enable: bool,
- time_plan: TimePlan,
- rect: list,
- event_type: int,
- threshold_time: int,
- merge_time: int,
- param: dict,
- cron: Optional[dict] = None
- ):
- self.lock_ = Lock()
- self.plan_uuid_ = plan_uuid # 计划id
- self.name_ = name # 计划名称
- self.dev_id_ = dev_id # 设备id
- self.dev_name_ = dev_name # 设备名称
- self.enable_ = enable # 是否启用
- self.time_plan_ = time_plan # 时间计划
- self.param_ = param # 参数
- # 维护状态(根据TimePlanu判断)
- self.status_ = 0 # 0未激活,1激活,-1过期
- self.status_update_ts_ = -1 # 状态更新时间,初始值为-1
- # 事件触发参数
- self.rect_ = rect # 检测区域 [left, top, width, height]
- self.threshold_time_ = threshold_time # 触发时间阈值
- self.merge_time_ = merge_time # 归并时间窗口
- self.event_type_ = event_type # 事件类型
- self.event_attr_ = self.init_event_attr() # 事件属性
- if self.event_attr_ is None:
- raise ValueError(f"Invalid event_type: {event_type}")
- self.cron_ = cron # {“hour": 6, "minute": 0}
- def execute(self):
- if self.status_ != 1:
- return
- g_las.g_alarm_plan_disp.dispatch(self.event_type_, self)
- # 初始化事件属性
- def init_event_attr(self):
- try:
- event_cls = event_attr_map.get(self.event_type_)
- if event_cls is None:
- return None
- event_attr = event_cls(self.event_type_)
- if ((self.event_type_ == EventType.TOILETING_FREQUENCY_ABNORMAL.value) or
- (self.event_type_ == EventType.NIGHT_TOILETING_FREQUENCY_ABNORMAL.value)):
- event_attr.threshold_count_ = int(self.param_.get("count", 0))
- if ((self.event_type_ == EventType.TARGET_ABSENCE.value)):
- event_attr.time_threshold_ = int(self.param_.get("time_threshold", 0))
- return event_attr
- except json.JSONDecodeError as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
- except Exception as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
- # 更新激活状态
- def update_status(self, now: Optional[datetime] = None) -> None:
- now = now or datetime.now()
- old_status = self.status_
- if not self.enable_:
- self.status_ = 0
- else:
- now_fmt = now.strftime("%Y-%m-%d")
- # 过期
- if now_fmt > self.time_plan_.stop_date_:
- self.status_ = -1
- elif now_fmt < self.time_plan_.start_date_:
- self.status_ = 0
- elif self.time_plan_.is_active_now(now):
- self.status_ = 1
- else:
- self.status_ = 0
- if self.status_ != old_status:
- self.status_update_ts = int(now.timestamp())
- LOGINFO(f"[Status] plan {self.plan_uuid_} status_ changed {old_status} -> {self.status_}")
- def is_point_in_rect(self, x:float, y:float, rect:list) -> bool:
- rx, ry, rw, rh = rect
- x_min = min(rx, rx + rw)
- x_max = max(rx, rx + rw)
- y_min = min(ry, ry - rh)
- y_max = max(ry, ry - rh)
- bRet: bool = x_min <= x <= x_max and y_min <= y <= y_max
- return bRet
- # 查找最近 t 秒内,最后一个落在 rect 内的 target_point 的 rtd_unit
- def find_latest_rtd_in_region(self, device: Device, rect: list, now: int=None, t: int=1):
- now_s = now if now else get_utc_time_s()
- rtd_que_copy = device.get_rtd_que_copy()
- with self.lock_:
- for rtd_unit in reversed(rtd_que_copy): # 倒序扫描
- ts_s = int(rtd_unit.get("timestamp", 0))
- if now_s - ts_s > t:
- break # 已经超过 t 秒,可以直接结束
- # 检查点是否在区域内
- for pt in rtd_unit.get("target_point", []):
- if len(pt) >= 2:
- x, y = pt[0], pt[1]
- if self.is_point_in_rect(x, y, rect):
- return rtd_unit
- return None
- # 停留事件
- def handle_stay_detection(self):
- try:
- dev_id = self.dev_id_
- device:Device = dev_map_find(dev_id)
- if not device:
- return
- now = get_utc_time_s()
- # 查找最新的落在检测区域的目标
- rtd_unit = self.find_latest_rtd_in_region(device, self.rect_, now, 3)
- if rtd_unit:
- timestamp = rtd_unit["timestamp"]
- pose = rtd_unit["pose"]
- target_point = rtd_unit["target_point"]
- if self.event_attr_.enter_ts_ == -1:
- self.event_attr_.enter_ts_ = timestamp
- else:
- self.event_attr_.leave_ts_ = timestamp
- if self.event_attr_.enter_ts_ == -1 or self.event_attr_.leave_ts_ == -1:
- return
- # 归并时间内,不认为事件结束
- if now - self.event_attr_.leave_ts_ < self.merge_time_:
- return
- self.event_attr_.stay_time_ = self.event_attr_.leave_ts_ - self.event_attr_.enter_ts_
- stay_time =self.event_attr_.stay_time_
- # 时间小于触发时间阈值,忽略并重置
- if stay_time < self.threshold_time_ :
- self.event_attr_.reset()
- LOGINFO(f"less than threshold_time, alarm_plan: {self.plan_uuid_}, event_type: {self.event_type_}")
- return
- # 构造事件
- # 入库
- info = {
- "start_time": utc_to_bj_s(self.event_attr_.enter_ts_),
- "end_time": utc_to_bj_s(self.event_attr_.leave_ts_),
- "stay_time": stay_time
- }
- event_uuid = str(uuid.uuid4())
- params = {
- "dev_id": dev_id,
- "uuid": event_uuid,
- "plan_uuid": self.plan_uuid_,
- "event_type": event_desc_map[self.event_type_],
- "info": json.dumps(info),
- "is_handle": 0,
- "create_time": get_bj_time_s(),
- "is_deleted": 0,
- "remark": json.dumps({}, ensure_ascii=False)
- }
- db_req_que.put(DBRequest(sql=sqls.sql_insert_events, params=params, callback=None))
- # 通知
- mqtt_send.alarm_event(dev_id, self.dev_name_, event_uuid, self.plan_uuid_, event_desc_map[self.event_type_], info, "events")
- LOGDBG(f"new event: {event_desc_map[self.event_type_]}, stay_time: {stay_time}")
- self.event_attr_.reset()
- except json.JSONDecodeError as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
- except Exception as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
- # 滞留事件
- def handle_retention_detection(self):
- try:
- dev_id = self.dev_id_
- device:Device = dev_map_find(dev_id)
- if not device:
- return
- now = get_utc_time_s()
- # 查找最新的落在检测区域的目标
- rtd_unit = self.find_latest_rtd_in_region(device, self.rect_, now, 3)
- if rtd_unit:
- timestamp = rtd_unit["timestamp"]
- pose = rtd_unit["pose"]
- target_point = rtd_unit["target_point"]
- if self.event_attr_.enter_ts_ == -1:
- self.event_attr_.enter_ts_ = timestamp
- else:
- self.event_attr_.leave_ts_ = timestamp
- if self.event_attr_.enter_ts_ == -1 or self.event_attr_.leave_ts_ == -1:
- return
- # 归并时间内,不认为事件结束
- if now - self.event_attr_.leave_ts_ < self.merge_time_:
- return
- self.event_attr_.stay_time_ = self.event_attr_.leave_ts_ - self.event_attr_.enter_ts_
- stay_time =self.event_attr_.stay_time_
- # 时间小于触发时间阈值,忽略并重置
- if stay_time < self.threshold_time_ :
- self.event_attr_.reset()
- LOGINFO(f"less than threshold_time, alarm_plan: {self.plan_uuid_}, event_type: {self.event_type_}")
- return
- # 构造事件
- # 入库
- info = {
- "start_time": utc_to_bj_s(self.event_attr_.enter_ts_),
- "end_time": utc_to_bj_s(self.event_attr_.leave_ts_),
- "stay_time": stay_time
- }
- event_uuid = str(uuid.uuid4())
- params = {
- "dev_id": dev_id,
- "uuid": event_uuid,
- "plan_uuid": self.plan_uuid_,
- "event_type": event_desc_map[self.event_type_],
- "info": json.dumps(info),
- "is_handle": 0,
- "create_time": get_bj_time_s(),
- "is_deleted": 0,
- "remark": json.dumps({}, ensure_ascii=False)
- }
- db_req_que.put(DBRequest(sql=sqls.sql_insert_events, params=params, callback=None))
- # 通知
- mqtt_send.alarm_event(dev_id, self.dev_name_, event_uuid, self.plan_uuid_, event_desc_map[self.event_type_], info, "events")
- LOGDBG(f"new event: {event_desc_map[self.event_type_]}, stay_time: {stay_time}")
- self.event_attr_.reset()
- except json.JSONDecodeError as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
- except Exception as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
- # 如厕事件
- def handle_toileting_detection(self):
- try:
- dev_id = self.dev_id_
- device:Device = dev_map_find(dev_id)
- if not device:
- return
- now = get_utc_time_s()
- # 查找最新的落在检测区域的目标
- rtd_unit = self.find_latest_rtd_in_region(device, self.rect_, now, 3)
- if rtd_unit:
- timestamp = rtd_unit["timestamp"]
- pose = rtd_unit["pose"]
- target_point = rtd_unit["target_point"]
- if self.event_attr_.enter_ts_ == -1:
- self.event_attr_.enter_ts_ = timestamp
- else:
- self.event_attr_.leave_ts_ = timestamp
- if self.event_attr_.enter_ts_ == -1 or self.event_attr_.leave_ts_ == -1:
- return
- # 归并时间内,不认为事件结束
- if now - self.event_attr_.leave_ts_ < self.merge_time_:
- return
- self.event_attr_.stay_time_ = self.event_attr_.leave_ts_ - self.event_attr_.enter_ts_
- stay_time =self.event_attr_.stay_time_
- # 时间小于触发时间阈值,忽略并重置
- if stay_time < self.threshold_time_ :
- self.event_attr_.reset()
- LOGINFO(f"less than threshold_time, alarm_plan: {self.plan_uuid_}, event_type: {self.event_type_}")
- return
- # 构造事件
- # 入库
- info = {
- "start_time": utc_to_bj_s(self.event_attr_.enter_ts_),
- "end_time": utc_to_bj_s(self.event_attr_.leave_ts_),
- "stay_time": stay_time
- }
- event_uuid = str(uuid.uuid4())
- params = {
- "dev_id": dev_id,
- "uuid": event_uuid,
- "plan_uuid": self.plan_uuid_,
- "event_type": event_desc_map[self.event_type_],
- "info": json.dumps(info),
- "is_handle": 0,
- "create_time": get_bj_time_s(),
- "is_deleted": 0,
- "remark": json.dumps({}, ensure_ascii=False)
- }
- db_req_que.put(DBRequest(sql=sqls.sql_insert_events, params=params, callback=None))
- # 通知
- mqtt_send.alarm_event(dev_id, self.dev_name_, event_uuid, self.plan_uuid_, event_desc_map[self.event_type_], info, "events")
- LOGDBG(f"new event: {event_desc_map[self.event_type_]}, stay_time: {stay_time}")
- self.event_attr_.reset()
- except json.JSONDecodeError as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
- except Exception as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
- # 如厕频次统计
- def handle_toileting_frequency(self):
- try:
- dev_id = self.dev_id_
- device:Device = dev_map_find(dev_id)
- if not device:
- return
- start_dt, end_dt = helper.get_query_time_range(self.param_)
- params = {
- "dev_id" : self.dev_id_,
- "event_type": event_desc_map[EventType.TOILETING_DETECTION.value],
- "start_dt": start_dt,
- "end_dt": end_dt
- }
- userdata = {
- "start_dt" : start_dt,
- "end_dt" : end_dt
- }
- db_req_que.put(DBRequest(
- sql=sqls.sql_query_events_by_datetime,
- params=params,
- callback=self.cb_toileting_frequency,
- userdata=userdata))
- except json.JSONDecodeError as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
- except Exception as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
- # 夜间如厕频次统计
- def handle_night_toileting_frequency(self):
- try:
- dev_id = self.dev_id_
- device:Device = dev_map_find(dev_id)
- if not device:
- return
- start_dt, end_dt = helper.get_query_time_range(self.param_)
- params = {
- "dev_id" : self.dev_id_,
- "event_type": event_desc_map[EventType.TOILETING_DETECTION.value],
- "start_dt": start_dt,
- "end_dt": end_dt
- }
- userdata = {
- "start_dt" : start_dt,
- "end_dt" : end_dt
- }
- db_req_que.put(DBRequest(
- sql=sqls.sql_query_events_by_datetime,
- params=params,
- callback=self.cb_night_toileting_frequency,
- userdata=userdata))
- except json.JSONDecodeError as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
- except Exception as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
- # 如厕频次异常
- def handle_toileting_frequency_abnormal(self):
- try:
- dev_id = self.dev_id_
- device:Device = dev_map_find(dev_id)
- if not device:
- return
- start_dt, end_dt = helper.get_query_time_range(self.param_)
- params = {
- "dev_id" : self.dev_id_,
- "event_type": event_desc_map[EventType.TOILETING_DETECTION.value],
- "start_dt": start_dt,
- "end_dt": end_dt
- }
- userdata = {
- "start_dt" : start_dt,
- "end_dt" : end_dt
- }
- db_req_que.put(DBRequest(
- sql=sqls.sql_query_events_by_datetime,
- params=params,
- callback=self.cb_toileting_frequency_abnormal,
- userdata=userdata))
- except json.JSONDecodeError as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
- except Exception as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
- # 起夜异常
- def handle_night_toileting_frequency_abnormal(self):
- try:
- dev_id = self.dev_id_
- device:Device = dev_map_find(dev_id)
- if not device:
- return
- start_dt, end_dt = helper.get_query_time_range(self.param_)
- params = {
- "dev_id" : self.dev_id_,
- "event_type": event_desc_map[EventType.TOILETING_DETECTION.value],
- "start_dt": start_dt,
- "end_dt": end_dt
- }
- userdata = {
- "start_dt" : start_dt,
- "end_dt" : end_dt
- }
- db_req_que.put(DBRequest(
- sql=sqls.sql_query_events_by_datetime,
- params=params,
- callback=self.cb_toileting_frequency_abnormal,
- userdata=userdata))
- except json.JSONDecodeError as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
- except Exception as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
- # 卫生间频次统计
- def handle_bathroom_stay_frequency(self):
- try:
- dev_id = self.dev_id_
- device:Device = dev_map_find(dev_id)
- if not device:
- return
- start_dt, end_dt = helper.get_query_time_range(self.param_)
- params = {
- "dev_id" : self.dev_id_,
- "event_type": event_desc_map[EventType.STAY_DETECTION.value],
- "start_dt": start_dt,
- "end_dt": end_dt
- }
- userdata = {
- "start_dt" : start_dt,
- "end_dt" : end_dt
- }
- db_req_que.put(DBRequest(
- sql=sqls.sql_query_events_by_datetime,
- params=params,
- callback=self.cb_bathroom_stay_frequency,
- userdata=userdata))
- except json.JSONDecodeError as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
- except Exception as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
- # 异常消失
- def handle_target_absence(self):
- try:
- dev_id = self.dev_id_
- device:Device = dev_map_find(dev_id)
- if not device:
- return
- now = get_utc_time_s()
- threshold: int = self.event_attr_.time_threshold_
- # 查找最新的落在检测区域的目标
- rtd_unit = self.find_latest_rtd_in_region(device, self.rect_, now, 3)
- if rtd_unit:
- return
- timestamp = get_utc_time_s()
- if self.event_attr_.leave_ts_ == -1:
- self.event_attr_.leave_ts_ = timestamp
- else:
- self.event_attr_.enter_ts_ = timestamp
- if self.event_attr_.enter_ts_ == -1 or self.event_attr_.leave_ts_ == -1:
- return
- # 归并时间内,不认为事件结束
- if now - self.event_attr_.enter_ts_ < self.merge_time_:
- return
- self.event_attr_.absence_time_ = self.event_attr_.enter_ts_ - self.event_attr_.leave_ts_
- absence_time = self.event_attr_.absence_time_
- # 时间小于触发时间阈值,忽略并重置
- if absence_time < threshold :
- self.event_attr_.reset()
- LOGINFO(f"less than threshold_time, alarm_plan: {self.plan_uuid_}, event_type: {self.event_type_}")
- return
- # 构造事件
- # 入库
- info = {
- "start_time": utc_to_bj_s(self.event_attr_.leave_ts_),
- "end_time": utc_to_bj_s(self.event_attr_.enter_ts_),
- "absence_time": absence_time
- }
- event_uuid = str(uuid.uuid4())
- params = {
- "dev_id": dev_id,
- "uuid": event_uuid,
- "plan_uuid": self.plan_uuid_,
- "event_type": event_desc_map[self.event_type_],
- "info": json.dumps(info),
- "is_handle": 0,
- "create_time": get_bj_time_s(),
- "is_deleted": 0,
- "remark": json.dumps({}, ensure_ascii=False)
- }
- db_req_que.put(DBRequest(sql=sqls.sql_insert_events, params=params, callback=None))
- # 通知
- mqtt_send.alarm_event(dev_id, self.dev_name_, event_uuid, self.plan_uuid_, event_desc_map[self.event_type_], info, "events")
- LOGDBG(f"new event: {event_desc_map[self.event_type_]}, absence_time: {absence_time}")
- self.event_attr_.reset()
- except json.JSONDecodeError as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
- except Exception as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
- # 如厕频次统计回调
- def cb_toileting_frequency(self, result, userdata):
- try:
- if result:
- count = 0
- event_list = []
- for row in result:
- dev_id: str = row.get("dev_id")
- event_uuid: str = row.get("uuid")
- plan_uuid: str = row.get("plan_uuid")
- event_type: int = row.get("event_type")
- info: dict = json.loads(row["info"]) if row.get("info") else {}
- is_handle: str = row.get("is_handle")
- create_time: str = row.get("create_time")
- update_time: str = row.get("update_time")
- is_deleted: str = row.get("is_deleted")
- remark: str = row.get("remark")
- event_list.append(info)
- this_event_uuid = str(uuid.uuid4())
- last_info = {
- "start_time" : userdata["start_dt"],
- "end_time" : userdata["end_dt"],
- "count" : len(event_list),
- "event_list" : event_list
- }
- # 入库
- event_uuid = str(uuid.uuid4())
- params = {
- "dev_id": dev_id,
- "uuid": this_event_uuid,
- "plan_uuid": self.plan_uuid_,
- "event_type": event_desc_map[self.event_type_],
- "info": json.dumps(last_info, ensure_ascii=False),
- "is_handle": 0,
- "create_time": get_bj_time_s(),
- "is_deleted": 0,
- "remark": json.dumps({}, ensure_ascii=False)
- }
- db_req_que.put(DBRequest(sql=sqls.sql_insert_events, params=params, callback=None))
- # 通知
- mqtt_send.alarm_event(dev_id, self.dev_name_, this_event_uuid, plan_uuid, event_desc_map[self.event_type_], last_info, "events")
- LOGINFO(f"EventAttr_ToiletingFrequency succeed")
- else:
- LOGDBG("EventAttr_ToiletingFrequency, empty result")
- except json.JSONDecodeError as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
- except Exception as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
- return
- # 夜间如厕频次统计回调
- def cb_night_toileting_frequency(self, result, userdata):
- try:
- if result:
- count = 0
- event_list = []
- for row in result:
- dev_id: str = row.get("dev_id")
- event_uuid: str = row.get("uuid")
- plan_uuid: str = row.get("plan_uuid")
- event_type: int = row.get("event_type")
- info: dict = json.loads(row["info"]) if row.get("info") else {}
- is_handle: str = row.get("is_handle")
- create_time: str = row.get("create_time")
- update_time: str = row.get("update_time")
- is_deleted: str = row.get("is_deleted")
- remark: str = row.get("remark")
- event_list.append(info)
- this_event_uuid = str(uuid.uuid4())
- last_info = {
- "start_time" : userdata["start_dt"],
- "end_time" : userdata["end_dt"],
- "count" : len(event_list),
- "event_list" : event_list
- }
- # 入库
- event_uuid = str(uuid.uuid4())
- params = {
- "dev_id": dev_id,
- "uuid": this_event_uuid,
- "plan_uuid": self.plan_uuid_,
- "event_type": event_desc_map[self.event_type_],
- "info": json.dumps(last_info, ensure_ascii=False),
- "is_handle": 0,
- "create_time": get_bj_time_s(),
- "is_deleted": 0,
- "remark": json.dumps({}, ensure_ascii=False)
- }
- db_req_que.put(DBRequest(sql=sqls.sql_insert_events, params=params, callback=None))
- # 通知
- mqtt_send.alarm_event(dev_id, self.dev_name_, this_event_uuid, plan_uuid, event_desc_map[self.event_type_], last_info, "events")
- LOGINFO(f"EventAttr_ToiletingFrequency succeed")
- else:
- LOGDBG("EventAttr_ToiletingFrequency, empty result")
- except json.JSONDecodeError as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
- except Exception as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
- return
- # 如厕频次异常回调
- def cb_toileting_frequency_abnormal(self, result, userdata):
- try:
- if result:
- count = 0
- event_list = []
- for row in result:
- dev_id: str = row.get("dev_id")
- event_uuid: str = row.get("uuid")
- plan_uuid: str = row.get("plan_uuid")
- event_type: int = row.get("event_type")
- info: dict = json.loads(row["info"]) if row.get("info") else {}
- is_handle: str = row.get("is_handle")
- create_time: str = row.get("create_time")
- update_time: str = row.get("update_time")
- is_deleted: str = row.get("is_deleted")
- remark: str = row.get("remark")
- event_list.append(info)
- if len(event_list) < self.event_attr_.threshold_count_:
- return
- this_event_uuid = str(uuid.uuid4())
- last_info = {
- "start_time" : userdata["start_dt"],
- "end_time" : userdata["end_dt"],
- "count" : len(event_list),
- "event_list" : event_list
- }
- # 入库
- event_uuid = str(uuid.uuid4())
- params = {
- "dev_id": dev_id,
- "uuid": this_event_uuid,
- "plan_uuid": self.plan_uuid_,
- "event_type": event_desc_map[self.event_type_],
- "info": json.dumps(last_info, ensure_ascii=False),
- "is_handle": 0,
- "create_time": get_bj_time_s(),
- "is_deleted": 0,
- "remark": json.dumps({}, ensure_ascii=False)
- }
- db_req_que.put(DBRequest(sql=sqls.sql_insert_events, params=params, callback=None))
- # 通知
- mqtt_send.alarm_event(dev_id, self.dev_name_, this_event_uuid, self.plan_uuid_, event_desc_map[self.event_type_], last_info, "events")
- LOGINFO(f"EventAttr_ToiletingFrequency succeed")
- else:
- LOGDBG("EventAttr_ToiletingFrequency, empty result")
- except json.JSONDecodeError as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
- except Exception as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
- return
- # 起夜异常回调
- def cb_toileting_frequency_abnormal(self, result, userdata):
- try:
- if result:
- count = 0
- event_list = []
- for row in result:
- dev_id: str = row.get("dev_id")
- event_uuid: str = row.get("uuid")
- plan_uuid: str = row.get("plan_uuid")
- event_type: int = row.get("event_type")
- info: dict = json.loads(row["info"]) if row.get("info") else {}
- is_handle: str = row.get("is_handle")
- create_time: str = row.get("create_time")
- update_time: str = row.get("update_time")
- is_deleted: str = row.get("is_deleted")
- remark: str = row.get("remark")
- event_list.append(info)
- if len(event_list) < self.event_attr_.threshold_count_:
- return
- this_event_uuid = str(uuid.uuid4())
- last_info = {
- "start_time" : userdata["start_dt"],
- "end_time" : userdata["end_dt"],
- "count" : len(event_list),
- "event_list" : event_list
- }
- # 入库
- event_uuid = str(uuid.uuid4())
- params = {
- "dev_id": dev_id,
- "uuid": this_event_uuid,
- "plan_uuid": self.plan_uuid_,
- "event_type": event_desc_map[self.event_type_],
- "info": json.dumps(last_info, ensure_ascii=False),
- "is_handle": 0,
- "create_time": get_bj_time_s(),
- "is_deleted": 0,
- "remark": json.dumps({}, ensure_ascii=False)
- }
- db_req_que.put(DBRequest(sql=sqls.sql_insert_events, params=params, callback=None))
- # 通知
- mqtt_send.alarm_event(dev_id, self.dev_name_, this_event_uuid, self.plan_uuid_, event_desc_map[self.event_type_], last_info, "events")
- LOGINFO(f"EventAttr_ToiletingFrequency succeed")
- else:
- LOGDBG("EventAttr_ToiletingFrequency, empty result")
- except json.JSONDecodeError as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
- except Exception as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
- return
- # 如厕频次统计回调
- def cb_bathroom_stay_frequency(self, result, userdata):
- try:
- if result:
- count = 0
- event_list = []
- for row in result:
- dev_id: str = row.get("dev_id")
- event_uuid: str = row.get("uuid")
- plan_uuid: str = row.get("plan_uuid")
- event_type: int = row.get("event_type")
- info: dict = json.loads(row["info"]) if row.get("info") else {}
- is_handle: str = row.get("is_handle")
- create_time: str = row.get("create_time")
- update_time: str = row.get("update_time")
- is_deleted: str = row.get("is_deleted")
- remark: str = row.get("remark")
- event_list.append(info)
- this_event_uuid = str(uuid.uuid4())
- last_info = {
- "start_time" : userdata["start_dt"],
- "end_time" : userdata["end_dt"],
- "count" : len(event_list),
- "event_list" : event_list
- }
- # 入库
- event_uuid = str(uuid.uuid4())
- params = {
- "dev_id": dev_id,
- "uuid": this_event_uuid,
- "plan_uuid": self.plan_uuid_,
- "event_type": event_desc_map[self.event_type_],
- "info": json.dumps(last_info, ensure_ascii=False),
- "is_handle": 0,
- "create_time": get_bj_time_s(),
- "is_deleted": 0,
- "remark": json.dumps({}, ensure_ascii=False)
- }
- db_req_que.put(DBRequest(sql=sqls.sql_insert_events, params=params, callback=None))
- # 通知
- mqtt_send.alarm_event(dev_id, self.dev_name_, this_event_uuid, plan_uuid, event_desc_map[self.event_type_], last_info, "events")
- LOGINFO(f"EventAttr_ToiletingFrequency succeed")
- else:
- LOGDBG("EventAttr_ToiletingFrequency, empty result")
- except json.JSONDecodeError as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
- except Exception as e:
- tb_info = traceback.extract_tb(e.__traceback__)
- for frame in tb_info:
- LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
- return
|