客流分析核心算法 trajectory_event_analyzer数据结构

发布于：2025-07-22 ⋅ 阅读:(16) ⋅ 点赞:(0)

客流分析核心算法 trajectory_event_analyzerV4.py数据结构

文章目录

一、算法描述

1、描述

一套基于视频分析的客流统计算法，旨在准确统计监控区域内（例如，零售店铺，办公室）进店、过店、出店和重进店的人数。该算法通过结合区域划分、多目标追踪、客流统计和行人重识别四个核心模块，实现对区域内的人物的运动轨迹的精确跟踪，并根据轨迹与预设区域的关系，判定并统计各类客流事件。

2、客流分析模块trajectory_event_analyzerV4.py解析

1. 分层统计：

  • 个体级：跟踪每个顾客的轨迹和状态变化
  • 区域级：统计每个区域的进出人数和当前人数
  • 全局级：聚合所有区域数据计算最终客流指标

2. 状态一致性检查：

if consecutive_frames >= N_FRAMES:
    if current_region != confirmed_region:
        update_region_stats()  # 只有连续N帧在同一区域才更新

3. 区域状态统计：

if track_id not in region_states[self.confirmed_region].out_degree_all_person_id:
    region_states[self.confirmed_region].out_degree_count += 1
    region_states[self.confirmed_region].out_degree_all_person_id.append(track_id)

if track_id not in region_states[self.current_instantaneous_region].in_degree_all_person_id:
    region_states[self.current_instantaneous_region].in_degree_count += 1
    region_states[self.current_instantaneous_region].in_degree_all_person_id.append(track_id)

4、客流状态统计

result_data['进店'] = region_states[RegionType.INDOOR].in_degree_count
result_data['出店'] = region_states[RegionType.INDOOR].out_degree_count
result_data['重进店'] = region_states[RegionType.INDOOR].re_in_degree_count

count_areas = [RegionType.LEFT_OUTSIDE, RegionType.FRONT_OUTSIDE, RegionType.RIGHT_OUTSIDE]
count_out_degree_sum = 0
for count_area in count_areas:
    count_out_degree_sum += region_states[count_area].in_degree_count
inout_same_ids_num = list(set(region_states[RegionType.INDOOR].in_degree_all_person_id) & set(
    region_states[RegionType.INDOOR].out_degree_all_person_id))

5. ReID集成：

isReid, dist, res = reid_person.reid_frame(frame, body_box)
if isReid:
    region_states[RegionType.INDOOR].re_in_degree_count += 1
    print(f'重识别到：{res},距离：{dist}')

    if debug_reid:
        self.save_temp_image(frame, body_box, res, dist, 'person')

else:
    print(f'发现新人物{reid_person.people_count - 1}')
    if debug_reid:
        os.makedirs(reid_temp_dir + "_" + 'person', exist_ok=True)
        reid_img_path = os.path.join(reid_temp_dir + "_" + 'person', f'{reid_person.people_count - 1}.png')

        # 截取ROI用于保存
        x1, y1, x2, y2 = map(int, body_box[:4])
        img2 = frame[y1:y2, x1:x2]
        cv2.imwrite(reid_img_path, img2)

6. 数据清理机制：

# 定期清理历史ID记录
if len(out_degree_all_person_id) > MAX_HISTORY:
    # 保留最近N个ID
    out_degree_all_person_id = out_degree_all_person_id[-MAX_HISTORY:]

通过多级数据结构实现了高效的客流统计，区域状态对象负责局部统计，全局分析器负责聚合计算，既能实时更新数据，又能保证统计准确性。

二、核心模块解释

1、核心模块解释

1. TrajectoryEventAnalyzer（主控制器）

  • 职责：管理所有追踪个体状态，协调区域状态更新和事件判定
  • 关键数据结构：
    tracked_individuals: Dict[int, TrackedIndividualState] # 个体ID到状态对象的映射
    region_states: Dict[RegionType, RegionState] # 区域类型到区域状态的映射
    result_data: Dict[str, int] # 全局客流统计结果

2. TrackedIndividualState（个体状态机）

  • 职责：管理单个个体的轨迹、区域状态和事件判定逻辑
  • 关键数据结构：
    trajectory_points: List[Tuple[frame, timestamp, coords, region]] # 历史轨迹点
    event_path: List[Tuple[confirmed_region, frame]] # 区域状态变化序列
    key_frames: Dict[str, int] # 关键事件帧记录

3. RegionState（区域状态）

  • 职责：维护区域级客流统计
  • 关键数据结构：
    in_degree_count: int # 进入该区域的总人数
    out_degree_count: int # 离开该区域的总人数
    in_degree_all_person_id: List[int] # 进入过该区域的ID列表
    out_degree_all_person_id: List[int] # 离开过该区域的ID列表

4. Event（事件记录）

  • 职责：存储已确认的事件信息
  • 数据结构：
    event_id: str # 唯一事件ID
    track_id: int # 关联个体ID
    event_type: EventType # 事件类型枚举
    event_frame: int # 事件发生帧
    details: Dict # 事件详情

2、核心算法流程

1.客流分析算法流程

在这里插入图片描述

2.关键交互逻辑

（1）区域状态更新
  • 当个体连续N帧处于同一区域时，更新confirmed_region
  • 上一个区域不为None且与confirmed_region区域变化时更新两个区域的进出度统计：
# 离开旧区域 
region_states[old_region].out_degree_count += 1 
region_states[old_region].out_degree_all_person_id.append(track_id)

# 进入新区域 
region_states[new_region].in_degree_count += 1 
region_states[new_region].in_degree_all_person_id.append(track_id)
（2） 客流统计计算
  • 进店数 = 店内区域的进人数
  • 出店数 = 店内区域的出人数
  • 过店数 = max(∑(店外区域进人数) - 出店数 ,过店数）
result_data['进店'] = region_states[RegionType.INDOOR].in_degree_count
result_data['出店'] = region_states[RegionType.INDOOR].out_degree_count
result_data['重进店'] = region_states[RegionType.INDOOR].re_in_degree_count

count_areas = [RegionType.LEFT_OUTSIDE, RegionType.FRONT_OUTSIDE, RegionType.RIGHT_OUTSIDE]
count_out_degree_sum = 0
for count_area in count_areas:
    count_out_degree_sum += region_states[count_area].in_degree_count
inout_same_ids_num = list(set(region_states[RegionType.INDOOR].in_degree_all_person_id) & set(
    region_states[RegionType.INDOOR].out_degree_all_person_id))

# 过店数=店外区域的出度-店内区域的出度+店内进出id的重复数
print(len(inout_same_ids_num))
print(region_states[RegionType.INDOOR].in_degree_all_person_id,
      region_states[RegionType.INDOOR].out_degree_all_person_id)
print(count_out_degree_sum - region_states[RegionType.INDOOR].out_degree_count + len(inout_same_ids_num))
print(count_out_degree_sum - region_states[RegionType.INDOOR].out_degree_count)
result_data['过店'] = max(count_out_degree_sum - region_states[RegionType.INDOOR].out_degree_count, result_data['过店'])
return result_data
（3）ReID识别流程
  • 当个体进入店内区域时触发重识别：
isReid, dist, res = reid_person.reid_frame(frame,body_box)
if isReid:
    region_states[RegionType.INDOOR].re_in_degree_count+=1
    print(f'重识别到：{res},距离：{dist}')

    if debug_reid:
        self.save_temp_image(frame,body_box,res,dist,'person')

else:
    print(f'发现新人物{reid_person.people_count-1}')
    if debug_reid:
        os.makedirs(reid_temp_dir+"_"+'person',exist_ok=True)
        reid_img_path=os.path.join(reid_temp_dir+"_"+'person',f'{reid_person.people_count-1}.png')

        # 截取ROI用于保存
        x1, y1, x2, y2 = map(int, body_box[:4])
        img2 = frame[y1:y2,x1:x2]
        cv2.imwrite(reid_img_path,img2)

三、示例

1.示例1：顾客入店流程

sequenceDiagram
participant 顾客
participant 轨迹分析器
participant 区域状态

顾客->>轨迹分析器: 进入左侧店外区域(LEFT_OUTSIDE)
轨迹分析器->>区域状态: 更新LEFT_OUTSIDE区域
区域状态-->>轨迹分析器: self.event_path.append(LEFT_OUTSIDE) 

顾客->>轨迹分析器: 进入识别区域(RECOGNITION)
轨迹分析器->>区域状态: 更新RECOGNITION区域
区域状态-->>轨迹分析器: 
LEFT_OUTSIDE.out_degree_count+=1 
RECOGNITION.in_degree_count+=1
RECOGNITION.in_degree_all_person_id=[101]

顾客->>轨迹分析器: 进入店内区域(INDOOR)
轨迹分析器->>区域状态: 更新INDOOR区域
区域状态-->>轨迹分析器:
RECOGNITION.out_degree_count+=1 
INDOOR.in_degree_count+=1
INDOOR.in_degree_all_person_id=[101]

轨迹分析器->>全局统计: 
更新进店数=INDOOR.in_degree_count

结果->> 进店数+1

2.示例2：顾客过店流程

sequenceDiagram
participant 顾客
participant 轨迹分析器
participant 区域状态

顾客->>轨迹分析器: 进入前方店外区域(FRONT_OUTSIDE)
轨迹分析器->>区域状态: 更新FRONT_OUTSIDE区域
区域状态-->>轨迹分析器:  self.event_path.append(FRONT_OUTSIDE) 

顾客->>轨迹分析器: 进入识别区域(RECOGNITION)
轨迹分析器->>区域状态: 更新RECOGNITION区域
区域状态-->>轨迹分析器: 
FRONT_OUTSIDE.out_degree_count+=1 
RECOGNITION.in_degree_count+=1
RECOGNITION.in_degree_all_person_id=[101]

顾客->>轨迹分析器: 右店外区域(RIGHT_OUTSIDE)
轨迹分析器->>区域状态: 更新RIGHT_OUTSIDE区域
区域状态-->>轨迹分析器: 
RECOGNITION.out_degree_count+=1 
RIGHT_OUTSIDE.in_degree_count+=1
RIGHT_OUTSIDE.in_degree_all_person_id=[101]

轨迹分析器->>全局统计: 计算过店数
轨迹分析器->>全局统计: 过店数 = (店外区域进人数总和) - (出店数) 
轨迹分析器->>全局统计: 过店数 = (1+0+0) - 0  = 1

3. 完整示例演示

场景描述：

  1. 顾客A：入店（ID 101）
  2. 顾客B：过店（ID 102）
  3. 顾客C：入店后出店（ID 103）
  4. 顾客D：重进店（ID 104，被ReID识别为老顾客）

数据结构变化：

# 初始状态
region_states = {
    LEFT_OUTSIDE: RegionState(in_degree=0, out_degree=0, ids=[]),
    FRONT_OUTSIDE: RegionState(in_degree=0, out_degree=0, ids=[]),
    RIGHT_OUTSIDE: RegionState(in_degree=0, out_degree=0, ids=[]),
    RECOGNITION: RegionState(in_degree=0, out_degree=0, ids=[]),
    INDOOR: RegionState(in_degree=0, out_degree=0, re_in_degree=0, ids=[])
}

# 顾客A入店后
region_states[LEFT_OUTSIDE].in_degree = 1  # 进入店外
region_states[RECOGNITION].in_degree = 1   # 进入识别区
region_states[INDOOR].in_degree = 1        # 进入店内

# 顾客B过店后
region_states[FRONT_OUTSIDE].in_degree = 1  # 进入店外
region_states[RECOGNITION].in_degree = 2   # 进入识别区
region_states[FRONT_OUTSIDE].in_degree = 2 # 返回店外（重复不计数）

# 顾客C入店后出店
region_states[RIGHT_OUTSIDE].in_degree = 1  # 进入店外
region_states[RECOGNITION].in_degree = 3    # 进入识别区
region_states[INDOOR].in_degree = 2         # 进入店内
region_states[INDOOR].out_degree = 1        # 离开店内
region_states[RECOGNITION].in_degree = 4    # 进入识别区（出店后）

# 顾客D重进店
region_states[FRONT_OUTSIDE].in_degree = 3  # 进入店外
region_states[RECOGNITION].in_degree = 5    # 进入识别区
region_states[INDOOR].in_degree = 3         # 进入店内
region_states[INDOOR].re_in_degree = 1       # 重识别计数

最终客流统计：
# 计算全局统计
stats = calculate_global_stats()

# 结果：
{
    '进店': 3,       # INDOOR.in_degree_count
    '出店': 1,       # INDOOR.out_degree_count
    '重进店': 1,     # INDOOR.re_in_degree_count
    '过店': (3+1+1) - 1 + 1 = 5  # (店外进店总数) - (出店数) + (重复ID数)
    # 店外进店总数: LEFT(1)+FRONT(3)+RIGHT(1)=5
    # 重复ID数: 顾客C(103)同时出现在进店和出店列表
}

网站公告

今日签到

点亮在社区的每一天
去签到

热门文章

最新发布