import argparse

import os

import pickle

import sys

import tempfile

import time

from typing import List, Optional

from pathlib import Path

import cv2

import numpy as np

import datetime

def configure_utf8_io():

try:

if [Link] == 'nt':

import ctypes

kernel32 = [Link].kernel32

[Link](65001)

[Link](65001)

except Exception:

pass

for stream_name in ('stdin', 'stdout', 'stderr'):

stream = getattr(sys, stream_name, None)

if stream is not None and hasattr(stream, 'reconfigure'):

 try:

[Link](encoding='utf-8', errors='replace')

except Exception:

pass

configure_utf8_io()

# Make project roots importable and ensure OpenGait's package is prioritized

THIS_DIR = [Link]([Link](__file__))

ROOT_DIR = [Link](THIS_DIR)

DA_SRC = [Link](ROOT_DIR, 'DA_RobotGuide', 'src')

OPEN_GAIT_ROOT = [Link](ROOT_DIR, 'OpenGait')

OPEN_GAIT_PKG = [Link](OPEN_GAIT_ROOT, 'opengait')

OPEN_GAIT_MISC = [Link](OPEN_GAIT_ROOT, 'misc')

OPEN_GAIT_OUT = [Link](OPEN_GAIT_ROOT, 'output', 'CASIA-B', 'GaitGL')

FACE_REGISTER_PATH = [Link](DA_SRC, 'register_capture_and_embed.py')

# Insert project roots first to avoid import collisions and make fusion_system importable

for p in (THIS_DIR, ROOT_DIR, OPEN_GAIT_ROOT, OPEN_GAIT_PKG, OPEN_GAIT_MISC,

DA_SRC):

if p not in [Link]:

[Link](0, p)

# Import local helpers (these modules exist in the repo)

import traceback

import importlib
_import_err = None

SCRFD = None

load_model_onnx = None

take_box_detector = None

alignment = None

process_kps = None

process_onnx = None

def try_import_module(candidates):

last_exc = None

for name in candidates:

try:

return importlib.import_module(name)

except Exception:

last_exc = traceback.format_exc()

raise RuntimeError(last_exc if last_exc is not None else 'Import failed')

def _log(msg: str):

t = [Link]().isoformat()

try:

print(f"[{t}] {msg}", flush=True)

except Exception:

print(msg)

def safe_load_gait_model(cfg_path: str):

"""Load OpenGait model using a local single-process distributed context.

 Returns (model, eval_trfs) or raises the original exception.

"""

_log(f'safe_load_gait_model: loading cfg {cfg_path}')

try:

import torch

except Exception:

torch = None

if torch is None or not hasattr(torch, 'distributed'):

raise RuntimeError('[Link] is not available for OpenGait runtime

initialization')

if not [Link].is_initialized():

fd, rendezvous_path = [Link](prefix='opengait_dist_', suffix='.tmp')

[Link](fd)

init_method = Path(rendezvous_path).resolve().as_uri()

_log(f'Initializing local [Link] process group via {init_method}')

[Link].init_process_group(

backend='gloo',

init_method=init_method,

rank=0,

world_size=1,

mod = gait_load_fallback
 model, eval_trfs = mod.load_model(cfg_path)

_log('safe_load_gait_model: load_model returned')

return model, eval_trfs

try:

# Common possible module paths inside this repo

lm = None

pd = None

try:

lm = try_import_module(['utils.load_model', '[Link].load_model', 'load_model'])

except Exception as e:

_import_err = str(e)

try:

pd = try_import_module(['utils.process_data', '[Link].process_data', 'process_data'])

except Exception as e:

_import_err = (_import_err or '') + '\n' + str(e)

if lm is not None:

SCRFD = getattr(lm, 'SCRFD', None)

load_model_onnx = getattr(lm, 'load_model_onnx', None)

if pd is not None:

take_box_detector = getattr(pd, 'take_box_detector', None)

alignment = getattr(pd, 'alignment', None)

process_kps = getattr(pd, 'process_kps', None)

process_onnx = getattr(pd, 'process_onnx', None)

except Exception:
 _import_err = traceback.format_exc()

# helper: load a module from a specific .py file path

import [Link]

def load_module_from_path(path, mod_name):

try:

spec = [Link].spec_from_file_location(mod_name, path)

if spec is None or [Link] is None:

return None

mod = [Link].module_from_spec(spec)

[Link].exec_module(mod)

return mod

except Exception:

print('load_module_from_path failed for', path)

traceback.print_exc()

return None

face_register_helper = None

if [Link](FACE_REGISTER_PATH):

face_register_helper = load_module_from_path(FACE_REGISTER_PATH,
'face_register_helper')

# Import sync helper functions

try:

from sync_databases import load_face_db, load_gait_db, group_centroids, build_synced

except Exception:
 sync_module = load_module_from_path([Link](THIS_DIR, 'sync_databases.py'),
'fusion_sync_databases')

if sync_module is None:

raise

load_face_db = sync_module.load_face_db

load_gait_db = sync_module.load_gait_db

group_centroids = sync_module.group_centroids

build_synced = sync_module.build_synced

# Prefer OpenGait's enrollment utilities (webcam_enroll) for gait capture + DB writes

enroll_from_webcam = None

enroll_from_two_webcams = None

gait_load_fallback = None

try:

# try package imports placed on [Link]

m = importlib.import_module('webcam_enroll')

enroll_from_webcam = getattr(m, 'enroll_from_webcam', None)

enroll_from_two_webcams = getattr(m, 'enroll_from_two_webcams', None)

gait_load_fallback = m

print('Loaded webcam_enroll from [Link]')

except Exception:

try:

m = importlib.import_module('misc.webcam_enroll')

enroll_from_webcam = getattr(m, 'enroll_from_webcam', None)

enroll_from_two_webcams = getattr(m, 'enroll_from_two_webcams', None)

gait_load_fallback = m
 print('Loaded misc.webcam_enroll')

except Exception:

try:

m = importlib.import_module('[Link].webcam_enroll')

enroll_from_webcam = getattr(m, 'enroll_from_webcam', None)

enroll_from_two_webcams = getattr(m, 'enroll_from_two_webcams', None)

gait_load_fallback = m

print('Loaded [Link].webcam_enroll')

except Exception:

# fallback: load from file path if present

wk_path = [Link](ROOT_DIR, 'OpenGait', 'misc', 'webcam_enroll.py')

print('OpenGait webcam_enroll fallback, trying file:', wk_path, 'exists=',

[Link](wk_path))

if [Link](wk_path):

wk_mod = load_module_from_path(wk_path, 'webcam_enroll_fallback')

if wk_mod is not None:

enroll_from_webcam = getattr(wk_mod, 'enroll_from_webcam', None)

enroll_from_two_webcams = getattr(wk_mod, 'enroll_from_two_webcams', None)

gait_load_fallback = wk_mod

print('Loaded webcam_enroll.py via file fallback')

else:

print('Failed to load webcam_enroll via file fallback')

def capture_face_embedding(face_cam: int, detector, backbone, quality, timeout: int = 20,

show: bool = True):

cap = [Link](face_cam)

if not [Link]():
 raise RuntimeError(f"Cannot open face camera {face_cam}")

print(f"Opening face camera {face_cam}. Press Space to capture, or wait {timeout}s to

auto-capture.")

start = [Link]()

emb = None

while True:

ok, frame = [Link]()

if not ok:

continue

vis = [Link]()

bbs, kpss = take_box_detector(frame, detector)

if bbs is not None and len(bbs) > 0:

i = max(range([Link][0]), key=lambda idx: float(bbs[idx][4]))

x1, y1, x2, y2, _ = bbs[i].astype(int)

[Link](vis, (x1, y1), (x2, y2), (0, 255, 0), 2)

[Link](vis, 'Press SPACE to capture face', (10, 30), cv2.FONT_HERSHEY_SIMPLEX,

0.8, (0,255,0), 2)

if show:

[Link]('Enroll Face', vis)

k = [Link](1) & 0xFF

else:

k = None

if k == 32: # space

print('Capturing face now...')

if bbs is None or len(bbs) == 0:

print('No face detected, try again')

continue
 face_box_local = None

try:

# use the top-scoring detected face

i = max(range([Link][0]), key=lambda idx: float(bbs[idx][4]))

x1, y1, x2, y2, _ = bbs[i].astype(int)

crop = frame[y1:y2, x1:x2]

kps = kpss[i]

_, _, _, _, _, _, _, _, l_eye, r_eye = process_kps(kps)

aligned = alignment(crop, l_eye, r_eye)

aligned = [Link](aligned, (112, 112))

q, emb_t = process_onnx(aligned, backbone, quality)

emb = emb_t.cpu().detach().numpy().astype(np.float32)[0]

emb = [Link](-1)

print('Captured face embedding shape:', [Link])

break

except Exception as e:

print('Face processing failed:', e)

continue

if [Link]() - start > timeout:

print('Auto-capturing (timeout)')

if bbs is None or len(bbs) == 0:

continue

try:

i = max(range([Link][0]), key=lambda idx: float(bbs[idx][4]))

x1, y1, x2, y2, _ = bbs[i].astype(int)

crop = frame[y1:y2, x1:x2]

 kps = kpss[i]

_, _, _, _, _, _, _, _, l_eye, r_eye = process_kps(kps)

aligned = alignment(crop, l_eye, r_eye)

aligned = [Link](aligned, (112, 112))

q, emb_t = process_onnx(aligned, backbone, quality)

emb = emb_t.cpu().detach().numpy().astype(np.float32)[0]

emb = [Link](-1)

print('Captured face embedding shape:', [Link])

break

except Exception as e:

print('Auto face process failed:', e)

continue

[Link]()

if show:

[Link]()

return emb

def capture_gait_sequence_from_local(gait_cam: int, gait_runtime, frames_required: int =

45, timeout: int = 30, show: bool = True):

if [Link] == 'nt':

cap = [Link](gait_cam, cv2.CAP_DSHOW)

if not [Link]():

[Link]()

cap = [Link](gait_cam)

else:

cap = [Link](gait_cam)
 if not [Link]():

raise RuntimeError(f"Cannot open gait camera {gait_cam}")

preprocess = None if gait_runtime is None else gait_runtime.get('preprocess')

if preprocess is None:

raise RuntimeError('Gait preprocess helper is not available')

bsub = cv2.createBackgroundSubtractorMOG2(history=500, varThreshold=16,

detectShadows=False)

buffer = []

start = [Link]()

print(f"Capturing gait from camera {gait_cam}. Move in the view. Collecting

{frames_required} usable frames.")

while True:

ok, frame = [Link]()

if not ok:

continue

gray = [Link](frame, cv2.COLOR_BGR2GRAY)

fg = [Link](gray)

_, fg_bin = [Link](fg, 127, 255, cv2.THRESH_BINARY)

try:

proc = preprocess(fg_bin)

except Exception as e:

print(f'Error occurred while preprocessing gait frame: {e}')

proc = None

if proc is not None:

[Link](proc)

print('\rCollected frames:', len(buffer), end='')

if show:
 [Link](f'Gait Cam {gait_cam}', frame)

if [Link](1) & 0xFF == ord('q'):

break

if len(buffer) >= frames_required:

break

if [Link]() - start > timeout and len(buffer) >= 10:

print('\nTimeout reached, using collected frames')

break

[Link]()

if show:

[Link]()

if len(buffer) == 0:

raise RuntimeError('No usable gait frames collected')

seq = [Link](buffer, axis=0)

print(f'Captured gait sequence from cam {gait_cam} shape: {[Link]}')

return seq

def capture_gait_sequence(gait_cam: int, gait_runtime, user_label: str, frames_required:

int = 45, timeout: int = 30, show: bool = True):

if show and gait_load_fallback is not None and hasattr(gait_load_fallback,

'capture_and_process_with_yolo'):

return gait_load_fallback.capture_and_process_with_yolo(

user_label=user_label,

camera_index=int(gait_cam),

num_frames=int(frames_required),

show=show,

)
 return capture_gait_sequence_from_local(

gait_cam=int(gait_cam),

gait_runtime=gait_runtime,

frames_required=int(frames_required),

timeout=int(timeout),

show=show,

def build_gait_sequence_windows(seq: [Link], samples_per_cam: int,

min_window_frames: int) -> List[[Link]]:

seq = [Link](seq)

if [Link] != 3:

raise ValueError(f'Expected gait sequence with shape [N, H, W], got {[Link]}')

total_frames = int([Link][0])

if total_frames <= max(1, int(min_window_frames)) or samples_per_cam <= 1:

return [seq]

windows = [seq]

if total_frames < int(min_window_frames) * 2:

return windows

extra_samples = max(0, int(samples_per_cam) - 1)

if extra_samples == 0:

return windows
 window_len = max(int(min_window_frames), int(round(total_frames * 0.7)))

window_len = min(window_len, total_frames)

if window_len >= total_frames:

return windows

max_start = total_frames - window_len

seen = {(0, total_frames)}

for start in [Link](0, max_start, num=extra_samples, dtype=int):

start = int(start)

end = start + window_len

key = (start, end)

if key in seen:

continue

[Link](key)

[Link](seq[start:end])

return windows[: max(1, int(samples_per_cam))]

def extract_gait_embeddings_from_sequence(seq: [Link], gait_runtime,

samples_per_cam: int, min_window_frames: int) -> List[[Link]]:

if gait_runtime is None or gait_runtime.get('extract') is None:

raise RuntimeError('Gait runtime is not initialized for embedding extraction')

windows = build_gait_sequence_windows(seq, samples_per_cam=samples_per_cam,

min_window_frames=min_window_frames)

embeddings: List[[Link]] = []

for index, window in enumerate(windows, start=1):

emb = gait_runtime['extract'](gait_runtime['model'], gait_runtime['eval_trfs'], window)

 emb = [Link](emb, dtype=np.float32)

[Link](emb)

print(f'Extracted gait embedding sample {index}/{len(windows)} with shape:

{[Link]}')

return embeddings

def capture_gait_embedding(gait_cam: int, gait_runtime, frames_required: int = 45,

timeout: int = 30, show: bool = True):

seq = capture_gait_sequence(

gait_cam=int(gait_cam),

gait_runtime=gait_runtime,

user_label=f'cam_{gait_cam}',

frames_required=int(frames_required),

timeout=int(timeout),

show=show,

embeddings = extract_gait_embeddings_from_sequence(seq, gait_runtime,

samples_per_cam=1, min_window_frames=10)

emb = embeddings[0]

print(f'Gait embedding from cam {gait_cam} shape: {[Link]}')

return emb

def enroll_gait_samples(gait_cams: List[int], person_id: str, gait_runtime, gait_pkl_path: str,

frames_required: int, samples_per_cam: int, min_window_frames: int,

show: bool = True) -> int:

total_added = 0

for cam_index in gait_cams:

 gait_timeout = 60 if show else 30

seq = capture_gait_sequence(

gait_cam=int(cam_index),

gait_runtime=gait_runtime,

user_label=str(person_id),

frames_required=int(frames_required),

timeout=int(gait_timeout),

show=show,

gait_embs = extract_gait_embeddings_from_sequence(

seq,

gait_runtime,

samples_per_cam=int(samples_per_cam),

min_window_frames=int(min_window_frames),

for emb in gait_embs:

append_gait_db(gait_pkl_path, emb, person_id)

total_added += 1

_log(f'Added {len(gait_embs)} gait embeddings from cam {cam_index} for

ID={person_id}')

return total_added

def append_face_db(face_x_path, face_y_path, emb: [Link], label: str):

with open(face_x_path, 'rb') as f:

X = [Link](f)

with open(face_y_path, 'rb') as f:

 y = [Link](f)

X = list(X)

y = list(y)

[Link]([Link](emb, dtype=np.float32))

[Link](label)

with open(face_x_path, 'wb') as f:

[Link]([Link](X, dtype=np.float32), f)

with open(face_y_path, 'wb') as f:

[Link](y, f)

print('Appended face embedding to', face_x_path)

def export_gait_xy_pickles(gait_pkl_path: str, gait_x_path: str, gait_y_path: str):

gait_db, gait_labels = load_gait_db(gait_pkl_path)

if gait_db is None:

gait_db = [Link]((0, 1, 1), dtype=np.float32)

gait_x = [Link](gait_db, dtype=np.float32)

gait_y = [str(label) for label in gait_labels]

with open(gait_x_path, 'wb') as f:

[Link](gait_x, f, protocol=pickle.HIGHEST_PROTOCOL)

with open(gait_y_path, 'wb') as f:

[Link](gait_y, f, protocol=pickle.HIGHEST_PROTOCOL)

print('Exported gait sidecar pickles to', gait_x_path, 'and', gait_y_path)

def append_gait_db(gait_pkl_path, emb: [Link], label: str):

new_emb = [Link](emb, dtype=np.float32)

 # Prefer the OpenGait DB helpers so the on-disk format stays exactly consistent.

if gait_load_fallback is not None and hasattr(gait_load_fallback, 'load_db') and

hasattr(gait_load_fallback, 'save_db'):

db_feats, db_labels = gait_load_fallback.load_db(gait_pkl_path)

db_labels = list(db_labels)

if db_feats.size == 0:

db_feats = new_emb[None, ...]

else:

if tuple(db_feats.shape[1:]) != tuple(new_emb.shape):

raise ValueError(

f'Gait embedding shape mismatch: DB has {db_feats.shape[1:]}, new is

{new_emb.shape}. '

'Aborting to preserve existing dimensions and vectors.'

db_feats = [Link]([db_feats, new_emb[None, ...]], axis=0)

db_labels.append(str(label))

gait_load_fallback.save_db(gait_pkl_path, db_feats, db_labels)

print('Appended gait embedding to', gait_pkl_path)

return

# Generic fallback for older/custom DB structures. Preserve the original structure and
reject mismatched shapes.

data = None

if [Link](gait_pkl_path):

with open(gait_pkl_path, 'rb') as f:

data = [Link](f)

if data is None:
 data = {'embeddings': new_emb[None, ...], 'labels': [str(label)]}

elif isinstance(data, dict):

embed_key = None

for key in ('embeddings', 'features', 'vectors', 'embs'):

if key in data:

embed_key = key

break

if embed_key is None:

embed_key = 'embeddings'

data[embed_key] = new_emb[None, ...]

else:

existing_embs = data[embed_key]

if isinstance(existing_embs, [Link]):

if existing_embs.size == 0:

data[embed_key] = new_emb[None, ...]

else:

if tuple(existing_embs.shape[1:]) != tuple(new_emb.shape):

raise ValueError(

f'Gait embedding shape mismatch: DB has {existing_embs.shape[1:]}, new is

{new_emb.shape}. '

'Aborting to preserve existing dimensions and vectors.'

data[embed_key] = [Link]([existing_embs, new_emb[None, ...]], axis=0)

elif isinstance(existing_embs, list):

if existing_embs:

first_shape = tuple([Link](existing_embs[0]).shape)
 if first_shape != tuple(new_emb.shape):

raise ValueError(

f'Gait embedding shape mismatch: DB has {first_shape}, new is

{new_emb.shape}. '

'Aborting to preserve existing dimensions and vectors.'

existing_embs.append(new_emb)

else:

raise RuntimeError('Unsupported gait embeddings container type')

label_key = 'labels' if 'labels' in data else 'y' if 'y' in data else 'labels'

existing_labels = list([Link](label_key, []))

existing_labels.append(str(label))

data[label_key] = existing_labels

else:

try:

embs, labels = data

except Exception:

raise RuntimeError('Unsupported gait DB format; modify file manually')

labels = list(labels)

if isinstance(embs, [Link]):

if [Link] == 0:

embs = new_emb[None, ...]

else:

if tuple([Link][1:]) != tuple(new_emb.shape):

raise ValueError(
 f'Gait embedding shape mismatch: DB has {[Link][1:]}, new is
{new_emb.shape}. '

'Aborting to preserve existing dimensions and vectors.'

embs = [Link]([embs, new_emb[None, ...]], axis=0)

else:

embs = list(embs)

if embs:

first_shape = tuple([Link](embs[0]).shape)

if first_shape != tuple(new_emb.shape):

raise ValueError(

f'Gait embedding shape mismatch: DB has {first_shape}, new is

{new_emb.shape}. '

'Aborting to preserve existing dimensions and vectors.'

[Link](new_emb)

[Link](str(label))

data = (embs, labels)

with open(gait_pkl_path, 'wb') as f:

[Link](data, f, protocol=pickle.HIGHEST_PROTOCOL)

print('Appended gait embedding to', gait_pkl_path)

def prepare_face_enrollment(face_name: str, face_cam: int, detector, backbone, quality,

face_x_path: str, face_y_path: str, data_dir: str, csv_path: str,

count: int = 20, timeout: int = 0, show: bool = True):

if face_register_helper is None:
 raise RuntimeError(f'Cannot load face registration helper: {FACE_REGISTER_PATH}')

face_name = face_register_helper.obtain_name_interactive(face_name)

if not face_name:

raise RuntimeError('No name provided for face enrollment')

face_name = face_register_helper.sanitize_name(face_name)

existing_x = []

existing_y = []

if [Link](face_x_path) and [Link](face_y_path):

with open(face_x_path, 'rb') as file_obj:

existing_x = [Link](file_obj)

with open(face_y_path, 'rb') as file_obj:

existing_y = [Link](file_obj)

active_labels = {str(label).strip() for label in existing_y}

existing_name = None

if active_labels:

label_to_name = face_register_helper.build_label_to_name_map(csv_path)

target_name_key = face_register_helper.normalize_name_key(face_name)

for label in active_labels:

mapped_name = str(label_to_name.get(label, '')).strip()

if mapped_name and face_register_helper.normalize_name_key(mapped_name) ==

target_name_key:

existing_name = mapped_name

break
if existing_name:

raise RuntimeError(f"Dữ liệu khuôn mặt của bạn đang trùng với '{existing_name}'")

[Link](data_dir, exist_ok=True)

person_id = face_register_helper.get_next_person_id(data_dir, csv_path)

_log(f'Assigned fused identity ID={person_id} for name={face_name}')

frames = face_register_helper.capture_frames(

source=int(face_cam),

count=int(count),

delay=0.25,

timeout=int(timeout),

show=show,

if len(frames) == 0:

raise RuntimeError('No face frames captured')

x_new, y_new = face_register_helper.frames_to_embeddings(

frames,

detector,

backbone,

quality,

label=str(person_id),

if len(x_new) == 0:

raise RuntimeError('No face embeddings extracted from captured frames')

 duplicate_name, duplicate_similarity = face_register_helper.find_duplicate_identity(

x_new,

existing_x,

existing_y,

csv_path,

threshold=0.70,

if duplicate_name is not None:

raise RuntimeError(

f"Dữ liệu khuôn mặt của bạn đang trùng với '{duplicate_name}'
(similarity={duplicate_similarity:.4f})"

return {

'person_id': str(person_id),

'name': face_name,

'frames': frames,

'x_new': x_new,

'y_new': y_new,

def commit_face_enrollment(face_info, face_x_path: str, face_y_path: str, data_dir: str,

csv_path: str):

if face_register_helper is None:

raise RuntimeError(f'Cannot load face registration helper: {FACE_REGISTER_PATH}')

 save_dir = face_register_helper.save_frames_to_data_input(

face_info['frames'],

person_id=int(face_info['person_id']),

base_dir=data_dir,

face_register_helper.append_and_save(face_x_path, face_y_path, face_info['x_new'],

face_info['y_new'])

face_register_helper.append_to_dsdb(csv_path, int(face_info['person_id']),
face_info['name'])

return save_dir

def main():

global SCRFD, load_model_onnx, take_box_detector, alignment, process_kps,

process_onnx

parser = [Link](description='Enroll face + gait together using a shared

numeric ID')

parser.add_argument('--label', required=False, help='Full name for this identity (omit to

be prompted)')

parser.add_argument('--face-cam', type=int, default=0)

parser.add_argument('--face-count', type=int, default=20, help='Number of face images

to capture for registration')

parser.add_argument('--face-timeout', type=int, default=0, help='Timeout for face

capture, 0 means no timeout')

parser.add_argument('--gait-cams', nargs='+', type=str, default=['0','1'], help='Gait camera

ids (space-separated or comma-separated). Default uses cam 0 and cam 1.')

parser.add_argument('--gait-frames', type=int, default=45)

parser.add_argument('--gait-samples-per-cam', type=int, default=20, help='Number of

gait embeddings to store per camera from one capture')
 parser.add_argument('--gait-min-window-frames', type=int, default=20, help='Minimum
number of frames per gait sub-sequence used to extract one embedding')

parser.add_argument('--no-gui', action='store_true', help='Run without GUI windows

(headless).')

parser.add_argument('--face-x', type=str, default=[Link](DA_SRC, '[Link]'))

parser.add_argument('--face-y', type=str, default=[Link](DA_SRC, '[Link]'))

parser.add_argument('--face-data-dir', type=str, default=[Link](DA_SRC,

'data_input'))

parser.add_argument('--face-csv', type=str, default=[Link](DA_SRC, '[Link]'))

parser.add_argument('--gait-pkl', type=str, default=[Link](ROOT_DIR, 'OpenGait',

'output', 'CASIA-B', 'GaitGL', 'gait_system_db.pkl'))

parser.add_argument('--gait-x', type=str, default=[Link](ROOT_DIR, 'OpenGait',

'output', 'CASIA-B', 'GaitGL', 'gait_X.pkl'))

parser.add_argument('--gait-y', type=str, default=[Link](ROOT_DIR, 'OpenGait',

'output', 'CASIA-B', 'GaitGL', 'gait_y.pkl'))

parser.add_argument('--sync-out', type=str, default=[Link](THIS_DIR,

'synced_identities.pkl'))

parser.add_argument('--sync-threshold', type=float, default=0.55, help='Fallback cosine

threshold used when face/gait labels do not already match')

args = parser.parse_args()

_log('Starting main()')

# guard distributed master address to local to avoid remote socket attempts

[Link]('MASTER_ADDR', '[Link]')

[Link]('MASTER_PORT', '29500')

# init face models

if SCRFD is None or load_model_onnx is None or take_box_detector is None:

# debug info

print('--- Enroll import debug ---')

print('[Link] (prefixes):', [Link][:6])

cand_files = [

[Link](DA_SRC, 'utils', 'load_model.py'),

[Link](DA_SRC, 'load_model.py'),

[Link](DA_SRC, 'utils', 'process_data.py'),

[Link](DA_SRC, 'process_data.py'),

for p in cand_files:

print(p, 'exists=', [Link](p))

# Attempt fallback: load modules directly from file paths if present

if SCRFD is None or load_model_onnx is None:

for candidate in [

[Link](DA_SRC, 'utils', 'load_model.py'),

[Link](DA_SRC, 'load_model.py')

]:

if [Link](candidate):

lm_mod = load_module_from_path(candidate, 'load_model_fallback')

if lm_mod is not None:

SCRFD = getattr(lm_mod, 'SCRFD', SCRFD)

load_model_onnx = getattr(lm_mod, 'load_model_onnx', load_model_onnx)

print('Loaded load_model from', candidate)

break
 if take_box_detector is None:

for candidate in [

[Link](DA_SRC, 'utils', 'process_data.py'),

[Link](DA_SRC, 'process_data.py')

]:

if [Link](candidate):

pd_mod = load_module_from_path(candidate, 'process_data_fallback')

if pd_mod is not None:

take_box_detector = getattr(pd_mod, 'take_box_detector', take_box_detector)

alignment = getattr(pd_mod, 'alignment', alignment)

process_kps = getattr(pd_mod, 'process_kps', process_kps)

process_onnx = getattr(pd_mod, 'process_onnx', process_onnx)

print('Loaded process_data from', candidate)

break

if SCRFD is None or load_model_onnx is None or take_box_detector is None:

msg = 'Face model helpers not available or failed to import.'

if _import_err:

msg += '\nImport traceback:\n' + _import_err

msg += '\n\nSuggestions:\n'

msg += ' - Ensure DA_RobotGuide/src is on PYTHONPATH or in the repo root.\n'

msg += f" - Check these files exist: {[Link](DA_SRC,'utils','load_model.py')},

{[Link](DA_SRC,'utils','process_data.py')}\n"

msg += ' - Make sure required packages are installed in `daenv` (onnxruntime, torch,
numpy, opencv-python).\n'

raise RuntimeError(msg)

onnx_dir = [Link](DA_SRC, 'onnx')

 det_path = [Link](onnx_dir, 'scrfd_2.5g_bnkps.onnx')

bb_path = [Link](onnx_dir, '[Link]')

q_path = [Link](onnx_dir, '[Link]')

_log(f'Loading face detector: {det_path}')

detector = SCRFD(model_file=det_path)

[Link](0)

_log('Face detector ready')

_log(f'Loading backbone onnx: {bb_path}')

backbone = load_model_onnx(bb_path)

_log('Backbone loaded')

_log(f'Loading quality onnx: {q_path}')

quality = load_model_onnx(q_path)

_log('Quality model loaded')

# prepare optional gait runtime from fallback module (used only for local fallback
capture)

gait_runtime = None

if gait_load_fallback is not None and hasattr(gait_load_fallback, 'load_model'):

_log('Initializing gait runtime from fallback module (may load checkpoint)')

try:

model, eval_trfs = safe_load_gait_model([Link](ROOT_DIR, 'OpenGait',

'configs', 'gaitgl', '[Link]'))

gait_runtime = {'model': model, 'eval_trfs': eval_trfs, 'preprocess':

getattr(gait_load_fallback, 'preprocess_silhouette_frame', None), 'extract':
getattr(gait_load_fallback, 'extract_embedding_from_sequence', None)}

_log('Initialized gait_runtime from fallback module')

except Exception as e:
 _log('gait_runtime fallback init failed: ' + str(e))

gait_runtime = None

show = not bool(getattr(args, 'no_gui', False))

face_info = prepare_face_enrollment(

face_name=[Link],

face_cam=int(args.face_cam),

detector=detector,

backbone=backbone,

quality=quality,

face_x_path=args.face_x,

face_y_path=args.face_y,

data_dir=args.face_data_dir,

csv_path=args.face_csv,

count=int(args.face_count),

timeout=int(args.face_timeout),

show=show,

person_id = face_info['person_id']

_log(f'Prepared face enrollment for ID={person_id}, name={face_info["name"]}')

gait_cam_tokens = []

if isinstance(args.gait_cams, (list, tuple)):

for tok in args.gait_cams:

if tok is None:

continue
 if ',' in tok:

gait_cam_tokens.extend([[Link]() for t in [Link](',') if [Link]()])

else:

if [Link]():

gait_cam_tokens.append([Link]())

else:

gait_cam_tokens = [[Link]() for t in str(args.gait_cams).split(',') if [Link]()]

if len(gait_cam_tokens) < 1:

raise RuntimeError('Provide at least one gait cam')

gait_cams = [int(tok) for tok in gait_cam_tokens]

gait_enrolled = False

gait_added = 0

if gait_runtime is None:

raise RuntimeError('Gait runtime could not be initialized; aborting before committing

face enrollment')

try:

gait_added = enroll_gait_samples(

gait_cams=gait_cams,

person_id=person_id,

gait_runtime=gait_runtime,

gait_pkl_path=args.gait_pkl,

frames_required=int(args.gait_frames),

samples_per_cam=max(1, int(args.gait_samples_per_cam)),
 min_window_frames=max(5, int(args.gait_min_window_frames)),

show=show,

gait_enrolled = gait_added > 0

except Exception as e:

print('Multi-sample gait enrollment failed:', e)

traceback.print_exc()

raise

if not gait_enrolled:

raise RuntimeError('Gait enrollment did not complete successfully; face data was not
committed')

export_gait_xy_pickles(args.gait_pkl, args.gait_x, args.gait_y)

save_dir = commit_face_enrollment(face_info, args.face_x, args.face_y,

args.face_data_dir, args.face_csv)

_log(f'Committed face enrollment to {save_dir}')

# rebuild synced file using existing sync logic

# load fresh DBs

X, y = load_face_db(args.face_x, args.face_y)

gait_db, gait_labels = load_gait_db(args.gait_pkl)

face_keys, face_centroids, face_raw = group_centroids(X, y)

gait_keys, gait_centroids, gait_raw = ([], [], [])

if gait_db is not None and len(gait_labels) > 0:

gait_keys, gait_centroids, gait_raw = group_centroids(gait_db, gait_labels)

 synced = build_synced(face_keys, face_centroids, face_raw, gait_keys, gait_centroids,
gait_raw, threshold=float(args.sync_threshold))

# add meta

synced['meta'] = {

'created_at': __import__('datetime').[Link]().isoformat() + 'Z',

'method': 'enroll_script_shared_label_then_centroid',

'gait_embeddings_added': int(gait_added),

'gait_cameras': list(gait_cams),

'gait_samples_per_cam': max(1, int(args.gait_samples_per_cam)),

'sync_threshold': float(args.sync_threshold),

with open(args.sync_out, 'wb') as f:

[Link](synced, f)

print('Wrote updated sync file to', args.sync_out)

if __name__ == '__main__':

main()

import time

from typing import Any, Dict, List, Optional, Tuple

import numpy as np

import fm_core as core

from fm_core import l2_normalize, cosine_sim, normalize_identity_key,

bbox_transition_zone, transition_zone_score

class GlobalIdentityBank:
def __init__(self, sim_threshold: float = 0.72, name_map: Optional[Dict[str, str]] = None):

self.sim_threshold = sim_threshold

self.name_map = name_map or {}

[Link]: Dict[str, Dict[str, Any]] = {}

self.next_id = 1

def _resolve_display_name(self, label: Any) -> str:

return core.resolve_display_name(label, self.name_map)

def _ensure_meta(self, gid: str) -> Dict[str, Any]:

if gid not in [Link]:

[Link][gid] = {

"emb": None,

"count": 0,

"identity_key": None,

"display_name": None,

"label_source": None,

"label_votes": {},

"cam_last_seen": {},

"cam_last_boxes": {},

"cam_last_zones": {},

"last_cam": None,

"last_ts": 0.0,

"last_zone": None,

"last_assoc_reason": "new",

"last_assoc_score": 0.0,
 }

return [Link][gid]

def _new_id(self) -> str:

gid = f"G{self.next_id:04d}"

self.next_id += 1

return gid

def _find_best_gid(self, emb: Optional[[Link]]) -> Tuple[Optional[str], float]:

if emb is None:

return None, -1.0

best_gid = None

best_sim = -1.0

for gid, meta in [Link]():

ref = [Link]("emb")

if ref is None:

continue

s = cosine_sim(emb, ref)

if s > best_sim:

best_sim = s

best_gid = gid

return best_gid, best_sim

def _merge_votes(self, dst: Dict[str, int], src: Dict[str, int]):

for key, value in [Link]():

dst[key] = int([Link](key, 0)) + int(value)

 def _has_conflicting_identity(self, meta: Dict[str, Any], preferred_label: Optional[str]) ->
bool:

if not preferred_label or preferred_label == "unknown":

return False

identity_key = [Link]("identity_key")

if not identity_key:

return False

return str(identity_key) != normalize_identity_key(preferred_label)

def _promote_gid(self, old_gid: str, new_gid: str):

if old_gid == new_gid or old_gid not in [Link]:

return

old_meta = [Link](old_gid)

new_meta = self._ensure_meta(new_gid)

if old_meta.get("emb") is not None:

if new_meta.get("emb") is None:

new_meta["emb"] = old_meta["emb"]

new_meta["count"] = int(old_meta.get("count", 1))

else:

old_count = max(1, int(old_meta.get("count", 1)))

new_count = max(1, int(new_meta.get("count", 1)))

merged = ((old_meta["emb"] * old_count) + (new_meta["emb"] * new_count)) /

float(old_count + new_count)

new_meta["emb"] = l2_normalize([Link](1, -1), axis=1)[0]

 new_meta["count"] = old_count + new_count

self._merge_votes(new_meta.setdefault("label_votes", {}), old_meta.get("label_votes",

{}))

if old_meta.get("identity_key") and not new_meta.get("identity_key"):

new_meta["identity_key"] = old_meta.get("identity_key")

new_meta["display_name"] = old_meta.get("display_name")

new_meta["label_source"] = old_meta.get("label_source")

for cam_idx, ts in old_meta.get("cam_last_seen", {}).items():

prev_ts = float(new_meta.setdefault("cam_last_seen", {}).get(cam_idx, 0.0))

if float(ts) > prev_ts:

new_meta["cam_last_seen"][cam_idx] = float(ts)

old_box = old_meta.get("cam_last_boxes", {}).get(cam_idx)

old_zone = old_meta.get("cam_last_zones", {}).get(cam_idx)

if old_box is not None:

new_meta.setdefault("cam_last_boxes", {})[cam_idx] = tuple(map(int, old_box))

if old_zone is not None:

new_meta.setdefault("cam_last_zones", {})[cam_idx] = str(old_zone)

if float(old_meta.get("last_ts", 0.0)) > float(new_meta.get("last_ts", 0.0)):

new_meta["last_cam"] = old_meta.get("last_cam")

new_meta["last_ts"] = float(old_meta.get("last_ts", 0.0))

new_meta["last_zone"] = old_meta.get("last_zone")

new_meta["last_assoc_reason"] = old_meta.get("last_assoc_reason",
new_meta.get("last_assoc_reason", "promoted"))
 new_meta["last_assoc_score"] = float(old_meta.get("last_assoc_score",
new_meta.get("last_assoc_score", 0.0)))

def _find_cross_camera_match(

self,

emb: Optional[[Link]],

cam_idx: int,

bbox: Tuple[int, int, int, int],

frame_shape: Tuple[int, int, int],

now: float,

reid_threshold: float,

overlap_threshold: float,

active_window_sec: float,

min_transition_sec: float,

max_transition_sec: float,

transition_margin_ratio: float,

) -> Tuple[Optional[str], float, Optional[str]]:

if emb is None:

return None, -1.0, None

entry_zone = bbox_transition_zone(bbox, frame_shape,

margin_ratio=transition_margin_ratio)

best_gid = None

best_score = -1.0

best_reason = None

for gid, meta in [Link]():

ref = [Link]("emb")

if ref is None:

continue

other_cams = [

(int(other_cam), float(ts))

for other_cam, ts in [Link]("cam_last_seen", {}).items()

if int(other_cam) != int(cam_idx)

if not other_cams:

continue

sim = cosine_sim(emb, ref)

last_cam, last_seen = max(other_cams, key=lambda item: item[1])

time_gap = max(0.0, float(now) - float(last_seen))

last_zone = [Link]("cam_last_zones", {}).get(last_cam)

reason = None

gate_bonus = 0.0

if time_gap <= float(active_window_sec):

if sim < float(overlap_threshold):

continue

reason = "sync-overlap"

gate_bonus = 0.08

else:

if time_gap < float(min_transition_sec) or time_gap > float(max_transition_sec):

 continue

zone_score = transition_zone_score(last_zone, entry_zone)

if zone_score <= 0.0 or sim < float(reid_threshold):

continue

reason = "sync-transition"

gate_bonus = 0.03 + 0.05 * zone_score

if [Link]("identity_key"):

gate_bonus += 0.02

score = sim + gate_bonus

if score > best_score:

best_gid = gid

best_score = score

best_reason = reason

return best_gid, best_score, best_reason

def observe(

self,

gid: str,

cam_idx: int,

bbox: Tuple[int, int, int, int],

frame_shape: Tuple[int, int, int],

now: Optional[float] = None,

assoc_reason: Optional[str] = None,

 assoc_score: Optional[float] = None,

transition_margin_ratio: float = 0.16,

):

meta = self._ensure_meta(gid)

obs_ts = float(now if now is not None else [Link]())

zone = bbox_transition_zone(bbox, frame_shape, margin_ratio=transition_margin_ratio)

[Link]("cam_last_seen", {})[int(cam_idx)] = obs_ts

[Link]("cam_last_boxes", {})[int(cam_idx)] = tuple(map(int, bbox))

[Link]("cam_last_zones", {})[int(cam_idx)] = zone

meta["last_cam"] = int(cam_idx)

meta["last_ts"] = obs_ts

meta["last_zone"] = zone

if assoc_reason:

meta["last_assoc_reason"] = str(assoc_reason)

if assoc_score is not None:

meta["last_assoc_score"] = float(assoc_score)

def assign(

self,

emb: Optional[[Link]],

cam_idx: int,

bbox: Tuple[int, int, int, int],

frame_shape: Tuple[int, int, int],

preferred_label: Optional[str] = None,

sync_id: bool = True,

reid_threshold: float = 0.84,

 overlap_threshold: float = 0.90,

active_window_sec: float = 1.2,

min_transition_sec: float = 0.10,

max_transition_sec: float = 4.0,

transition_margin_ratio: float = 0.16,

now: Optional[float] = None,

) -> Tuple[str, str, float]:

obs_ts = float(now if now is not None else [Link]())

best_gid, best_sim = self._find_best_gid(emb)

# If we got a confident known label from face/gait, bind global ID to that label.

if preferred_label and preferred_label != "unknown":

gid = f"ID_{preferred_label}"

if (

best_gid is not None

and best_sim >= self.sim_threshold

and best_gid != gid

and not self._has_conflicting_identity([Link](best_gid, {}), preferred_label)

):

self._promote_gid(best_gid, gid)

meta = self._ensure_meta(gid)

meta["identity_key"] = normalize_identity_key(preferred_label)

meta["display_name"] = self._resolve_display_name(preferred_label)

meta["label_source"] = "face"

if emb is not None:

self._update(gid, emb)
 [Link](

gid,

cam_idx,

bbox,

frame_shape,

now=obs_ts,

assoc_reason="face",

assoc_score=best_sim if best_sim >= 0.0 else 1.0,

transition_margin_ratio=transition_margin_ratio,

return gid, "face", best_sim if best_sim >= 0.0 else 1.0

if emb is None:

gid = self._new_id()

[Link](

gid,

cam_idx,

bbox,

frame_shape,

now=obs_ts,

assoc_reason="new",

assoc_score=0.0,

transition_margin_ratio=transition_margin_ratio,

return gid, "new", 0.0

if sync_id:

matched_gid, matched_score, matched_reason = self._find_cross_camera_match(

emb=emb,

cam_idx=cam_idx,

bbox=bbox,

frame_shape=frame_shape,

now=obs_ts,

reid_threshold=reid_threshold,

overlap_threshold=overlap_threshold,

active_window_sec=active_window_sec,

min_transition_sec=min_transition_sec,

max_transition_sec=max_transition_sec,

transition_margin_ratio=transition_margin_ratio,

if matched_gid is not None and matched_reason is not None:

self._update(matched_gid, emb)

[Link](

matched_gid,

cam_idx,

bbox,

frame_shape,

now=obs_ts,

assoc_reason=matched_reason,

assoc_score=matched_score,

transition_margin_ratio=transition_margin_ratio,

)
 return matched_gid, matched_reason, matched_score

# If cross-camera sync didn't match, still attempt same-camera appearance

matching.

# This prevents duplicate GIDs/tracks for the same person appearing simultaneously

# in a single camera when tracking briefly creates two track IDs.

if best_gid is not None and best_sim >= self.sim_threshold:

# avoid promoting/merging if identity would conflict with a preferred label

if not self._has_conflicting_identity([Link](best_gid, {}), preferred_label):

self._update(best_gid, emb)

[Link](

best_gid,

cam_idx,

bbox,

frame_shape,

now=obs_ts,

assoc_reason="appearance",

assoc_score=best_sim,

transition_margin_ratio=transition_margin_ratio,

return best_gid, "appearance", best_sim

if not sync_id and best_gid is not None and best_sim >= self.sim_threshold:

self._update(best_gid, emb)

[Link](

best_gid,
 cam_idx,

bbox,

frame_shape,

now=obs_ts,

assoc_reason="appearance",

assoc_score=best_sim,

transition_margin_ratio=transition_margin_ratio,

return best_gid, "appearance", best_sim

gid = self._new_id()

[Link][gid] = {

"emb": [Link](emb, dtype=np.float32),

"count": 1,

"identity_key": None,

"display_name": None,

"label_source": None,

"label_votes": {},

"cam_last_seen": {},

"cam_last_boxes": {},

"cam_last_zones": {},

"last_cam": None,

"last_ts": 0.0,

"last_zone": None,

"last_assoc_reason": "new",

"last_assoc_score": 0.0,
 }

[Link](

gid,

cam_idx,

bbox,

frame_shape,

now=obs_ts,

assoc_reason="new",

assoc_score=0.0,

transition_margin_ratio=transition_margin_ratio,

return gid, "new", 0.0

def _update(self, gid: str, emb: [Link]):

emb = [Link](emb, dtype=np.float32)

meta = self._ensure_meta(gid)

if [Link]("emb") is None:

meta["emb"] = emb

meta["count"] = 1

return

c = max(1, int([Link]("count", 1)))

meta["emb"] = l2_normalize(((meta["emb"] * c + emb) / (c + 1)).reshape(1, -1), axis=1)[0]

meta["count"] = c + 1

def update_identity(

self,
 gid: str,

cam_idx: int,

face_label: str,

face_score: float,

gait_label: Optional[str],

gait_score: float,

fused_id: str,

fused_name: str,

fused_score: float,

face_thresh: float,

gait_thresh: float,

now: Optional[float] = None,

):

meta = self._ensure_meta(gid)

update_ts = float(now if now is not None else [Link]())

[Link]("cam_last_seen", {})[int(cam_idx)] = update_ts

if face_label != "unknown" and face_score >= face_thresh:

meta["identity_key"] = normalize_identity_key(face_label)

meta["display_name"] = fused_name if fused_name != "unknown" else

self._resolve_display_name(face_label)

meta["label_source"] = "face"

meta["label_votes"] = {meta["identity_key"]: max(3, int([Link]("label_votes",

{}).get(meta["identity_key"], 0)))}

return

if [Link]("label_source") == "face" and [Link]("identity_key"):

 return

candidate_key = None

candidate_name = None

if fused_id != "unknown" and fused_score >= max(gait_thresh + 0.05, 0.55):

candidate_key = normalize_identity_key(fused_id)

candidate_name = fused_name if fused_name != "unknown" else

self._resolve_display_name(fused_id)

elif gait_label and gait_label != "unknown" and gait_score >= gait_thresh + 0.08:

candidate_key = normalize_identity_key(gait_label)

candidate_name = self._resolve_display_name(gait_label)

if not candidate_key:

return

votes = [Link]("label_votes", {})

votes[candidate_key] = int([Link](candidate_key, 0)) + 1

recent_cams = [key for key, ts in [Link]("cam_last_seen", {}).items() if update_ts -

float(ts) <= 1.5]

required_votes = 4 if len(recent_cams) >= 2 else 3

if votes[candidate_key] >= required_votes:

meta["identity_key"] = candidate_key

meta["display_name"] = candidate_name

meta["label_source"] = "gait"

def get_identity(self, gid: str) -> Tuple[Optional[str], Optional[str], Optional[str]]:

meta = [Link](gid)
 if not meta:

return None, None, None

return [Link]("identity_key"), [Link]("display_name"), [Link]("label_source")

import os

import sys

import csv

import pickle

import threading

import time

from typing import Any, Dict, List, Optional, Tuple

import cv2

import numpy as np

# Paths and [Link] setup (same logic as original script)

THIS_DIR = [Link]([Link](__file__))

ROOT_DIR = [Link](THIS_DIR)

DA_SRC = [Link](ROOT_DIR, "DA_RobotGuide", "src")

DA_UTILS = [Link](DA_SRC, "utils")

OPEN_GAIT_ROOT = [Link](ROOT_DIR, "OpenGait")

OPEN_GAIT_MISC = [Link](OPEN_GAIT_ROOT, "misc")

OPEN_GAIT_PKG = [Link](OPEN_GAIT_ROOT, "opengait")

OPEN_GAIT_MTMCT = [Link](OPEN_GAIT_ROOT, "mtmct")

for p in (DA_SRC, DA_UTILS, OPEN_GAIT_ROOT, OPEN_GAIT_MISC, OPEN_GAIT_PKG,
OPEN_GAIT_MTMCT):

if p not in [Link]:

[Link](0, p)

def l2_normalize(x: [Link], axis=1, eps: float = 1e-12) -> [Link]:

x = [Link](x, dtype=np.float32)

n = [Link](x, axis=axis, keepdims=True)

n = [Link](n, eps)

return x / n

def cosine_top1(query: [Link], db: [Link]):

if db is None or len(db) == 0:

return -1, 0.0

q = l2_normalize([Link](1, -1), axis=1)[0]

sims = [Link](db, q)

idx = int([Link](sims))

return idx, float(sims[idx])

def cosine_topk(query: [Link], db: [Link], k: int = 2) -> List[Tuple[int, float]]:

if db is None or len(db) == 0:

return []

q = l2_normalize([Link](1, -1), axis=1)[0]

sims = [Link](db, q)

topk = min(max(1, int(k)), len(sims))

order = [Link](-sims)[:topk]
 return [(int(idx), float(sims[idx])) for idx in order]

def cosine_sim(a: [Link], b: [Link]) -> float:

a = [Link](a, dtype=np.float32)

b = [Link](b, dtype=np.float32)

na = [Link](a)

nb = [Link](b)

if na < 1e-12 or nb < 1e-12:

return 0.0

return float([Link](a, b) / (na * nb))

# DB loaders

def load_face_db():

x_path = [Link](DA_SRC, "[Link]")

y_path = [Link](DA_SRC, "[Link]")

if not [Link](x_path) or not [Link](y_path):

raise FileNotFoundError("Khong tim thay [Link]/[Link] trong DA_RobotGuide/src")

with open(x_path, "rb") as f:

x = [Link](f)

with open(y_path, "rb") as f:

y = [Link](f)

x = [Link](x, dtype=np.float32)

if [Link] == 1:

x = [Link](1, -1)
 x = l2_normalize(x, axis=1)

y = [str(v) for v in y]

return x, y

def load_gait_db(gait_path: Optional[str] = None):

if gait_path is None:

gait_path = [Link](OPEN_GAIT_ROOT, "output", "CASIA-B", "GaitGL",

"gait_system_db.pkl")

if not [Link](gait_path):

return None, []

with open(gait_path, "rb") as f:

data = [Link](f)

embs = [Link]("embeddings")

labels = [str(v) for v in [Link]("labels", [])]

if embs is None or len(labels) == 0:

return None, []

vecs = [Link](embs, dtype=np.float32).reshape(([Link][0], -1))

vecs = l2_normalize(vecs, axis=1)

return vecs, labels

def gait_embed_to_vec(g_emb: [Link]) -> [Link]:

arr = [Link](g_emb, dtype=np.float32)

if [Link] > 1:
 arr = [Link](-1)

arr = l2_normalize([Link](1, -1), axis=1)[0]

return arr

def normalize_identity_key(label: Any) -> str:

s = str(label).strip()

digits = "".join(ch for ch in s if [Link]())

if digits:

return str(int(digits))

return [Link]()

def load_name_map() -> Dict[str, str]:

csv_candidates = [

[Link](DA_SRC, "DSDB_DH2025.csv"),

[Link](DA_SRC, "[Link]"),

mapping: Dict[str, str] = {}

for p in csv_candidates:

if not [Link](p):

continue

try:

with open(p, "r", encoding="utf-8-sig", newline="") as f:

reader = [Link](f)

for row in reader:

stt = [Link]("STT")

name = [Link]("Họ và tên") or [Link]("Ho va ten") or [Link]("name")

 if stt is None or name is None:

continue

k = normalize_identity_key(stt)

if k:

mapping[k] = str(name).strip()

except Exception:

continue

if mapping:

break

return mapping

def resolve_display_name(label: Any, name_map: Dict[str, str], unknown: str = "unknown")

-> str:

if label is None:

return unknown

text = str(label).strip()

if not text or [Link]() == "unknown":

return unknown

return name_map.get(normalize_identity_key(text), text)

def configure_fullscreen_window(window_name: str):

[Link](window_name, cv2.WINDOW_NORMAL)

if hasattr(cv2, "WND_PROP_FULLSCREEN") and hasattr(cv2, "WINDOW_FULLSCREEN"):

try:

[Link](window_name, cv2.WND_PROP_FULLSCREEN,
cv2.WINDOW_FULLSCREEN)

return
 except Exception:

pass

if hasattr(cv2, "WND_PROP_TOPMOST"):

try:

[Link](window_name, cv2.WND_PROP_TOPMOST, 1)

except Exception:

pass

class AsyncCameraReader:

def __init__(self, source: Any, cam_index: int):

[Link] = source

self.cam_index = cam_index

[Link] = None

try:

[Link] = [Link](source)

except Exception:

[Link] = [Link](source)

if not [Link]():

raise RuntimeError(f"Khong mo duoc camera {[Link]}")

[Link] = [Link]()

self.latest_frame: Optional[[Link]] = None

self.latest_ts = 0.0

self.latest_seq = 0

[Link] = True
 [Link] = [Link](target=self._reader_loop, daemon=True,
name=f"FusionCam{cam_index}")

[Link]()

def _reader_loop(self):

bad_reads = 0

while [Link]:

ok, frame = [Link]()

if not ok or frame is None:

bad_reads += 1

if bad_reads % 60 == 0:

print(f"[WARN] cam={self.cam_index} source={[Link]!r} read fail

x{bad_reads}")

[Link](0.01)

continue

bad_reads = 0

with [Link]:

self.latest_frame = frame

self.latest_ts = [Link]()

self.latest_seq += 1

def get_latest(self) -> Tuple[Optional[[Link]], float, int]:

with [Link]:

if self.latest_frame is None:

return None, 0.0, 0

return self.latest_frame.copy(), self.latest_ts, self.latest_seq

 def stop(self):

[Link] = False

if [Link].is_alive():

[Link](timeout=0.5)

if [Link] is not None:

[Link]()

def collect_synchronized_frames(

readers: List[AsyncCameraReader],

max_delta_ms: int,

wait_ms: int,

) -> Tuple[List[Tuple[Optional[[Link]], float, int]], float]:

deadline = [Link]() + max(0.0, float(wait_ms) / 1000.0)

best_snapshot: Optional[List[Tuple[Optional[[Link]], float, int]]] = None

best_delta_ms = float("inf")

while True:

snapshot = [reader.get_latest() for reader in readers]

available = [item for item in snapshot if item[0] is not None]

if available:

if len(available) == 1:

delta_ms = 0.0

else:

timestamps = [item[1] for item in available]

delta_ms = (max(timestamps) - min(timestamps)) * 1000.0

 if delta_ms < best_delta_ms:

best_snapshot = snapshot

best_delta_ms = delta_ms

if delta_ms <= float(max_delta_ms):

return snapshot, delta_ms

if [Link]() >= deadline:

if best_snapshot is not None:

return best_snapshot, best_delta_ms

return snapshot, float("inf")

[Link](0.005)

def build_placeholder(shape: Tuple[int, int, int], text: str) -> [Link]:

vis = [Link](shape, dtype=np.uint8)

[Link](vis, text, (20, max(40, shape[0] // 2)), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0,

200, 255), 2)

return vis

def bbox_area(bbox: Tuple[int, int, int, int]) -> int:

x1, y1, x2, y2 = bbox

return max(0, x2 - x1) * max(0, y2 - y1)

def bbox_iou(box_a: Tuple[int, int, int, int], box_b: Tuple[int, int, int, int]) -> float:

ax1, ay1, ax2, ay2 = box_a

bx1, by1, bx2, by2 = box_b

ix1 = max(ax1, bx1)

 iy1 = max(ay1, by1)

ix2 = min(ax2, bx2)

iy2 = min(ay2, by2)

iw = max(0, ix2 - ix1)

ih = max(0, iy2 - iy1)

inter = iw * ih

union = bbox_area(box_a) + bbox_area(box_b) - inter

if union <= 0:

return 0.0

return float(inter) / float(union)

def bbox_center_distance(box_a: Tuple[int, int, int, int], box_b: Tuple[int, int, int, int]) ->
float:

ax1, ay1, ax2, ay2 = box_a

bx1, by1, bx2, by2 = box_b

acx = 0.5 * (ax1 + ax2)

acy = 0.5 * (ay1 + ay2)

bcx = 0.5 * (bx1 + bx2)

bcy = 0.5 * (by1 + by2)

return float(((acx - bcx) ** 2 + (acy - bcy) ** 2) ** 0.5)

def bbox_transition_zone(

bbox: Tuple[int, int, int, int],

frame_shape: Tuple[int, int, int],

margin_ratio: float = 0.16,

) -> str:
 frame_h, frame_w = frame_shape[:2]

if frame_h <= 0 or frame_w <= 0:

return "center"

x1, y1, x2, y2 = bbox

margin_x = max(8, int(frame_w * float(margin_ratio)))

margin_y = max(8, int(frame_h * float(margin_ratio)))

touches = []

if x1 <= margin_x:

[Link](("left", margin_x - x1))

if x2 >= frame_w - margin_x:

[Link](("right", x2 - (frame_w - margin_x)))

if y1 <= margin_y:

[Link](("top", margin_y - y1))

if y2 >= frame_h - margin_y:

[Link](("bottom", y2 - (frame_h - margin_y)))

if not touches:

return "center"

return max(touches, key=lambda item: item[1])[0]

def transition_zone_score(prev_zone: Optional[str], next_zone: Optional[str]) -> float:

if not prev_zone or not next_zone:

return 0.0

if prev_zone == "center" or next_zone == "center":

 return 0.0

if prev_zone == next_zone:

return 0.55

opposite_pairs = {

("left", "right"),

("right", "left"),

("top", "bottom"),

("bottom", "top"),

if (prev_zone, next_zone) in opposite_pairs:

return 1.0

horizontal = {"left", "right"}

vertical = {"top", "bottom"}

if prev_zone in horizontal and next_zone in horizontal:

return 0.7

if prev_zone in vertical and next_zone in vertical:

return 0.7

return 0.35

def non_max_suppress_detections(detections: List[Dict[str, Any]], iou_thresh: float = 0.65)

-> List[Dict[str, Any]]:

kept: List[Dict[str, Any]] = []

for det in sorted(detections, key=lambda item: float([Link]("conf", 0.0)), reverse=True):

bbox = tuple(map(int, det["bbox"]))

 if any(bbox_iou(bbox, tuple(map(int, prev["bbox"]))) >= iou_thresh for prev in kept):

continue

[Link](det)

return kept

def match_track_to_detection_bbox(

track_bbox: Tuple[int, int, int, int],

detections: List[Dict[str, Any]],

min_iou: float = 0.15,

max_center_ratio: float = 0.75,

) -> Optional[Tuple[int, int, int, int]]:

best_bbox = None

best_score = None

track_w = max(1, int(track_bbox[2] - track_bbox[0]))

track_h = max(1, int(track_bbox[3] - track_bbox[1]))

max_center_dist = max(60.0, max(track_w, track_h) * float(max_center_ratio))

for det in detections:

dbbox = tuple(map(int, det["bbox"]))

center_dist = bbox_center_distance(track_bbox, dbbox)

overlap = bbox_iou(track_bbox, dbbox)

if overlap < min_iou and center_dist > max_center_dist:

continue

score = (-overlap, center_dist)

if best_score is None or score < best_score:

best_score = score
 best_bbox = dbbox

return best_bbox

import os

import time

from collections import deque

from typing import Any, Dict, List, Optional, Tuple

import cv2

import numpy as np

# ensure core paths are set

import fm_core as core

# import model helpers (these modules are on [Link] due to fm_core)

from load_model import SCRFD, load_model_onnx

from process_data import take_box_detector, alignment, process_kps, process_onnx

def build_face_models():

onnx_dir = [Link](core.DA_SRC, "onnx")

det_path = [Link](onnx_dir, "scrfd_2.5g_bnkps.onnx")

bb_path = [Link](onnx_dir, "[Link]")

q_path = [Link](onnx_dir, "[Link]")

for p in (det_path, bb_path, q_path):

 if not [Link](p):

raise FileNotFoundError(f"Thieu model: {p}")

detector = SCRFD(model_file=det_path)

[Link](0)

backbone = load_model_onnx(bb_path)

quality = load_model_onnx(q_path)

return detector, backbone, quality

def pick_face_in_person(person_crop: [Link], detector, backbone, quality,

min_face_size: int = 28):

if person_crop is None or person_crop.size == 0:

return None, None, None

h, w = person_crop.shape[:2]

scales = [1.0]

if min(h, w) < 320:

[Link]([1.5, 2.0])

chosen_bbox = None

chosen_kps = None

chosen_score = None

for scale in scales:

candidate = person_crop

if scale != 1.0:
 try:

candidate = [Link](person_crop, (int(round(w * scale)), int(round(h * scale))))

except Exception:

candidate = person_crop

bbs, kpss = take_box_detector(candidate, detector)

if bbs is None or len(bbs) == 0:

continue

i = max(range([Link][0]), key=lambda idx: float(bbs[idx][4]))

x1s, y1s, x2s, y2s, det_score = bbs[i].astype(np.float32)

x1 = int(max(0, round(x1s / scale)))

y1 = int(max(0, round(y1s / scale)))

x2 = int(min(w - 1, round(x2s / scale)))

y2 = int(min(h - 1, round(y2s / scale)))

if x2 <= x1 or y2 <= y1:

continue

if min(x2 - x1, y2 - y1) < int(min_face_size):

continue

kps = [Link](kpss[i], dtype=np.float32) / float(scale)

chosen_bbox = (x1, y1, x2, y2)

chosen_kps = kps

chosen_score = float(det_score)

break

if chosen_bbox is None or chosen_kps is None:

return None, None, None

 x1, y1, x2, y2 = chosen_bbox

face_crop = person_crop[y1:y2, x1:x2]

if face_crop.size == 0:

return None, None, None

try:

_, _, _, _, _, _, _, _, l_eye, r_eye = process_kps(chosen_kps)

l_eye_local = [Link](l_eye, dtype=np.float32) - [Link]([x1, y1],

dtype=np.float32)

r_eye_local = [Link](r_eye, dtype=np.float32) - [Link]([x1, y1],

dtype=np.float32)

aligned = alignment(face_crop, l_eye_local, r_eye_local)

aligned = [Link](aligned, (112, 112))

q, emb_t = process_onnx(aligned, backbone, quality)

emb = emb_t.cpu().detach().numpy().astype(np.float32)[0]

emb = core.l2_normalize([Link](1, -1), axis=1)[0]

score_q = float(q[0]) if q is not None and len(q) > 0 else 0.0

return chosen_bbox, emb, score_q

except Exception:

return None, None, None

def extract_face_embedding_from_bbox(

frame: [Link],

bbox: Tuple[int, int, int, int],

kps: [Link],

backbone,

quality,

):
 x1, y1, x2, y2 = bbox

face_crop = frame[y1:y2, x1:x2]

if face_crop.size == 0:

return None, 0.0

try:

_, _, _, _, _, _, _, _, l_eye, r_eye = process_kps(kps)

l_eye_local = [Link](l_eye, dtype=np.float32) - [Link]([x1, y1],

dtype=np.float32)

r_eye_local = [Link](r_eye, dtype=np.float32) - [Link]([x1, y1],

dtype=np.float32)

aligned = alignment(face_crop, l_eye_local, r_eye_local)

aligned = [Link](aligned, (112, 112))

q, emb_t = process_onnx(aligned, backbone, quality)

emb = emb_t.cpu().detach().numpy().astype(np.float32)[0]

emb = core.l2_normalize([Link](1, -1), axis=1)[0]

score_q = float(q[0]) if q is not None and len(q) > 0 else 0.0

return emb, score_q

except Exception:

return None, 0.0

def detect_faces_in_frame(

frame: [Link],

detector,

backbone,

quality,

min_face_size: int = 56,

) -> List[Dict[str, Any]]:

bbs, kpss = take_box_detector(frame, detector)

if bbs is None or len(bbs) == 0:

return []

faces: List[Dict[str, Any]] = []

frame_h, frame_w = [Link][:2]

for idx in range([Link][0]):

x1, y1, x2, y2, det_score = bbs[idx].astype(int)

x1 = max(0, x1)

y1 = max(0, y1)

x2 = min(frame_w - 1, x2)

y2 = min(frame_h - 1, y2)

if x2 <= x1 or y2 <= y1:

continue

if min(x2 - x1, y2 - y1) < min_face_size:

continue

emb, face_quality = extract_face_embedding_from_bbox(

frame,

(x1, y1, x2, y2),

kpss[idx],

backbone,

quality,

if emb is None:

continue
 [Link](

"bbox": (x1, y1, x2, y2),

"emb": emb,

"det_score": float(det_score),

"quality": float(face_quality),

return faces

def face_overlap_ratio(face_bbox: Tuple[int, int, int, int], body_bbox: Tuple[int, int, int, int]) -
> float:

fx1, fy1, fx2, fy2 = face_bbox

bx1, by1, bx2, by2 = body_bbox

ix1 = max(fx1, bx1)

iy1 = max(fy1, by1)

ix2 = min(fx2, bx2)

iy2 = min(fy2, by2)

iw = max(0, ix2 - ix1)

ih = max(0, iy2 - iy1)

inter = iw * ih

face_area = max(1, (fx2 - fx1) * (fy2 - fy1))

return float(inter) / float(face_area)

def choose_stable_face_identity(

history_deque: List[Tuple[str, float, float]],

 face_thresh: float,

min_consistent_frames: int = 3,

) -> Tuple[Optional[str], float]:

"""Decide a stable face label from recent (label, score, ts) entries.

Returns (label, avg_score) or (None, 0.0)

"""

if not history_deque:

return None, 0.0

counts: Dict[str, List[float]] = {}

for lbl, score, _ts in history_deque:

[Link](lbl, []).append(float(score))

# prefer non-'unknown' labels

candidates = [(lbl, scores) for lbl, scores in [Link]() if lbl != "unknown"]

if not candidates:

return None, 0.0

# rank by number of occurrences then mean score

[Link](key=lambda it: (len(it[1]), float([Link](it[1]))), reverse=True)

best_lbl, best_scores = candidates[0]

if len(best_scores) < int(min_consistent_frames):

return None, 0.0

avg_score = float([Link](best_scores))

# apply a slightly relaxed threshold for persistence

if avg_score >= max(0.0, float(face_thresh) * 0.85):

 return best_lbl, avg_score

return None, 0.0

import os

import socket

from typing import Any, Dict, List, Optional, Tuple

import cv2

import numpy as np

import fm_core as core

def _pick_free_local_port() -> int:

s = [Link](socket.AF_INET, socket.SOCK_STREAM)

try:

[Link](("[Link]", 0))

return int([Link]()[1])

finally:

try:

[Link]()

except Exception:

pass

def try_init_gait_runtime(cfg_path: str, dist_port: int = 0):

helper_module = None

try:
 import webcam_enroll as helper_module # type: ignore

except Exception:

try:

import webcam_recognize as helper_module # type: ignore

except Exception as e:

print("[WARN] Khong import duoc OpenGait runtime:", e)

return None

gait_load_model = getattr(helper_module, "load_model", None)

preprocess_silhouette_frame = getattr(helper_module, "preprocess_silhouette_frame",

None)

extract_embedding_from_sequence = getattr(helper_module,
"extract_embedding_from_sequence", None)

if not callable(gait_load_model) or not callable(preprocess_silhouette_frame) or not

callable(extract_embedding_from_sequence):

print("[WARN] OpenGait helper module thieu runtime functions")

return None

# Avoid distributed port conflicts with any running OpenGait training.

# Also guard against a dirty environment (e.g. MASTER_ADDR set by other tools).

[Link]["MASTER_ADDR"] = "[Link]"

attempts = 5

last_exc: Optional[BaseException] = None

for _ in range(attempts):

port = None

try:
 port = int(dist_port)

except Exception:

port = 0

if port <= 0:

port = _pick_free_local_port()

[Link]["OPENGAIT_DIST_PORT"] = str(int(port))

[Link]["MASTER_PORT"] = str(int(port))

try:

model, eval_trfs = gait_load_model(cfg_path)

return {

"model": model,

"eval_trfs": eval_trfs,

"preprocess": preprocess_silhouette_frame,

"extract": extract_embedding_from_sequence,

"build_candidates": getattr(helper_module, "build_silhouette_candidates", None),

"choose_candidate": getattr(helper_module, "choose_silhouette_candidate",

None),

"quality_eval": getattr(helper_module, "evaluate_silhouette_quality", None),

except Exception as e:

last_exc = e

msg = str(e).lower()

# Common when port is already in use (WinError 10048).

if "address already in use" in msg or "10048" in msg or "errno 98" in msg:

 continue

break

print("[WARN] Khoi tao OpenGait that bai:", last_exc)

return None

def refine_person_bbox_from_fg(

fg_mask: [Link],

bbox: Tuple[int, int, int, int],

frame_shape: Tuple[int, int, int],

min_contour_ratio: float,

) -> Tuple[Tuple[int, int, int, int], float, bool]:

x1, y1, x2, y2 = bbox

if x2 <= x1 or y2 <= y1:

return bbox, 0.0, False

fg_crop = fg_mask[y1:y2, x1:x2]

if fg_crop.size == 0:

return bbox, 0.0, False

_, fg_bin = [Link](fg_crop, 127, 255, cv2.THRESH_BINARY)

kernel = [Link]((3, 3), dtype=np.uint8)

fg_bin = [Link](fg_bin, cv2.MORPH_OPEN, kernel, iterations=1)

fg_bin = [Link](fg_bin, cv2.MORPH_CLOSE, kernel, iterations=2)

 contours, _ = [Link](fg_bin, cv2.RETR_EXTERNAL,
cv2.CHAIN_APPROX_SIMPLE)

if not contours:

return bbox, 0.0, False

largest = max(contours, key=[Link])

contour_area = float([Link](largest))

box_area = float(max(1, (x2 - x1) * (y2 - y1)))

contour_ratio = contour_area / box_area

rx, ry, rw, rh = [Link](largest)

# If the detection bbox is too large, contour_ratio can be very small even when

# the contour clearly corresponds to a person. In that case, allow refinement

# based on the contour bbox size/aspect instead of ratio alone.

if contour_ratio < float(min_contour_ratio):

try:

bh = float(max(1, (y2 - y1)))

bw = float(max(1, (x2 - x1)))

rect_area = float(max(1, rw * rh))

rect_h_ok = float(rh) >= max(56.0, 0.22 * bh)

rect_w_ok = float(rw) >= max(18.0, 0.08 * bw)

rect_aspect_ok = float(rh) / float(max(1, rw)) >= 0.95

rect_area_ok = rect_area >= max(600.0, 0.003 * box_area)

if not (rect_h_ok and rect_w_ok and rect_aspect_ok and rect_area_ok):

return bbox, contour_ratio, False

except Exception:
 return bbox, contour_ratio, False

pad_x = max(2, int(rw * 0.08))

pad_y = max(2, int(rh * 0.06))

nx1 = max(0, x1 + rx - pad_x)

ny1 = max(0, y1 + ry - pad_y)

nx2 = min(frame_shape[1] - 1, x1 + rx + rw + pad_x)

ny2 = min(frame_shape[0] - 1, y1 + ry + rh + pad_y)

refined = (int(nx1), int(ny1), int(nx2), int(ny2))

return refined, contour_ratio, True

def extract_refined_gait_silhouette(

frame: [Link],

fg_mask: [Link],

bbox: Tuple[int, int, int, int],

gait_runtime: Dict[str, Any],

min_contour_ratio: float,

) -> Tuple[Optional[[Link]], Tuple[int, int, int, int], float, Optional[str]]:

gait_bbox, contour_ratio, refined_ok = refine_person_bbox_from_fg(

fg_mask,

bbox,

[Link],

min_contour_ratio=min_contour_ratio,

active_bbox = gait_bbox if refined_ok else bbox

x1, y1, x2, y2 = map(int, active_bbox)

if x2 <= x1 or y2 <= y1:

return None, active_bbox, contour_ratio, None

person_crop = frame[y1:y2, x1:x2]

fg_crop = fg_mask[y1:y2, x1:x2]

if person_crop.size == 0 or fg_crop.size == 0:

return None, active_bbox, contour_ratio, None

build_candidates = gait_runtime.get("build_candidates")

choose_candidate = gait_runtime.get("choose_candidate")

quality_eval = gait_runtime.get("quality_eval")

selected_mode = None

sil_mask = None

def _quality_ok(mask: [Link]) -> bool:

if mask is None or [Link] == 0:

return False

if callable(quality_eval):

try:

return bool(quality_eval(mask))

except Exception:

return False

nz = int(np.count_nonzero(mask))

ratio = nz / float(max(1, int([Link])))

return 0.03 <= ratio <= 0.95

 def _keep_largest(mask: [Link]) -> [Link]:

try:

contours, _ = [Link](mask, cv2.RETR_EXTERNAL,

cv2.CHAIN_APPROX_SIMPLE)

if not contours:

return mask

largest = max(contours, key=[Link])

out = np.zeros_like(mask)

[Link](out, [largest], 0, 255, -1)

return out

except Exception:

return mask

def _try_otsu_silhouette() -> Tuple[Optional[[Link]], Optional[str]]:

try:

gray = [Link](person_crop, cv2.COLOR_BGR2GRAY)

_, fg_bin_local = [Link](fg_crop, 127, 255, cv2.THRESH_BINARY)

_, m = [Link](gray, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)

candidates = [(m, "otsu"), (cv2.bitwise_not(m), "otsu_inv")]

kernel = [Link]((3, 3), dtype=np.uint8)

scored: List[Tuple[float, [Link], str]] = []

for cand, mode in candidates:

cand = [Link](cand, cv2.MORPH_CLOSE, kernel, iterations=1)

cand = [Link](cand, cv2.MORPH_OPEN, kernel, iterations=1)

cand = _keep_largest(cand)

if not _quality_ok(cand):
 continue

denom = float(max(1, int(np.count_nonzero(cand))))

overlap = float(np.count_nonzero(cv2.bitwise_and(cand, fg_bin_local))) / denom

[Link]((overlap, cand, mode))

if not scored:

return None, None

[Link](key=lambda t: t[0], reverse=True)

return scored[0][1], scored[0][2]

except Exception:

return None, None

if callable(build_candidates) and callable(choose_candidate):

candidates = build_candidates(person_crop, fg_crop)

# Prefer cleaned candidate to reduce bgsub noise.

sil_mask, selected_mode = choose_candidate(candidates, preferred_mode="clean")

if sil_mask is None:

_, sil_mask = [Link](fg_crop, 127, 255, cv2.THRESH_BINARY)

if sil_mask is None or sil_mask.size == 0:

alt, alt_mode = _try_otsu_silhouette()

if alt is not None:

sil_mask = alt

selected_mode = alt_mode

if sil_mask is None or sil_mask.size == 0:

 return None, active_bbox, contour_ratio, selected_mode

# Reject cases where silhouette contains multiple similarly-sized contours

try:

contours, _ = [Link](sil_mask.copy(), cv2.RETR_EXTERNAL,

cv2.CHAIN_APPROX_SIMPLE)

if contours and len(contours) > 1:

areas = sorted([float([Link](c)) for c in contours], reverse=True)

if len(areas) > 1:

if areas[1] / max(areas[0], 1.0) > 0.35:

return None, active_bbox, contour_ratio, selected_mode

box_area = float(max(1, fg_crop.shape[0] * fg_crop.shape[1]))

if areas[1] / box_area > 0.08:

return None, active_bbox, contour_ratio, selected_mode

except Exception:

pass

if not _quality_ok(sil_mask):

alt, alt_mode = _try_otsu_silhouette()

if alt is None:

return None, active_bbox, contour_ratio, selected_mode

sil_mask = alt

selected_mode = alt_mode

# Pad silhouette to a stable gait aspect ratio before calling preprocess

try:

target_h, target_w = (64, 44)

 cur_h, cur_w = sil_mask.shape[:2]

cur_aspect = float(cur_h) / float(max(1, cur_w))

target_aspect = float(target_h) / float(target_w)

if abs(cur_aspect - target_aspect) > 1e-3:

if cur_aspect < target_aspect:

desired_h = int(round(cur_w * target_aspect))

pad = max(0, desired_h - cur_h)

pad_top = pad // 2

pad_bottom = pad - pad_top

sil_mask = [Link](sil_mask, ((pad_top, pad_bottom), (0, 0)), mode="constant",

constant_values=0)

else:

desired_w = int(round(cur_h / target_aspect))

pad = max(0, desired_w - cur_w)

pad_left = pad // 2

pad_right = pad - pad_left

sil_mask = [Link](sil_mask, ((0, 0), (pad_left, pad_right)), mode="constant",

constant_values=0)

except Exception:

pass

preprocess = gait_runtime.get("preprocess")

if not callable(preprocess):

return None, active_bbox, contour_ratio, selected_mode

try:

sil_norm = preprocess(sil_mask)

except Exception:
 sil_norm = None

return sil_norm, active_bbox, contour_ratio, selected_mode

def choose_stable_gait_identity(

history_items: List[Dict[str, Any]],

gait_thresh: float,

gait_vote_min: int,

gait_margin_min: float,

gait_vote_margin: int,

) -> Tuple[Optional[str], float]:

label_votes: Dict[str, int] = {}

label_scores: Dict[str, List[float]] = {}

for item in history_items:

score = float([Link]("cos_sim", 0.0))

margin = float([Link]("margin", 0.0))

if score < float(gait_thresh) or margin < float(gait_margin_min):

continue

label_id = str([Link]("label_id", "unknown"))

if label_id == "unknown":

continue

label_votes[label_id] = int(label_votes.get(label_id, 0)) + 1

label_scores.setdefault(label_id, []).append(score)

if not label_votes:

return None, 0.0

 ordered_labels = sorted(

label_votes.keys(),

key=lambda key: (

label_votes[key],

max(label_scores.get(key, [0.0])),

sum(sorted(label_scores.get(key, [0.0]), reverse=True)[:3]) / float(max(1, min(3,

len(label_scores.get(key, []))))),

),

reverse=True,

stable_label = ordered_labels[0]

stable_votes = int(label_votes.get(stable_label, 0))

second_votes = int(label_votes.get(ordered_labels[1], 0)) if len(ordered_labels) > 1 else 0

if stable_votes < max(3, int(gait_vote_min)):

return None, 0.0

if stable_votes < second_votes + max(1, int(gait_vote_margin)):

return None, 0.0

top_scores = sorted(label_scores.get(stable_label, [0.0]), reverse=True)[:3]

stable_score = float(sum(top_scores) / float(max(1, len(top_scores))))

return stable_label, stable_score

import argparse

import csv

import os
import pickle

import sys

import threading

import time

from collections import deque

from typing import Any, Dict, List, Optional, Tuple

import cv2

import numpy as np

try:

import torch

import [Link] as nn

except Exception:

torch = None

nn = None

try:

from ultralytics import YOLO

except Exception:

YOLO = None

import fm_core as core

from fm_core import (

l2_normalize,
 cosine_top1,

cosine_topk,

cosine_sim,

load_face_db,

load_gait_db,

gait_embed_to_vec,

normalize_identity_key,

load_name_map,

resolve_display_name,

configure_fullscreen_window,

AsyncCameraReader,

collect_synchronized_frames,

build_placeholder,

bbox_area,

bbox_iou,

bbox_center_distance,

bbox_transition_zone,

transition_zone_score,

non_max_suppress_detections,

match_track_to_detection_bbox,

from load_model import SCRFD, load_model_onnx

from process_data import take_box_detector, alignment, process_kps, process_onnx

from tracker_wrapper import TrackerWrapper

try:
 from reid_encoder import ReIDEncoder

except Exception:

ReIDEncoder = None

from fm_gait import try_init_gait_runtime, extract_refined_gait_silhouette,

choose_stable_gait_identity

from fm_face import (

build_face_models,

pick_face_in_person,

extract_face_embedding_from_bbox,

detect_faces_in_frame,

face_overlap_ratio,

choose_stable_face_identity,

if nn is not None:

class AlphaFusionMLP([Link]):

def __init__(self, in_dim: int = 6, hidden_dim: int = 16):

super().__init__()

[Link] = [Link](

[Link](in_dim, hidden_dim),

[Link](inplace=True),
 [Link](hidden_dim, 1),

[Link](),

def forward(self, x: "[Link]") -> "[Link]":

return [Link](x)

else:

AlphaFusionMLP = None

def try_load_fusion_mlp(path: Optional[str]):

"""Load an alpha-prediction MLP.

Expected: a PyTorch state_dict saved from AlphaFusionMLP(in_dim=6) or a

TorchScript module that returns alpha in [0,1].

"""

if not path:

return None

if torch is None or nn is None:

print("[WARN] torch is not available; cannot load fusion MLP")

return None

if not [Link](path):

print(f"[WARN] fusion MLP not found: {path}")

return None

try:
 obj = [Link](path, map_location="cpu")

[Link]()

return obj

except Exception:

pass

try:

state = [Link](path, map_location="cpu")

if isinstance(state, dict) and "state_dict" in state and isinstance(state["state_dict"],

dict):

state = state["state_dict"]

model = AlphaFusionMLP(in_dim=6, hidden_dim=16)

model.load_state_dict(state, strict=False)

[Link]()

return model

except Exception as e:

print(f"[WARN] failed to load fusion MLP: {e}")

return None

def compute_context_alpha(

base_alpha: float,

face_label: str,

face_score: float,

face_thresh: float,

face_area_ratio: float,
 stable_face: bool,

contour_ratio: float,

min_contour_ratio: float,

gait_mode: Optional[str],

) -> float:

"""Heuristic, context-aware alpha in [0.15, 0.98]."""

a = float(base_alpha)

# If face is missing/weak, let gait contribute more.

if face_label == "unknown" or float(face_score) < float(face_thresh) * 0.85:

a *= 0.55

# If gait silhouette quality is low, lean more on face.

try:

if str(gait_mode or "").lower() == "bgsub":

a += 0.10

if float(contour_ratio) < float(min_contour_ratio) * 1.25:

a += 0.12

except Exception:

pass

# Larger faces -> face is more reliable.

if float(face_area_ratio) >= 0.08:

a += 0.10

elif float(face_area_ratio) > 0.0 and float(face_area_ratio) <= 0.02:

a -= 0.08
 # Stable face over recent frames -> trust face more.

if stable_face:

a += 0.10

return float([Link](a, 0.15, 0.98))

def parse_cameras(camera_arg: str) -> List[Any]:

cams: List[Any] = []

for tok in [[Link]() for x in camera_arg.split(",") if [Link]()]:

if [Link]():

[Link](int(tok))

else:

[Link](tok)

return cams if cams else [0]

def normalize_video_source(source: Any) -> Any:

if isinstance(source, str):

source = [Link]()

if [Link]():

return int(source)

return source
def open_camera_source(source: Any) -> [Link]:

source = normalize_video_source(source)

backends: List[Tuple[str, Optional[int]]] = []

if isinstance(source, int):

if [Link] == "nt" and hasattr(cv2, "CAP_DSHOW"):

[Link](("DirectShow", cv2.CAP_DSHOW))

if [Link] == "nt" and hasattr(cv2, "CAP_MSMF"):

[Link](("MSMF", cv2.CAP_MSMF))

[Link](("default", None))

else:

if isinstance(source, str) and "://" in source and hasattr(cv2, "CAP_FFMPEG"):

[Link](("FFMPEG", cv2.CAP_FFMPEG))

[Link](("default", None))

for backend_name, backend in backends:

cap = [Link](source, backend) if backend is not None else

[Link](source)

if [Link]():

try:

[Link](cv2.CAP_PROP_BUFFERSIZE, 1)

except Exception:

pass

print(f"[INFO] Opened camera {source!r} with {backend_name}")

return cap

[Link]()
 return [Link](source)

class AsyncCameraReader:

def __init__(self, source: Any, cam_index: int):

[Link] = normalize_video_source(source)

self.cam_index = cam_index

[Link] = open_camera_source([Link])

if not [Link]():

raise RuntimeError(f"Khong mo duoc camera {[Link]}")

[Link] = [Link]()

self.latest_frame: Optional[[Link]] = None

self.latest_ts = 0.0

self.latest_seq = 0

[Link] = True

[Link] = [Link](target=self._reader_loop, daemon=True,

name=f"FusionCam{cam_index}")

[Link]()

def _reader_loop(self):

bad_reads = 0

while [Link]:

ok, frame = [Link]()

if not ok or frame is None:

 bad_reads += 1

if bad_reads % 60 == 0:

print(f"[WARN] cam={self.cam_index} source={[Link]!r} read fail

x{bad_reads}")

[Link](0.01)

continue

bad_reads = 0

with [Link]:

self.latest_frame = frame

self.latest_ts = [Link]()

self.latest_seq += 1

def get_latest(self) -> Tuple[Optional[[Link]], float, int]:

with [Link]:

if self.latest_frame is None:

return None, 0.0, 0

return self.latest_frame.copy(), self.latest_ts, self.latest_seq

def stop(self):

[Link] = False

if [Link].is_alive():

[Link](timeout=0.5)

[Link]()
def collect_synchronized_frames(

readers: List[AsyncCameraReader],

max_delta_ms: int,

wait_ms: int,

) -> Tuple[List[Tuple[Optional[[Link]], float, int]], float]:

deadline = [Link]() + max(0.0, float(wait_ms) / 1000.0)

best_snapshot: Optional[List[Tuple[Optional[[Link]], float, int]]] = None

best_delta_ms = float("inf")

while True:

snapshot = [reader.get_latest() for reader in readers]

available = [item for item in snapshot if item[0] is not None]

if available:

if len(available) == 1:

delta_ms = 0.0

else:

timestamps = [item[1] for item in available]

delta_ms = (max(timestamps) - min(timestamps)) * 1000.0

if delta_ms < best_delta_ms:

best_snapshot = snapshot

best_delta_ms = delta_ms

if delta_ms <= float(max_delta_ms):

return snapshot, delta_ms

if [Link]() >= deadline:

if best_snapshot is not None:

 return best_snapshot, best_delta_ms

return snapshot, float("inf")

[Link](0.005)

def build_placeholder(shape: Tuple[int, int, int], text: str) -> [Link]:

vis = [Link](shape, dtype=np.uint8)

[Link](vis, text, (20, max(40, shape[0] // 2)), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0,

200, 255), 2)

return vis

def bbox_area(bbox: Tuple[int, int, int, int]) -> int:

x1, y1, x2, y2 = bbox

return max(0, x2 - x1) * max(0, y2 - y1)

def bbox_iou(box_a: Tuple[int, int, int, int], box_b: Tuple[int, int, int, int]) -> float:

ax1, ay1, ax2, ay2 = box_a

bx1, by1, bx2, by2 = box_b

ix1 = max(ax1, bx1)

iy1 = max(ay1, by1)

ix2 = min(ax2, bx2)

iy2 = min(ay2, by2)

iw = max(0, ix2 - ix1)

ih = max(0, iy2 - iy1)

 inter = iw * ih

union = bbox_area(box_a) + bbox_area(box_b) - inter

if union <= 0:

return 0.0

return float(inter) / float(union)

def bbox_center_distance(box_a: Tuple[int, int, int, int], box_b: Tuple[int, int, int, int]) ->
float:

ax1, ay1, ax2, ay2 = box_a

bx1, by1, bx2, by2 = box_b

acx = 0.5 * (ax1 + ax2)

acy = 0.5 * (ay1 + ay2)

bcx = 0.5 * (bx1 + bx2)

bcy = 0.5 * (by1 + by2)

return float(((acx - bcx) ** 2 + (acy - bcy) ** 2) ** 0.5)

def bbox_transition_zone(

bbox: Tuple[int, int, int, int],

frame_shape: Tuple[int, int, int],

margin_ratio: float = 0.16,

) -> str:

frame_h, frame_w = frame_shape[:2]

if frame_h <= 0 or frame_w <= 0:

return "center"
 x1, y1, x2, y2 = bbox

margin_x = max(8, int(frame_w * float(margin_ratio)))

margin_y = max(8, int(frame_h * float(margin_ratio)))

touches = []

if x1 <= margin_x:

[Link](("left", margin_x - x1))

if x2 >= frame_w - margin_x:

[Link](("right", x2 - (frame_w - margin_x)))

if y1 <= margin_y:

[Link](("top", margin_y - y1))

if y2 >= frame_h - margin_y:

[Link](("bottom", y2 - (frame_h - margin_y)))

if not touches:

return "center"

return max(touches, key=lambda item: item[1])[0]

def transition_zone_score(prev_zone: Optional[str], next_zone: Optional[str]) -> float:

if not prev_zone or not next_zone:

return 0.0

if prev_zone == "center" or next_zone == "center":

return 0.0

if prev_zone == next_zone:
 return 0.55

opposite_pairs = {

("left", "right"),

("right", "left"),

("top", "bottom"),

("bottom", "top"),

if (prev_zone, next_zone) in opposite_pairs:

return 1.0

horizontal = {"left", "right"}

vertical = {"top", "bottom"}

if prev_zone in horizontal and next_zone in horizontal:

return 0.7

if prev_zone in vertical and next_zone in vertical:

return 0.7

return 0.35

def non_max_suppress_detections(detections: List[Dict[str, Any]], iou_thresh: float = 0.65)

-> List[Dict[str, Any]]:

kept: List[Dict[str, Any]] = []

for det in sorted(detections, key=lambda item: float([Link]("conf", 0.0)), reverse=True):

bbox = tuple(map(int, det["bbox"]))

if any(bbox_iou(bbox, tuple(map(int, prev["bbox"]))) >= iou_thresh for prev in kept):

 continue

[Link](det)

return kept

def match_track_to_detection_bbox(

track_bbox: Tuple[int, int, int, int],

detections: List[Dict[str, Any]],

min_iou: float = 0.15,

max_center_ratio: float = 0.75,

) -> Optional[Tuple[int, int, int, int]]:

best_bbox = None

best_score = None

track_w = max(1, int(track_bbox[2] - track_bbox[0]))

track_h = max(1, int(track_bbox[3] - track_bbox[1]))

max_center_dist = max(60.0, max(track_w, track_h) * float(max_center_ratio))

for det in detections:

dbbox = tuple(map(int, det["bbox"]))

center_dist = bbox_center_distance(track_bbox, dbbox)

overlap = bbox_iou(track_bbox, dbbox)

if overlap < min_iou and center_dist > max_center_dist:

continue

# Prefer high IoU, then small center distance.

score = (-overlap, center_dist)

if best_score is None or score < best_score:

 best_score = score

best_bbox = dbbox

return best_bbox

def refine_person_bbox_from_fg(

fg_mask: [Link],

bbox: Tuple[int, int, int, int],

frame_shape: Tuple[int, int, int],

min_contour_ratio: float,

) -> Tuple[Tuple[int, int, int, int], float, bool]:

x1, y1, x2, y2 = bbox

if x2 <= x1 or y2 <= y1:

return bbox, 0.0, False

fg_crop = fg_mask[y1:y2, x1:x2]

if fg_crop.size == 0:

return bbox, 0.0, False

_, fg_bin = [Link](fg_crop, 127, 255, cv2.THRESH_BINARY)

kernel = [Link]((3, 3), dtype=np.uint8)

fg_bin = [Link](fg_bin, cv2.MORPH_OPEN, kernel, iterations=1)

fg_bin = [Link](fg_bin, cv2.MORPH_CLOSE, kernel, iterations=2)

contours, _ = [Link](fg_bin, cv2.RETR_EXTERNAL,

cv2.CHAIN_APPROX_SIMPLE)
 if not contours:

return bbox, 0.0, False

largest = max(contours, key=[Link])

contour_area = float([Link](largest))

box_area = float(max(1, (x2 - x1) * (y2 - y1)))

contour_ratio = contour_area / box_area

if contour_ratio < min_contour_ratio:

return bbox, contour_ratio, False

rx, ry, rw, rh = [Link](largest)

pad_x = max(2, int(rw * 0.08))

pad_y = max(2, int(rh * 0.06))

nx1 = max(0, x1 + rx - pad_x)

ny1 = max(0, y1 + ry - pad_y)

nx2 = min(frame_shape[1] - 1, x1 + rx + rw + pad_x)

ny2 = min(frame_shape[0] - 1, y1 + ry + rh + pad_y)

refined = (int(nx1), int(ny1), int(nx2), int(ny2))

return refined, contour_ratio, True

# gait and face helper implementations are provided by fm_gait and fm_face

# (extracted and imported at module top). This avoids duplicate definitions

# and keeps the fusion logic centralized.

class GlobalIdentityBank:

def __init__(self, sim_threshold: float = 0.72, name_map: Optional[Dict[str, str]] = None):

 self.sim_threshold = sim_threshold

self.name_map = name_map or {}

[Link]: Dict[str, Dict[str, Any]] = {}

self.next_id = 1

def _resolve_display_name(self, label: Any) -> str:

return resolve_display_name(label, self.name_map)

def _ensure_meta(self, gid: str) -> Dict[str, Any]:

if gid not in [Link]:

[Link][gid] = {

"emb": None,

"count": 0,

"identity_key": None,

"display_name": None,

"label_source": None,

"label_votes": {},

"cam_last_seen": {},

"cam_last_boxes": {},

"cam_last_zones": {},

"last_cam": None,

"last_ts": 0.0,

"last_zone": None,

"last_assoc_reason": "new",

"last_assoc_score": 0.0,

}
 return [Link][gid]

def _new_id(self) -> str:

gid = f"G{self.next_id:04d}"

self.next_id += 1

return gid

@staticmethod

def _gid_sort_key(gid: str) -> Tuple[int, Any]:

if isinstance(gid, str) and [Link]("ID_"):

return (0, gid)

if isinstance(gid, str) and [Link]("G") and gid[1:].isdigit():

return (1, int(gid[1:]))

return (2, str(gid))

def correlation_cluster_merge(

self,

now: float,

active_window_sec: float = 3.0,

sim_threshold: float = 0.93,

same_cam_window_sec: float = 0.6,

) -> Dict[str, str]:

"""Merge highly similar GIDs (approx correlation clustering).

Conservative rules:

- Only consider identities seen recently (within active_window_sec).

- Disallow merges when the two identities appear at the same time in the

same camera but at clearly different locations.

- Disallow merges when identity_key conflicts.

Returns: mapping {old_gid: new_gid} for merged IDs.

"""

# Collect eligible nodes

nodes: List[str] = []

for gid, meta in [Link]():

if [Link]("emb") is None:

continue

last_ts = float([Link]("last_ts", 0.0) or 0.0)

if float(now) - last_ts > float(active_window_sec):

continue

[Link](gid)

if len(nodes) < 2:

return {}

# Union-Find

parent = {gid: gid for gid in nodes}

def find(x: str) -> str:

while parent[x] != x:

parent[x] = parent[parent[x]]
 x = parent[x]

return x

def union(a: str, b: str):

ra, rb = find(a), find(b)

if ra != rb:

parent[rb] = ra

def conflicts(meta_a: Dict[str, Any], meta_b: Dict[str, Any]) -> bool:

ka = meta_a.get("identity_key")

kb = meta_b.get("identity_key")

if ka and kb and str(ka) != str(kb):

return True

return False

def same_cam_negative(meta_a: Dict[str, Any], meta_b: Dict[str, Any]) -> bool:

cams = set(meta_a.get("cam_last_seen", {}).keys()) &

set(meta_b.get("cam_last_seen", {}).keys())

if not cams:

return False

for cam in cams:

try:

ta = float(meta_a.get("cam_last_seen", {}).get(cam, 0.0) or 0.0)

tb = float(meta_b.get("cam_last_seen", {}).get(cam, 0.0) or 0.0)

if abs(ta - tb) > float(same_cam_window_sec):

continue
 ba = meta_a.get("cam_last_boxes", {}).get(cam)

bb = meta_b.get("cam_last_boxes", {}).get(cam)

if ba is None or bb is None:

continue

ba = tuple(map(int, ba))

bb = tuple(map(int, bb))

iou = bbox_iou(ba, bb)

dist = bbox_center_distance(ba, bb)

# If they are far apart with near-zero overlap at the same time in same cam ->
likely different people.

diag_a = float(((ba[2] - ba[0]) ** 2 + (ba[3] - ba[1]) ** 2) ** 0.5)

diag_b = float(((bb[2] - bb[0]) ** 2 + (bb[3] - bb[1]) ** 2) ** 0.5)

diag = max(1.0, min(diag_a, diag_b))

if iou < 0.02 and dist > max(60.0, 0.6 * diag):

return True

except Exception:

continue

return False

# Build positive edges by similarity threshold

for i in range(len(nodes)):

gid_i = nodes[i]

meta_i = [Link](gid_i, {})

emb_i = meta_i.get("emb")

if emb_i is None:

continue
 for j in range(i + 1, len(nodes)):

gid_j = nodes[j]

meta_j = [Link](gid_j, {})

emb_j = meta_j.get("emb")

if emb_j is None:

continue

if conflicts(meta_i, meta_j):

continue

if same_cam_negative(meta_i, meta_j):

continue

sim = cosine_sim(emb_i, emb_j)

if float(sim) >= float(sim_threshold):

union(gid_i, gid_j)

# Group by root

groups: Dict[str, List[str]] = {}

for gid in nodes:

[Link](find(gid), []).append(gid)

merged: Dict[str, str] = {}

for _root, members in [Link]():

if len(members) < 2:

continue
 # Choose a representative (prefer ID_*, then smallest)

members_sorted = sorted(members, key=self._gid_sort_key)

rep = members_sorted[0]

# Merge others into rep

for other in members_sorted[1:]:

if other == rep:

continue

if other not in [Link] or rep not in [Link]:

continue

self._promote_gid(other, rep)

merged[other] = rep

return merged

def _find_best_gid(self, emb: Optional[[Link]]) -> Tuple[Optional[str], float]:

if emb is None:

return None, -1.0

best_gid = None

best_sim = -1.0

for gid, meta in [Link]():

ref = [Link]("emb")

if ref is None:

continue

s = cosine_sim(emb, ref)

if s > best_sim:
 best_sim = s

best_gid = gid

return best_gid, best_sim

def _merge_votes(self, dst: Dict[str, int], src: Dict[str, int]):

for key, value in [Link]():

dst[key] = int([Link](key, 0)) + int(value)

def _has_conflicting_identity(self, meta: Dict[str, Any], preferred_label: Optional[str]) ->

bool:

if not preferred_label or preferred_label == "unknown":

return False

identity_key = [Link]("identity_key")

if not identity_key:

return False

return str(identity_key) != normalize_identity_key(preferred_label)

def _promote_gid(self, old_gid: str, new_gid: str):

if old_gid == new_gid or old_gid not in [Link]:

return

old_meta = [Link](old_gid)

new_meta = self._ensure_meta(new_gid)

if old_meta.get("emb") is not None:

if new_meta.get("emb") is None:
 new_meta["emb"] = old_meta["emb"]

new_meta["count"] = int(old_meta.get("count", 1))

else:

old_count = max(1, int(old_meta.get("count", 1)))

new_count = max(1, int(new_meta.get("count", 1)))

merged = ((old_meta["emb"] * old_count) + (new_meta["emb"] * new_count)) /

float(old_count + new_count)

new_meta["emb"] = l2_normalize([Link](1, -1), axis=1)[0]

new_meta["count"] = old_count + new_count

self._merge_votes(new_meta.setdefault("label_votes", {}), old_meta.get("label_votes",

{}))

if old_meta.get("identity_key") and not new_meta.get("identity_key"):

new_meta["identity_key"] = old_meta.get("identity_key")

new_meta["display_name"] = old_meta.get("display_name")

new_meta["label_source"] = old_meta.get("label_source")

for cam_idx, ts in old_meta.get("cam_last_seen", {}).items():

prev_ts = float(new_meta.setdefault("cam_last_seen", {}).get(cam_idx, 0.0))

if float(ts) > prev_ts:

new_meta["cam_last_seen"][cam_idx] = float(ts)

old_box = old_meta.get("cam_last_boxes", {}).get(cam_idx)

old_zone = old_meta.get("cam_last_zones", {}).get(cam_idx)

if old_box is not None:

new_meta.setdefault("cam_last_boxes", {})[cam_idx] = tuple(map(int, old_box))

if old_zone is not None:

 new_meta.setdefault("cam_last_zones", {})[cam_idx] = str(old_zone)

if float(old_meta.get("last_ts", 0.0)) > float(new_meta.get("last_ts", 0.0)):

new_meta["last_cam"] = old_meta.get("last_cam")

new_meta["last_ts"] = float(old_meta.get("last_ts", 0.0))

new_meta["last_zone"] = old_meta.get("last_zone")

new_meta["last_assoc_reason"] = old_meta.get("last_assoc_reason",
new_meta.get("last_assoc_reason", "promoted"))

new_meta["last_assoc_score"] = float(old_meta.get("last_assoc_score",
new_meta.get("last_assoc_score", 0.0)))

def _find_cross_camera_match(

self,

emb: Optional[[Link]],

cam_idx: int,

bbox: Tuple[int, int, int, int],

frame_shape: Tuple[int, int, int],

now: float,

reid_threshold: float,

overlap_threshold: float,

active_window_sec: float,

min_transition_sec: float,

max_transition_sec: float,

transition_margin_ratio: float,

) -> Tuple[Optional[str], float, Optional[str]]:

if emb is None:

return None, -1.0, None

 entry_zone = bbox_transition_zone(bbox, frame_shape,
margin_ratio=transition_margin_ratio)

best_gid = None

best_score = -1.0

best_reason = None

for gid, meta in [Link]():

ref = [Link]("emb")

if ref is None:

continue

other_cams = [

(int(other_cam), float(ts))

for other_cam, ts in [Link]("cam_last_seen", {}).items()

if int(other_cam) != int(cam_idx)

if not other_cams:

continue

sim = cosine_sim(emb, ref)

last_cam, last_seen = max(other_cams, key=lambda item: item[1])

time_gap = max(0.0, float(now) - float(last_seen))

last_zone = [Link]("cam_last_zones", {}).get(last_cam)

reason = None

gate_bonus = 0.0
 if time_gap <= float(active_window_sec):

if sim < float(overlap_threshold):

continue

reason = "sync-overlap"

gate_bonus = 0.08

else:

if time_gap < float(min_transition_sec) or time_gap > float(max_transition_sec):

continue

zone_score = transition_zone_score(last_zone, entry_zone)

if zone_score <= 0.0 or sim < float(reid_threshold):

continue

reason = "sync-transition"

gate_bonus = 0.03 + 0.05 * zone_score

if [Link]("identity_key"):

gate_bonus += 0.02

score = sim + gate_bonus

if score > best_score:

best_gid = gid

best_score = score

best_reason = reason

return best_gid, best_score, best_reason

def observe(

self,

gid: str,

cam_idx: int,

bbox: Tuple[int, int, int, int],

frame_shape: Tuple[int, int, int],

now: Optional[float] = None,

assoc_reason: Optional[str] = None,

assoc_score: Optional[float] = None,

transition_margin_ratio: float = 0.16,

):

meta = self._ensure_meta(gid)

obs_ts = float(now if now is not None else [Link]())

zone = bbox_transition_zone(bbox, frame_shape, margin_ratio=transition_margin_ratio)

[Link]("cam_last_seen", {})[int(cam_idx)] = obs_ts

[Link]("cam_last_boxes", {})[int(cam_idx)] = tuple(map(int, bbox))

[Link]("cam_last_zones", {})[int(cam_idx)] = zone

meta["last_cam"] = int(cam_idx)

meta["last_ts"] = obs_ts

meta["last_zone"] = zone

if assoc_reason:

meta["last_assoc_reason"] = str(assoc_reason)

if assoc_score is not None:

meta["last_assoc_score"] = float(assoc_score)

def assign(
 self,

emb: Optional[[Link]],

cam_idx: int,

bbox: Tuple[int, int, int, int],

frame_shape: Tuple[int, int, int],

preferred_label: Optional[str] = None,

preferred_source: Optional[str] = None,

sync_id: bool = True,

reid_threshold: float = 0.84,

overlap_threshold: float = 0.90,

active_window_sec: float = 1.2,

min_transition_sec: float = 0.10,

max_transition_sec: float = 4.0,

transition_margin_ratio: float = 0.16,

now: Optional[float] = None,

) -> Tuple[str, str, float]:

obs_ts = float(now if now is not None else [Link]())

best_gid, best_sim = self._find_best_gid(emb)

# If we got a confident known label from face/gait, bind global ID to that label.

if preferred_label and preferred_label != "unknown":

src = str(preferred_source or "face").strip().lower()

if src not in ("face", "gait"):

src = "face"

gid = f"ID_{preferred_label}"

if (
 best_gid is not None

and best_sim >= self.sim_threshold

and best_gid != gid

and not self._has_conflicting_identity([Link](best_gid, {}), preferred_label)

):

self._promote_gid(best_gid, gid)

meta = self._ensure_meta(gid)

meta["identity_key"] = normalize_identity_key(preferred_label)

meta["display_name"] = self._resolve_display_name(preferred_label)

meta["label_source"] = src

if emb is not None:

self._update(gid, emb)

[Link](

gid,

cam_idx,

bbox,

frame_shape,

now=obs_ts,

assoc_reason=src,

assoc_score=1.0,

transition_margin_ratio=transition_margin_ratio,

return gid, src, 1.0

if emb is None:

gid = self._new_id()
 [Link](

gid,

cam_idx,

bbox,

frame_shape,

now=obs_ts,

assoc_reason="new",

assoc_score=0.0,

transition_margin_ratio=transition_margin_ratio,

return gid, "new", 0.0

if sync_id:

matched_gid, matched_score, matched_reason = self._find_cross_camera_match(

emb=emb,

cam_idx=cam_idx,

bbox=bbox,

frame_shape=frame_shape,

now=obs_ts,

reid_threshold=reid_threshold,

overlap_threshold=overlap_threshold,

active_window_sec=active_window_sec,

min_transition_sec=min_transition_sec,

max_transition_sec=max_transition_sec,

transition_margin_ratio=transition_margin_ratio,

)
 if matched_gid is not None and matched_reason is not None:

self._update(matched_gid, emb)

[Link](

matched_gid,

cam_idx,

bbox,

frame_shape,

now=obs_ts,

assoc_reason=matched_reason,

assoc_score=matched_score,

transition_margin_ratio=transition_margin_ratio,

return matched_gid, matched_reason, matched_score

# If cross-camera sync didn't match, still attempt same-camera appearance

matching.

# This prevents duplicate GIDs/tracks for the same person appearing simultaneously

# in a single camera when tracking briefly creates two track IDs.

if best_gid is not None and best_sim >= self.sim_threshold:

# avoid promoting/merging if identity would conflict with a preferred label

if not self._has_conflicting_identity([Link](best_gid, {}), preferred_label):

self._update(best_gid, emb)

[Link](

best_gid,

cam_idx,

bbox,
 frame_shape,

now=obs_ts,

assoc_reason="appearance",

assoc_score=best_sim,

transition_margin_ratio=transition_margin_ratio,

return best_gid, "appearance", best_sim

if not sync_id and best_gid is not None and best_sim >= self.sim_threshold:

self._update(best_gid, emb)

[Link](

best_gid,

cam_idx,

bbox,

frame_shape,

now=obs_ts,

assoc_reason="appearance",

assoc_score=best_sim,

transition_margin_ratio=transition_margin_ratio,

return best_gid, "appearance", best_sim

gid = self._new_id()

[Link][gid] = {

"emb": [Link](emb, dtype=np.float32),

"count": 1,
 "identity_key": None,

"display_name": None,

"label_source": None,

"label_votes": {},

"cam_last_seen": {},

"cam_last_boxes": {},

"cam_last_zones": {},

"last_cam": None,

"last_ts": 0.0,

"last_zone": None,

"last_assoc_reason": "new",

"last_assoc_score": 0.0,

[Link](

gid,

cam_idx,

bbox,

frame_shape,

now=obs_ts,

assoc_reason="new",

assoc_score=0.0,

transition_margin_ratio=transition_margin_ratio,

return gid, "new", 0.0

def _update(self, gid: str, emb: [Link]):

 emb = [Link](emb, dtype=np.float32)

meta = self._ensure_meta(gid)

if [Link]("emb") is None:

meta["emb"] = emb

meta["count"] = 1

return

c = max(1, int([Link]("count", 1)))

meta["emb"] = l2_normalize(((meta["emb"] * c + emb) / (c + 1)).reshape(1, -1), axis=1)[0]

meta["count"] = c + 1

def update_identity(

self,

gid: str,

cam_idx: int,

face_label: str,

face_score: float,

gait_label: Optional[str],

gait_score: float,

fused_id: str,

fused_name: str,

fused_score: float,

face_thresh: float,

gait_thresh: float,

now: Optional[float] = None,

):

meta = self._ensure_meta(gid)
 update_ts = float(now if now is not None else [Link]())

[Link]("cam_last_seen", {})[int(cam_idx)] = update_ts

if face_label != "unknown" and face_score >= face_thresh:

meta["identity_key"] = normalize_identity_key(face_label)

meta["display_name"] = fused_name if fused_name != "unknown" else

self._resolve_display_name(face_label)

meta["label_source"] = "face"

meta["label_votes"] = {meta["identity_key"]: max(3, int([Link]("label_votes",

{}).get(meta["identity_key"], 0)))}

return

if [Link]("label_source") == "face" and [Link]("identity_key"):

return

candidate_key = None

candidate_name = None

if fused_id != "unknown" and fused_score >= max(gait_thresh + 0.05, 0.55):

candidate_key = normalize_identity_key(fused_id)

candidate_name = fused_name if fused_name != "unknown" else

self._resolve_display_name(fused_id)

elif gait_label and gait_label != "unknown" and gait_score >= gait_thresh + 0.08:

candidate_key = normalize_identity_key(gait_label)

candidate_name = self._resolve_display_name(gait_label)

if not candidate_key:

return
 votes = [Link]("label_votes", {})

votes[candidate_key] = int([Link](candidate_key, 0)) + 1

recent_cams = [key for key, ts in [Link]("cam_last_seen", {}).items() if update_ts -

float(ts) <= 1.5]

required_votes = 4 if len(recent_cams) >= 2 else 3

if votes[candidate_key] >= required_votes:

meta["identity_key"] = candidate_key

meta["display_name"] = candidate_name

meta["label_source"] = "gait"

def get_identity(self, gid: str) -> Tuple[Optional[str], Optional[str], Optional[str]]:

meta = [Link](gid)

if not meta:

return None, None, None

return [Link]("identity_key"), [Link]("display_name"), [Link]("label_source")

# face helper functions are provided by fm_face and imported at module top

def run_realtime_mtmct(

alpha: float,

face_thresh: float,

gait_thresh: float,

fuse_thresh: float,
cameras: List[Any],

gait_cfg: str,

yolo_model_path: str,

gait_dist_port: int = 0,

gait_db_path: Optional[str] = None,

dynamic_alpha: bool = False,

fusion_mlp_path: Optional[str] = None,

cc_merge: bool = False,

cc_merge_sim: float = 0.93,

cc_merge_window_sec: float = 3.0,

cc_merge_period_sec: float = 1.0,

det_conf: float = 0.35,

min_height: int = 64,

min_area: int = 2000,

min_fg_ratio: float = 0.02,

min_fg_motion: float = 0.002,

min_aspect: float = 1.0,

use_opengait_tracking: bool = False,

display_sync_ms: int = 80,

display_wait_ms: int = 120,

min_contour_ratio: float = 0.10,

max_box_area_ratio: float = 0.80,

max_box_width_ratio: float = 0.90,

face_min_size: int = 20,

face_body_overlap: float = 0.35,

person_face_region_ratio: float = 0.78,

 bbox_history: int = 5,

gait_ready_frames: int = 24,

gait_history_maxlen: int = 12,

gait_vote_min: int = 3,

gait_margin_min: float = 0.001,

gait_vote_margin: int = 1,

sync_id: bool = True,

cross_cam_reid_thresh: float = 0.84,

cross_cam_overlap_thresh: float = 0.90,

cross_cam_active_window_sec: float = 1.20,

cross_cam_min_transition_sec: float = 0.10,

cross_cam_max_transition_sec: float = 4.00,

cross_cam_transition_margin: float = 0.16,

prefer_face: bool = False,

debug_gait: bool = False,

):

if YOLO is None:

raise RuntimeError("Chua cai ultralytics. Cai bang: pip install ultralytics")

face_db, face_labels = load_face_db()

gait_db, gait_labels = load_gait_db(gait_db_path)

if gait_db is None or len(gait_labels) == 0:

raise RuntimeError(

f"Gait DB khong ton tai/rong. Kiem tra PKL: {gait_db_path or

[Link](core.OPEN_GAIT_ROOT, 'output', 'CASIA-B', 'GaitGL', 'gait_system_db.pkl')}"

)
 name_map = load_name_map()

gait_runtime = try_init_gait_runtime(gait_cfg, dist_port=int(gait_dist_port))

if gait_runtime is None:

raise RuntimeError("Khong khoi tao duoc OpenGait runtime")

detector, backbone, quality = build_face_models()

person_det = YOLO(yolo_model_path)

if ReIDEncoder is None:

raise RuntimeError("Khong import duoc ReIDEncoder. Kiem tra

OpenGait/mtmct/reid_encoder.py")

reid = ReIDEncoder()

readers: List[AsyncCameraReader] = []

trackers: List[TrackerWrapper] = []

gait_buffers: List[Dict[int, deque]] = []

gait_embed_histories: List[Dict[int, deque]] = []

gait_label_histories: List[Dict[int, deque]] = []

gait_top1_histories: List[Dict[int, deque]] = []

gait_reject_streaks: List[Dict[int, int]] = []

gait_append_counts: List[Dict[int, int]] = []

gait_last_extract_counts: List[Dict[int, int]] = []

face_label_histories: List[Dict[int, deque]] = []

cam_track_gid: List[Dict[int, str]] = []

 cam_track_assoc: List[Dict[int, Dict[str, Any]]] = []

bsubs: List[Any] = []

track_bbox_histories: List[Dict[int, deque]] = []

# OpenGait-style simple tracker (centroid assignment + per-track buffers)

og_tracks_list: List[Dict[int, Dict[str, Any]]] = []

og_next_tid: List[int] = []

last_processed_seq: List[int] = []

last_views: List[Optional[[Link]]] = []

prev_fg_masks: List[Optional[[Link]]] = []

for ci, cam in enumerate(cameras):

[Link](AsyncCameraReader(cam, ci))

[Link](TrackerWrapper(max_age=30, n_init=3,
logger_name=f"MOT_CAM{ci}"))

gait_buffers.append({})

gait_embed_histories.append({})

gait_label_histories.append({})

gait_top1_histories.append({})

gait_reject_streaks.append({})

gait_append_counts.append({})

gait_last_extract_counts.append({})

face_label_histories.append({})

cam_track_gid.append({})

cam_track_assoc.append({})

# background subtractor for silhouette extraction (frame-level)

bsub = cv2.createBackgroundSubtractorMOG2(history=500, varThreshold=16,

detectShadows=False)
 [Link](bsub)

track_bbox_histories.append({})

og_tracks_list.append({})

og_next_tid.append(0)

last_processed_seq.append(0)

last_views.append(None)

# per-camera previous foreground mask for simple temporal motion check

prev_fg_masks.append(None)

bank = GlobalIdentityBank(sim_threshold=0.72, name_map=name_map)

fusion_mlp = try_load_fusion_mlp(fusion_mlp_path)

last_cc_merge_ts = 0.0

configure_fullscreen_window("Fusion MTMCT")

print(f"[INFO] MTMCT started. cams={len(cameras)} face_db={len(face_labels)}

gait_db={len(gait_labels)} sync_id={'on' if sync_id else 'off'}")

try:

while True:

views = []

snapshots, sync_delta_ms = collect_synchronized_frames(readers,

display_sync_ms, display_wait_ms)

valid_shapes = [item[0].shape for item in snapshots if item[0] is not None]

base_shape = valid_shapes[0] if valid_shapes else (480, 640, 3)

for ci, (frame, frame_ts, frame_seq) in enumerate(snapshots):

if frame is None:

fallback = last_views[ci]
 vis = [Link]() if fallback is not None else build_placeholder(base_shape,
f"CAM {ci} NO SIGNAL")

[Link](vis)

continue

if frame_seq == last_processed_seq[ci] and last_views[ci] is not None:

vis = last_views[ci].copy()

age_ms = max(0.0, ([Link]() - frame_ts) * 1000.0)

[Link](vis, f"CAM {ci} WAITING {age_ms:.0f}ms", (10, 30),

cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 200, 255), 2)

[Link](vis)

continue

vis = [Link]()

last_processed_seq[ci] = frame_seq

# compute frame-level foreground mask used for silhouette extraction

gray_full = [Link](frame, cv2.COLOR_BGR2GRAY)

fg_mask = bsubs[ci].apply(gray_full)

frame_faces = detect_faces_in_frame(

frame,

detector,

backbone,

quality,

min_face_size=max(int(face_min_size), int(min([Link][0], [Link][1]) *

0.025)),

)
 used_face_indices = set()

frame_area = [Link][0] * [Link][1]

# 1) YOLO person detection for gait bbox

result = person_det(frame, verbose=False)[0]

boxes = getattr(result, "boxes", None)

detections = []

if boxes is not None:

for b in boxes:

cls_id = int([Link][0].item()) if hasattr(b, "cls") else int([Link]())

name = [Link][cls_id] if isinstance([Link], dict) else

[Link][cls_id]

if str(name).lower() != "person":

continue

conf = float([Link][0].item()) if hasattr(b, "conf") else float([Link]())

if conf < det_conf:

continue

xyxy = [Link][0].cpu().numpy().astype(int)

x1, y1, x2, y2 = xyxy

x1, y1 = max(0, x1), max(0, y1)

x2, y2 = min([Link][1] - 1, x2), min([Link][0] - 1, y2)

if x2 <= x1 or y2 <= y1:

continue

w = x2 - x1

h = y2 - y1

area = w * h
# basic size/aspect filters to avoid tiny false positives

if h < min_height or area < min_area:

continue

if area > int(frame_area * max_box_area_ratio):

continue

if w > int([Link][1] * max_box_width_ratio):

continue

aspect = float(h) / float(max(1, w))

if aspect < min_aspect:

continue

# check foreground ratio inside bbox using MOG2 fg mask

fg_crop = None

try:

fg_crop = fg_mask[y1:y2, x1:x2]

except Exception:

fg_crop = None

if fg_crop is None or fg_crop.size == 0:

continue

fg_sum = float(np.count_nonzero(fg_crop))

# simple temporal motion test: compare to previous fg mask crop

motion = 0.0

try:

prev_mask = prev_fg_masks[ci]

if prev_mask is not None:

prev_crop = prev_mask[y1:y2, x1:x2]

if prev_crop.shape == fg_crop.shape:
 motion = float(np.count_nonzero(fg_crop != prev_crop)) / float(max(1,
area))

except Exception:

motion = 0.0

# accept detection only if it has sufficient fg ratio OR has recent motion

if fg_sum / float(max(1, area)) < float(min_fg_ratio) and motion <

float(min_fg_motion):

# mostly background and no recent motion -> likely static background

continue

crop = frame[y1:y2, x1:x2]

if [Link] == 0:

continue

[Link](

"bbox": (x1, y1, x2, y2),

"conf": float(conf),

"feature": None,

"fg_ratio": fg_sum / float(max(1, area)),

"contour_ratio": 0.0,

detections = non_max_suppress_detections(detections, iou_thresh=0.65)

person_crops = [frame[y1:y2, x1:x2] for x1, y1, x2, y2 in [tuple(map(int, d["bbox"]))

for d in detections]]
person_crops = [crop for crop in person_crops if [Link] > 0]

if len(person_crops) != len(detections):

filtered_detections = []

filtered_crops = []

for det in detections:

x1, y1, x2, y2 = map(int, det["bbox"])

crop = frame[y1:y2, x1:x2]

if [Link] == 0:

continue

filtered_detections.append(det)

filtered_crops.append(crop)

detections = filtered_detections

person_crops = filtered_crops

# 2) ReID feature for tracker association

if person_crops:

feats = reid.encode_batch(person_crops)

for i in range(min(len(feats), len(detections))):

detections[i]["feature"] = feats[i]

# 3) Tracking per camera

tracks = []

if use_opengait_tracking:

# centroid-based simple tracker similar to OpenGait's webcam_recognize

# og_tracks: tid -> {'bbox':(x1,y1,x2,y2),'buffer':[], 'lost':int, 'history':[]}

og_tracks = og_tracks_list[ci]
next_tid = og_next_tid[ci]

current_seen = {}

match_thresh = 80.0

for det in detections:

x1, y1, x2, y2 = det['bbox']

cx = 0.5 * (x1 + x2)

cy = 0.5 * (y1 + y2)

best_tid = None

best_dist = None

for tid, tinfo in og_tracks.items():

tx1, ty1, tx2, ty2 = tinfo['bbox']

tcx = 0.5 * (tx1 + tx2)

tcy = 0.5 * (ty1 + ty2)

d = ((cx - tcx) ** 2 + (cy - tcy) ** 2) ** 0.5

if best_dist is None or d < best_dist:

best_dist = d

best_tid = tid

if best_tid is not None and best_dist is not None and best_dist < match_thresh:

tid = best_tid

else:

tid = next_tid

og_tracks[tid] = {'bbox': (x1, y1, x2, y2), 'buffer': [], 'lost': 0, 'history': [], 'hits': 0}

next_tid += 1

og_tracks[tid]['bbox'] = (x1, y1, x2, y2)

og_tracks[tid]['hits'] = int(og_tracks[tid].get('hits', 0)) + 1

current_seen[tid] = (x1, y1, x2, y2)

 # update per-track fg buffer using fg_mask

try:

fg_crop = fg_mask[y1:y2, x1:x2]

except Exception:

fg_crop = None

if fg_crop is not None and fg_crop.size > 0:

_, fg_bin = [Link](fg_crop, 127, 255, cv2.THRESH_BINARY)

norm = gait_runtime['preprocess'](fg_bin)

if norm is not None:

buf = og_tracks[tid]['buffer']

[Link](norm)

if len(buf) > 45:

[Link](0)

# update lost counters and delete stale tracks

to_delete = []

for tid, tinfo in list(og_tracks.items()):

if tid not in current_seen:

tinfo['lost'] += 1

else:

tinfo['lost'] = 0

if tinfo['lost'] > 30:

to_delete.append(tid)

for tid in to_delete:

del og_tracks[tid]

# produce 'tracks' list compatible with downstream code

for tid, tinfo in og_tracks.items():

 if int([Link]('hits', 0)) < 2:

continue

tbbox = list(map(int, tinfo['bbox']))

[Link]({'track_id': tid, 'bbox': tbbox})

og_tracks_list[ci] = og_tracks

og_next_tid[ci] = next_tid

else:

tracks = trackers[ci].update(detections, frame=frame) if detections else []

# 4) Per-track face+gait recognition and score fusion

active_tids = {int(track["track_id"]) for track in tracks}

for store in (

gait_buffers[ci],

gait_label_histories[ci],

gait_reject_streaks[ci],

track_bbox_histories[ci],

cam_track_gid[ci],

cam_track_assoc[ci],

):

stale_ids = [key for key in list([Link]()) if int(key) not in active_tids]

for stale_id in stale_ids:

[Link](stale_id, None)

for t in tracks:

tid = int(t["track_id"])

track_bbox = tuple(map(int, t["bbox"]))

 matched_det_bbox = match_track_to_detection_bbox(track_bbox, detections,
min_iou=0.15, max_center_ratio=0.90)

# recent history for this track (used for interpolation and adaptive history)

prev_hist = track_bbox_histories[ci].get(tid)

# If detection missing, try simple motion-vector interpolation using last two boxes

if matched_det_bbox is not None:

base_bbox = matched_det_bbox

else:

if prev_hist and len(prev_hist) >= 2:

last = prev_hist[-1]

prev = prev_hist[-2]

lc_x = 0.5 * (last[0] + last[2])

lc_y = 0.5 * (last[1] + last[3])

pc_x = 0.5 * (prev[0] + prev[2])

pc_y = 0.5 * (prev[1] + prev[3])

dx = lc_x - pc_x

dy = lc_y - pc_y

w_last = max(1, last[2] - last[0])

h_last = max(1, last[3] - last[1])

dx = float([Link](dx, -2.0 * w_last, 2.0 * w_last))

dy = float([Link](dy, -2.0 * h_last, 2.0 * h_last))

pred_x1 = int(round(last[0] + dx))

pred_y1 = int(round(last[1] + dy))

pred_x2 = int(round(last[2] + dx))

 pred_y2 = int(round(last[3] + dy))

pred_x1 = max(0, pred_x1)

pred_y1 = max(0, pred_y1)

pred_x2 = min([Link][1] - 1, pred_x2)

pred_y2 = min([Link][0] - 1, pred_y2)

if pred_x2 > pred_x1 and pred_y2 > pred_y1:

base_bbox = (pred_x1, pred_y1, pred_x2, pred_y2)

else:

base_bbox = track_bbox

else:

base_bbox = track_bbox

bx1, by1, bx2, by2 = base_bbox

# Adaptive history length: distant/small persons -> longer smoothing; nearby ->
shorter

box_h = max(1, by2 - by1)

height_ratio = box_h / float(max(1, [Link][0]))

if height_ratio < 0.12:

history_maxlen = max(6, int(bbox_history * 2))

elif height_ratio < 0.22:

history_maxlen = max(5, int(bbox_history * 1.5))

else:

history_maxlen = max(3, int(bbox_history))

history_maxlen = int([Link](history_maxlen, 3, 48))

hist = track_bbox_histories[ci].get(tid)

if hist is None:

hist = deque(maxlen=history_maxlen)

track_bbox_histories[ci][tid] = hist

elif [Link] != history_maxlen:

items = list(hist)[-history_maxlen:]

hist = deque(items, maxlen=history_maxlen)

track_bbox_histories[ci][tid] = hist

# append the (matched or predicted) bbox into history

[Link]((bx1, by1, bx2, by2))

# compute median bbox

arr = [Link](hist, dtype=np.int32)

mx1 = int([Link](arr[:, 0]))

my1 = int([Link](arr[:, 1]))

mx2 = int([Link](arr[:, 2]))

my2 = int([Link](arr[:, 3]))

# use smoothed bbox for visualization and gait/face crops

sx1, sy1, sx2, sy2 = mx1, my1, mx2, my2

# adjust smoothed bbox to a stable gait aspect ratio to avoid distortion

try:

target_h, target_w = (64, 44)

target_aspect = float(target_h) / float(target_w)

cur_h = max(1, sy2 - sy1)

cur_w = max(1, sx2 - sx1)

 cur_aspect = float(cur_h) / float(cur_w)

if cur_aspect < target_aspect:

desired_h = int(round(cur_w * target_aspect))

add_h = max(0, desired_h - cur_h)

top = add_h // 2

bottom = add_h - top

sy1 = max(0, sy1 - top)

sy2 = min([Link][0] - 1, sy2 + bottom)

elif cur_aspect > target_aspect:

desired_w = int(round(cur_h / target_aspect))

add_w = max(0, desired_w - cur_w)

left = add_w // 2

right = add_w - left

sx1 = max(0, sx1 - left)

sx2 = min([Link][1] - 1, sx2 + right)

except Exception:

pass

x1, y1 = max(0, sx1), max(0, sy1)

x2, y2 = min([Link][1] - 1, sx2), min([Link][0] - 1, sy2)

if x2 <= x1 or y2 <= y1:

continue

smoothed_area = bbox_area((x1, y1, x2, y2))

if smoothed_area > int(frame_area * max_box_area_ratio) or (x2 - x1) >

int([Link][1] * max_box_width_ratio):

if matched_det_bbox is not None:

 x1, y1, x2, y2 = matched_det_bbox

else:

continue

person = frame[y1:y2, x1:x2]

if [Link] == 0:

continue

# face branch: use full-frame face detections, then match to current body box

face_emb = None

face_label = "unknown"

face_score = 0.0

matched_face_bbox = None

best_face_idx = None

best_face_ratio = 0.0

for face_idx, face_info in enumerate(frame_faces):

overlap_ratio = face_overlap_ratio(face_info["bbox"], (x1, y1, x2, y2))

if overlap_ratio > best_face_ratio:

best_face_ratio = overlap_ratio

best_face_idx = face_idx

if best_face_idx is not None and best_face_ratio >= float(face_body_overlap):

used_face_indices.add(best_face_idx)

matched_face = frame_faces[best_face_idx]

matched_face_bbox = matched_face["bbox"]

face_emb = matched_face["emb"]

fx1, fy1, fx2, fy2 = matched_face_bbox

 [Link](vis, (fx1, fy1), (fx2, fy2), (255, 255, 0), 1)

f_idx, f_score = cosine_top1(face_emb, face_db)

face_score = f_score

if f_idx >= 0 and f_score >= face_thresh:

face_label = face_labels[f_idx]

else:

upper_h = max(1, int([Link][0] * float(person_face_region_ratio)))

person_upper = person[:upper_h, :]

local_face_bbox, fallback_emb, _fallback_q = pick_face_in_person(

person_upper,

detector,

backbone,

quality,

min_face_size=max(16, int(face_min_size) - 4),

if fallback_emb is not None and local_face_bbox is not None:

lx1, ly1, lx2, ly2 = local_face_bbox

matched_face_bbox = (x1 + lx1, y1 + ly1, x1 + lx2, y1 + ly2)

face_emb = fallback_emb

fx1, fy1, fx2, fy2 = matched_face_bbox

[Link](vis, (fx1, fy1), (fx2, fy2), (255, 255, 0), 1)

f_idx, f_score = cosine_top1(face_emb, face_db)

face_score = f_score

if f_idx >= 0 and f_score >= face_thresh:

face_label = face_labels[f_idx]
# --- track recent face detections for stability ---

try:

fh = face_label_histories[ci]

except Exception:

fh = None

if fh is not None:

if tid not in fh:

fh[tid] = deque(maxlen=6)

fh[tid].append((face_label, float(face_score), [Link]()))

# compute stable face label over recent frames

stable_face_label = None

stable_face_score = 0.0

if fh is not None and tid in fh:

stable_face_label, stable_face_score = choose_stable_face_identity(

list(fh[tid]), face_thresh=float(face_thresh), min_consistent_frames=3

# gait branch from frame-level silhouette (use bgsub to stabilize)

gait_score = 0.0

gait_margin = 0.0

gait_emb = None

gait_label_top1 = "unknown"

gait_selected_mode = None

if tid not in gait_buffers[ci]:

gait_buffers[ci][tid] = deque(maxlen=45)

if tid not in gait_label_histories[ci]:

 gait_label_histories[ci][tid] = deque(maxlen=max(4, int(gait_history_maxlen)))

if tid not in gait_top1_histories[ci]:

gait_top1_histories[ci][tid] = deque(maxlen=8)

if tid not in gait_reject_streaks[ci]:

gait_reject_streaks[ci][tid] = 0

# Always initialize history_items so we can call

# choose_stable_gait_identity even before the gait buffer is ready.

history_items = list(gait_label_histories[ci][tid])

gait_bbox = (x1, y1, x2, y2)

sil_norm, gait_bbox, contour_ratio, gait_selected_mode =

extract_refined_gait_silhouette(

frame,

fg_mask,

gait_bbox,

gait_runtime,

min_contour_ratio=min_contour_ratio,

# Adapt gait identity thresholds based on silhouette *content*.

# Note: contour_ratio is based on the fg contour vs the original bbox area,

# and mostly reflects bbox tightness/refinement success; it's not a reliable

# silhouette quality metric once sil_norm has passed quality checks.

gait_thresh_eff = float(gait_thresh)

gait_margin_min_eff = float(gait_margin_min)

sil_fill = None

try:
 if sil_norm is not None and getattr(sil_norm, "size", 0) > 0:

sil_fill = float(np.count_nonzero(sil_norm)) / float(max(1, int(sil_norm.size)))

except Exception:

sil_fill = None

try:

mode_l = str(gait_selected_mode or "").lower()

if mode_l == "bgsub":

gait_thresh_eff = max(gait_thresh_eff, float(gait_thresh) + 0.02)

gait_margin_min_eff = max(gait_margin_min_eff, float(gait_margin_min) +

0.0005)

if sil_fill is not None:

# Extremely sparse/dense silhouettes are usually noisy.

if sil_fill < 0.08 or sil_fill > 0.78:

gait_thresh_eff = max(gait_thresh_eff, float(gait_thresh) + 0.05)

gait_margin_min_eff = max(gait_margin_min_eff, float(gait_margin_min) +

0.0010)

elif sil_fill < 0.12 or sil_fill > 0.70:

gait_thresh_eff = max(gait_thresh_eff, float(gait_thresh) + 0.03)

gait_margin_min_eff = max(gait_margin_min_eff, float(gait_margin_min) +

0.0005)

gait_margin_min_eff = float(min(gait_margin_min_eff, 0.01))

except Exception:

gait_thresh_eff = float(gait_thresh)

gait_margin_min_eff = float(gait_margin_min)

if sil_norm is None:

gait_reject_streaks[ci][tid] = int(gait_reject_streaks[ci].get(tid, 0)) + 1

 if debug_gait:

print(

f"[GAIT DEBUG] cam={ci} tid={tid} rejected silhouette bbox={gait_bbox}

contour_ratio={contour_ratio:.4f} mode={gait_selected_mode}"

else:

gait_reject_streaks[ci][tid] = 0

gait_buffers[ci][tid].append(sil_norm)

try:

gait_append_counts[ci][tid] = int(gait_append_counts[ci].get(tid, 0)) + 1

except Exception:

pass

if debug_gait and (len(gait_buffers[ci][tid]) % 10 == 0 or len(gait_buffers[ci][tid])

< 5):

print(

f"[GAIT DEBUG] cam={ci} tid={tid} appended sil_norm;

buffer_len={len(gait_buffers[ci][tid])} contour_ratio={contour_ratio:.4f}
mode={gait_selected_mode}"

# If we keep rejecting silhouettes for a track, clear gait buffer to avoid using stale
frames

# that might produce consistent but wrong IDs.

try:

if int(gait_reject_streaks[ci].get(tid, 0)) >= 12:

gait_buffers[ci][tid].clear()

if tid in gait_embed_histories[ci]:
 gait_embed_histories[ci][tid].clear()

try:

gait_append_counts[ci].pop(tid, None)

gait_last_extract_counts[ci].pop(tid, None)

except Exception:

pass

gait_reject_streaks[ci][tid] = 0

if debug_gait:

print(f"[GAIT DEBUG] cam={ci} tid={tid} cleared gait buffer due to reject

streak")

except Exception:

pass

# Throttle extraction: only extract when we have enough new accepted

silhouettes,

# and avoid repeatedly extracting from stale buffers when current frames are
rejected.

can_extract = False

try:

append_cnt = int(gait_append_counts[ci].get(tid, 0))

last_cnt = int(gait_last_extract_counts[ci].get(tid, 0))

if int(gait_reject_streaks[ci].get(tid, 0)) == 0 and (append_cnt - last_cnt) >= 6:

can_extract = True

except Exception:

can_extract = True

if can_extract and len(gait_buffers[ci][tid]) >= max(12, int(gait_ready_frames)):

 seq = [Link](list(gait_buffers[ci][tid]), axis=0)

if debug_gait:

print(f"[GAIT DEBUG] cam={ci} tid={tid} calling extract

[Link]={[Link]}")

try:

g = gait_runtime["extract"](gait_runtime["model"], gait_runtime["eval_trfs"],
seq)

g_arr = [Link](g)

if debug_gait:

print(f"[GAIT DEBUG] cam={ci} tid={tid} extract returned

shape={g_arr.shape}")

gait_emb = gait_embed_to_vec(g_arr)

try:

gait_last_extract_counts[ci][tid] = int(gait_append_counts[ci].get(tid, 0))

except Exception:

pass

# store recent gait embedding for fallback use

if tid not in gait_embed_histories[ci]:

gait_embed_histories[ci][tid] = deque(maxlen=6)

gait_embed_histories[ci][tid].append(gait_emb)

# compute similarity to gait DB (always compute top match even if below

thresholds)

top_matches = cosine_topk(gait_emb, gait_db, k=2)

if top_matches:

idx, gait_score = top_matches[0]

second_score = float(top_matches[1][1]) if len(top_matches) > 1 else 0.0

 gait_margin = float(gait_score - second_score) if len(top_matches) > 1 else
float(gait_score)

if idx >= 0 and idx < len(gait_labels):

gait_label_top1 = gait_labels[idx]

# keep a light history of top-1 predictions even if they are below strict
thresholds

try:

gait_top1_histories[ci][tid].append({

"label_id": gait_label_top1,

"cos_sim": float(gait_score),

"margin": float(gait_margin),

})

except Exception:

pass

# only append to history if it passes identity thresholds

if gait_label_top1 != "unknown" and gait_score >= float(gait_thresh_eff) and

gait_margin >= float(gait_margin_min_eff):

gait_label_histories[ci][tid].append({

"label_id": gait_label_top1,

"cos_sim": float(gait_score),

"margin": float(gait_margin),

})

if debug_gait:

print(

f"[GAIT DEBUG] cam={ci} tid={tid} gait_label={gait_label_top1}

gait_score={gait_score:.4f} margin={gait_margin:.4f} mode={gait_selected_mode}"

)
 except Exception as e:

if debug_gait:

print(f"[GAIT DEBUG] cam={ci} tid={tid} extract exception: {e}")

gait_emb = None

gait_score = 0.0

gait_margin = 0.0

# If we don't have a fresh gait_score from current extraction, try to

# compute a fallback similarity from recent stored gait embeddings.

try:

if (gait_emb is None or float(gait_score) == 0.0) and tid in

gait_embed_histories[ci] and len(gait_embed_histories[ci][tid]) > 0:

arr_embs = [Link](list(gait_embed_histories[ci][tid]), dtype=np.float32)

mean_emb = l2_normalize([Link](arr_embs, axis=0, keepdims=True),

axis=1)[0]

top_matches_hist = cosine_topk(mean_emb, gait_db, k=2)

if top_matches_hist:

idx_hist, hist_score = top_matches_hist[0]

gait_score = float(hist_score)

second_score = float(top_matches_hist[1][1]) if len(top_matches_hist) > 1

else 0.0

gait_margin = float(gait_score - second_score) if len(top_matches_hist) >

1 else float(gait_score)

if idx_hist >= 0 and idx_hist < len(gait_labels):

# do not overwrite gait_label_top1 used for strict voting unless it passes

thresholds

if gait_label_top1 == "unknown":
 gait_label_top1 = gait_labels[idx_hist]

except Exception:

pass

history_items = list(gait_label_histories[ci][tid])

stable_gait_label, stable_gait_score = choose_stable_gait_identity(

history_items,

gait_thresh=gait_thresh_eff,

gait_vote_min=gait_vote_min,

gait_margin_min=gait_margin_min_eff,

gait_vote_margin=gait_vote_margin,

if stable_gait_label is not None and gait_label_top1 == stable_gait_label:

stable_gait_score = max(float(stable_gait_score), float(gait_score))

if stable_gait_label is None:

stable_gait_score = 0.0

gait_label_for_bank = stable_gait_label if stable_gait_label is not None else

"unknown"

# Soft-consistent gait label used ONLY for cross-camera identity binding when
face is unknown.

link_gait_label = None

link_gait_score = 0.0

try:

if stable_gait_label is None and tid in gait_top1_histories[ci]:

soft_thresh = max(0.0, float(gait_thresh) - 0.04)

 # For cross-camera linking, margin can be unreliable in bgsub/noisy
silhouettes.

# Require temporal consistency instead of a strict margin when in bgsub

mode.

if str(gait_selected_mode or "").lower() == "bgsub":

soft_margin = 0.0

else:

soft_margin = max(0.02, float(gait_margin_min_eff))

link_gait_label, link_gait_score = choose_stable_gait_identity(

list(gait_top1_histories[ci][tid]),

gait_thresh=soft_thresh,

gait_vote_min=max(4, int(gait_vote_min) + 1),

gait_margin_min=soft_margin,

gait_vote_margin=max(1, int(gait_vote_margin)),

except Exception:

link_gait_label = None

link_gait_score = 0.0

same_identity = False

if face_label != "unknown" and gait_label_for_bank != "unknown":

same_identity = normalize_identity_key(face_label) ==
normalize_identity_key(gait_label_for_bank)

# Determine effective face label/score (prefer stable face when present)

face_effective_label = (

stable_face_label if ("stable_face_label" in locals() and stable_face_label) else

face_label
 )

face_effective_score = (

float(stable_face_score) if ("stable_face_label" in locals() and

stable_face_label) else float(face_score)

# Stable face label/score used for banking and association bookkeeping

face_label_for_update = stable_face_label if ('stable_face_label' in locals() and

stable_face_label) else face_label

face_score_for_update = stable_face_score if ('stable_face_label' in locals() and

stable_face_label) else float(face_score)

# Compute fused score in all cases (prefer face override if requested)

# Use a relaxed gait fallback when stable_gait_label is not available so

# gait can still contribute (soft blending) even if it didn't reach

# strict voting thresholds used for identity selection.

gait_relax_factor = 0.6

gait_confidence = 0.0

try:

if stable_gait_label is not None and stable_gait_label != "unknown":

gait_confidence = float(stable_gait_score)

else:

gait_confidence = float(gait_score) * float(gait_relax_factor)

except Exception:

gait_confidence = 0.0

gait_confidence = max(0.0, min(1.0, gait_confidence))

 # Context-aware / neural alpha

face_area_ratio = 0.0

try:

if matched_face_bbox is not None:

face_area_ratio = float(bbox_area(tuple(map(int, matched_face_bbox)))) /

float(

max(1, bbox_area((x1, y1, x2, y2)))

except Exception:

face_area_ratio = 0.0

alpha_eff = float(alpha)

if fusion_mlp is not None and torch is not None:

try:

feats = [Link](

float(face_effective_score),

float(gait_confidence),

float(face_area_ratio),

float(contour_ratio),

1.0 if (stable_face_label is not None and stable_face_label != "unknown")

else 0.0,

1.0 if str(gait_selected_mode or "").lower() == "bgsub" else 0.0,

],

dtype=np.float32,

with torch.no_grad():
 x = torch.from_numpy([Link](1, -1))

y = fusion_mlp(x)

alpha_eff = float([Link](-1)[0].item())

except Exception:

alpha_eff = float(alpha)

elif dynamic_alpha:

alpha_eff = compute_context_alpha(

base_alpha=float(alpha),

face_label=str(face_effective_label),

face_score=float(face_effective_score),

face_thresh=float(face_thresh),

face_area_ratio=float(face_area_ratio),

stable_face=bool(stable_face_label is not None and stable_face_label !=

"unknown"),

contour_ratio=float(contour_ratio),

min_contour_ratio=float(min_contour_ratio),

gait_mode=str(gait_selected_mode) if gait_selected_mode is not None else

None,

alpha_eff = float([Link](alpha_eff, 0.15, 0.98))

if prefer_face and face_effective_label != "unknown" and

float(face_effective_score) >= float(face_thresh):

fused = float(face_effective_score)

else:

# When both modalities provide some confidence, blend them.

 if (face_effective_label != "unknown" and float(face_effective_score) > 0.0) and
gait_confidence > 0.0:

fused = float(alpha_eff) * float(face_effective_score) + (1.0 - float(alpha_eff)) *

float(gait_confidence)

# If only face available use it, otherwise fallback to gait_confidence

elif face_effective_label != "unknown" and float(face_effective_score) > 0.0:

fused = float(face_effective_score)

else:

fused = float(gait_confidence)

# MTMCT global id assignment

app_emb = None

track_embs = trackers[ci].get_embeddings(tid)

if track_embs:

app_emb = l2_normalize([Link]([Link](track_embs, dtype=np.float32),

axis=0, keepdims=True), axis=1)[0]

obs_ts = float(frame_ts if frame_ts > 0 else [Link]())

# prefer a stable face label when available to avoid transient flips

preferred = None

preferred_source = None

if 'stable_face_label' in locals() and stable_face_label:

preferred = stable_face_label

preferred_source = "face"

elif face_label != "unknown":

preferred = face_label

preferred_source = "face"

elif stable_gait_label is not None and stable_gait_label != "unknown":

 preferred = stable_gait_label

preferred_source = "gait"

elif link_gait_label is not None and link_gait_label != "unknown":

preferred = link_gait_label

preferred_source = "gait"

existing_gid = cam_track_gid[ci].get(tid)

assoc_reason = "track"

assoc_score = 1.0

if existing_gid:

if preferred and preferred != "unknown":

# Don't let a gait-based preference override an already face-bound identity.

try:

existing_meta = bank._ensure_meta(existing_gid)

if str(preferred_source or "").lower() == "gait" and

existing_meta.get("label_source") == "face":

preferred = None

preferred_source = None

except Exception:

pass

if preferred and preferred != "unknown":

preferred_gid = f"ID_{preferred}"

if existing_gid != preferred_gid:

if not bank._has_conflicting_identity(bank._ensure_meta(existing_gid),
preferred):

bank._promote_gid(existing_gid, preferred_gid)

existing_gid = preferred_gid
 meta = bank._ensure_meta(existing_gid)

meta["identity_key"] = normalize_identity_key(preferred)

meta["display_name"] = resolve_display_name(preferred, name_map)

meta["label_source"] = (preferred_source or ("face" if face_label != "unknown"

else "fused"))

if app_emb is not None:

bank._update(existing_gid, app_emb)

[Link](

existing_gid,

ci,

(x1, y1, x2, y2),

[Link],

now=obs_ts,

assoc_reason=(preferred_source or "track"),

assoc_score=(float(face_score_for_update) if preferred_source == "face"

else float(stable_gait_score if stable_gait_label is not None else gait_score)),

transition_margin_ratio=cross_cam_transition_margin,

gid = existing_gid

assoc_reason = str(preferred_source or "track")

assoc_score = float(face_score_for_update) if preferred_source == "face" else

float(stable_gait_score if stable_gait_label is not None else gait_score)

else:

gid = existing_gid

if app_emb is not None:

bank._update(gid, app_emb)

prev_assoc = cam_track_assoc[ci].get(tid, {})

 assoc_reason = str(prev_assoc.get("reason", "track"))

assoc_score = float(prev_assoc.get("score", 1.0))

[Link](

gid,

ci,

(x1, y1, x2, y2),

[Link],

now=obs_ts,

assoc_reason=assoc_reason,

assoc_score=assoc_score,

transition_margin_ratio=cross_cam_transition_margin,

else:

gid, assoc_reason, assoc_score = [Link](

app_emb,

cam_idx=ci,

bbox=(x1, y1, x2, y2),

frame_shape=[Link],

preferred_label=preferred,

preferred_source=preferred_source,

sync_id=sync_id,

reid_threshold=cross_cam_reid_thresh,

overlap_threshold=cross_cam_overlap_thresh,

active_window_sec=cross_cam_active_window_sec,

min_transition_sec=cross_cam_min_transition_sec,

max_transition_sec=cross_cam_max_transition_sec,
 transition_margin_ratio=cross_cam_transition_margin,

now=obs_ts,

bank.update_identity(

gid=gid,

cam_idx=ci,

face_label=face_label_for_update,

face_score=face_score_for_update,

gait_label=gait_label_for_bank,

gait_score=stable_gait_score,

fused_id="unknown",

fused_name="unknown",

fused_score=0.0,

face_thresh=face_thresh,

gait_thresh=gait_thresh,

now=obs_ts,

cam_track_gid[ci][tid] = gid

cam_track_assoc[ci][tid] = {

"reason": assoc_reason,

"score": float(assoc_score),

bank_identity_key, bank_display_name, bank_label_source =

bank.get_identity(gid)

allow_bank_display = False

gait_display_score = float(stable_gait_score) if stable_gait_label is not None else

float(gait_score)
 if bank_display_name:

if bank_label_source == "face":

allow_bank_display = face_label != "unknown"

elif bank_label_source == "gait":

# allow gait-labeled global identity display when we have at least a weak gait
signal

allow_bank_display = bank_identity_key is not None and gait_display_score

>= max(0.0, float(gait_thresh) - 0.03)

if sync_id and bank_display_name and assoc_reason in ("sync-overlap", "sync-

transition"):

allow_bank_display = True

# draw bbox and labels with optional cross-camera sync association

if allow_bank_display:

display = bank_display_name

if assoc_reason in ("sync-overlap", "sync-transition"):

source_tag = "SYNC"

else:

source_tag = bank_label_source.upper() if bank_label_source else "GLOBAL"

elif face_label != "unknown":

display = resolve_display_name(face_label, name_map)

source_tag = "FACE"

elif stable_gait_label is not None and stable_gait_score >= float(gait_thresh_eff):

display = resolve_display_name(stable_gait_label, name_map)

source_tag = "GAIT"

else:

display = "unknown"
 source_tag = "TRACK"

bbox_color = (0, 0, 255) if display == "unknown" else (0, 255, 0)

[Link](vis, (x1, y1), (x2, y2), bbox_color, 2)

txt = f"C{ci} T{tid} {display} [{source_tag}] | F:{face_score:.2f}

G:{gait_display_score:.2f} S:{fused:.2f}"

[Link](vis, txt, (x1, max(20, y1 - 8)), cv2.FONT_HERSHEY_SIMPLEX, 0.5,

bbox_color, 2)

# Close-up face mode: if body box is absent but face is large enough, draw face
only.

for face_idx, face_info in enumerate(frame_faces):

if face_idx in used_face_indices:

continue

fx1, fy1, fx2, fy2 = map(int, face_info["bbox"])

face_area_ratio = bbox_area((fx1, fy1, fx2, fy2)) / float(max(1, frame_area))

if face_area_ratio < 0.04:

continue

overlaps_body = False

for t in tracks:

tx1, ty1, tx2, ty2 = map(int, t["bbox"])

if face_overlap_ratio((fx1, fy1, fx2, fy2), (tx1, ty1, tx2, ty2)) >= 0.40:

overlaps_body = True

break

if overlaps_body:

continue
 f_idx, f_score = cosine_top1(face_info["emb"], face_db)

face_label = "unknown"

display_name = "unknown"

if f_idx >= 0 and f_score >= face_thresh:

face_label = face_labels[f_idx]

display_name = resolve_display_name(face_label, name_map)

[Link](vis, (fx1, fy1), (fx2, fy2), (0, 180, 255), 2)

txt = f"C{ci} FACE {display_name} | F:{f_score:.2f}"

[Link](vis, txt, (fx1, max(20, fy1 - 8)), cv2.FONT_HERSHEY_SIMPLEX, 0.55,

(0, 180, 255), 2)

age_ms = max(0.0, ([Link]() - frame_ts) * 1000.0)

[Link](vis, f"CAM {ci} age={age_ms:.0f}ms", (10, 30),

cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 200, 255), 2)

# store previous fg mask for temporal checks

try:

prev_fg_masks[ci] = fg_mask.copy()

except Exception:

pass

last_views[ci] = [Link]()

[Link](vis)

if not views:

continue

# Optional: correlation-clustering-style merge to reduce cross-camera ID splits

if cc_merge:

try:

now_ts = [Link]()

if (now_ts - float(last_cc_merge_ts)) >= float(cc_merge_period_sec):

merged_map = bank.correlation_cluster_merge(

now=now_ts,

active_window_sec=float(cc_merge_window_sec),

sim_threshold=float(cc_merge_sim),

if merged_map:

for mci in range(len(cam_track_gid)):

for tid, gid in list(cam_track_gid[mci].items()):

if gid in merged_map:

cam_track_gid[mci][tid] = merged_map[gid]

last_cc_merge_ts = float(now_ts)

except Exception:

pass

# Combined MTMCT dashboard

max_h = max([Link][0] for v in views)

resized = []

for v in views:

if [Link][0] != max_h:

scale = max_h / float([Link][0])

nw = int([Link][1] * scale)

v = [Link](v, (nw, max_h))

 [Link](v)

board = [Link](resized)

sync_mode = "ON" if sync_id else "OFF"

sync_text = f"SYNC {sync_mode} waiting" if not [Link](sync_delta_ms) else

f"SYNC {sync_mode} delta={sync_delta_ms:.0f}ms"

[Link](board, sync_text, (10, max(30, [Link][0] - 20)),

cv2.FONT_HERSHEY_SIMPLEX, 0.8, (50, 220, 50), 2)

[Link]("Fusion MTMCT", board)

if [Link](1) & 0xFF == ord("q"):

break

finally:

for reader in readers:

[Link]()

[Link]()

def run_realtime(alpha: float, face_thresh: float, fuse_thresh: float, camera: int, use_gait:
bool, gait_cfg: str):

face_db, face_labels = load_face_db()

gait_db, gait_labels = load_gait_db()

name_map = load_name_map()

gait_runtime = try_init_gait_runtime(gait_cfg) if use_gait else None

if use_gait:

if gait_db is None or len(gait_labels) == 0:

raise RuntimeError("Da bat gait fusion nhung gait DB trong/khong ton tai. Kiem tra
OpenGait/output/.../gait_system_db.pkl")

if gait_runtime is None:
 raise RuntimeError("Da bat gait fusion nhung OpenGait runtime khong khoi tao duoc.
Kiem tra torch/opengait config")

detector, backbone, quality = build_face_models()

gait_buffer = deque(maxlen=30)

cap = [Link](camera)

if not [Link]():

raise RuntimeError(f"Khong mo duoc camera: {camera}")

print("[INFO] Bat dau nhan dien realtime. Nhan 'q' de thoat.")

print(f"[INFO] Face DB: {len(face_labels)} mau | Gait DB: {len(gait_labels)} mau")

configure_fullscreen_window("Fusion Realtime")

try:

while True:

ok, frame = [Link]()

if not ok or frame is None:

continue

img = [Link]()

h, w = [Link][:2]

bboxes, kpss = take_box_detector(img, detector)

if bboxes is not None and len(bboxes) > 0:

# Lay khuon mat co score cao nhat trong frame

best = max(range([Link][0]), key=lambda i: float(bboxes[i][4]))

 bbox = bboxes[best]

x1, y1, x2, y2, _ = [Link](int)

x1 = max(0, x1)

y1 = max(0, y1)

x2 = min(w - 1, x2)

y2 = min(h - 1, y2)

if x2 > x1 and y2 > y1:

crop = img[y1:y2, x1:x2]

try:

kps = kpss[best]

_, _, _, _, _, _, _, _, l_eye, r_eye = process_kps(kps)

aligned = alignment(crop, l_eye, r_eye)

aligned = [Link](aligned, (112, 112))

face_q, emb_t = process_onnx(aligned, backbone, quality)

emb = emb_t.cpu().detach().numpy().astype(np.float32)[0]

except Exception:

emb = None

face_q = [0.0]

if emb is not None:

f_idx, f_score = cosine_top1(emb, face_db)

face_label = face_labels[f_idx] if (f_idx >= 0 and f_score >= face_thresh) else

"unknown"

gait_label = "unknown"
 gait_score = 0.0

gait_ready = False

if gait_runtime is not None and gait_db is not None and len(gait_labels) > 0:

gray = [Link](img, cv2.COLOR_BGR2GRAY)

_, sil = [Link](gray, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)

sil_norm = gait_runtime["preprocess"](sil)

if sil_norm is not None:

gait_buffer.append(sil_norm)

if len(gait_buffer) >= 15:

seq = [Link](list(gait_buffer), axis=0)

g_emb = gait_runtime["extract"](gait_runtime["model"],
gait_runtime["eval_trfs"], seq)

g_emb = [Link](g_emb, dtype=np.float32).reshape(1, -1)[0]

g_idx, gait_score = cosine_top1(g_emb, gait_db)

if g_idx >= 0:

gait_label = gait_labels[g_idx]

gait_ready = True

# Score-level fusion

if use_gait and not gait_ready:

# Gait chua du frame, tam thoi hien thi nhan dien face de tranh startup qua
lau.

fused = f_score

final_label = face_label if f_score >= face_thresh else "unknown"

else:

fused = alpha * f_score + (1.0 - alpha) * gait_score

if fused >= fuse_thresh:

 if face_label != "unknown" and (gait_label == "unknown" or f_score >=
gait_score):

final_label = face_label

elif gait_label != "unknown":

final_label = gait_label

else:

final_label = "unknown"

else:

final_label = "unknown"

display_name = resolve_display_name(final_label, name_map)

[Link](img, (x1, y1), (x2, y2), (0, 255, 0), 2)

gait_state = "ON" if gait_ready else ("WARM" if use_gait else "OFF")

txt = f"{display_name} | F:{f_score:.2f} G:{gait_score:.2f} S:{fused:.2f}

{gait_state}"

[Link](img, txt, (x1, max(20, y1 - 10)), cv2.FONT_HERSHEY_SIMPLEX, 0.6,

(0, 255, 0), 2)

[Link]("Fusion Realtime", img)

if [Link](1) & 0xFF == ord("q"):

break

finally:

[Link]()

[Link]()
def parse_args():

parser = [Link](description="Full multimodal fusion + MTMCT")

parser.add_argument("--cameras", type=str, default="0,1", help="Danh sach camera, vd:

0,1 hoac rtsp://...,1")

parser.add_argument("--alpha", type=float, default=0.7, help="Trong so face score")

parser.add_argument("--face-thresh", type=float, default=0.55)

parser.add_argument("--gait-thresh", type=float, default=0.43)

parser.add_argument("--fuse-thresh", type=float, default=0.55)

parser.add_argument("--det-conf", type=float, default=0.45)

parser.add_argument("--min-height", type=int, default=64)

parser.add_argument("--min-area", type=int, default=2000)

parser.add_argument("--min-fg-ratio", type=float, default=0.02)

parser.add_argument("--min-fg-motion", type=float, default=0.002, help="Min fraction of

pixel change between frames inside bbox to accept detection (temporal motion)")

parser.add_argument("--min-aspect", type=float, default=1.0)

parser.add_argument("--min-contour-ratio", type=float, default=0.10)

parser.add_argument("--max-box-area-ratio", type=float, default=0.80)

parser.add_argument("--max-box-width-ratio", type=float, default=0.90)

parser.add_argument("--face-min-size", type=int, default=32)

parser.add_argument("--face-body-overlap", type=float, default=0.55)

parser.add_argument("--person-face-region-ratio", type=float, default=0.65)

parser.add_argument("--display-sync-ms", type=int, default=80)

parser.add_argument("--display-wait-ms", type=int, default=120)

parser.add_argument("--bbox-history", type=int, default=5)

parser.add_argument("--gait-ready-frames", type=int, default=24)

parser.add_argument("--gait-history-maxlen", type=int, default=12)

parser.add_argument("--gait-vote-min", type=int, default=3)

 parser.add_argument("--gait-margin-min", type=float, default=0.001)

parser.add_argument("--gait-vote-margin", type=int, default=1)

parser.set_defaults(sync_id=True)

parser.add_argument("--sync-id", dest="sync_id", action="store_true", help="Bat cross-

camera synchronized identity")

parser.add_argument("--no-sync-id", dest="sync_id", action="store_false", help="Tat

cross-camera synchronized identity")

parser.add_argument("--cross-cam-reid-thresh", type=float, default=0.84)

parser.add_argument("--cross-cam-overlap-thresh", type=float, default=0.90)

parser.add_argument("--cross-cam-active-window-sec", type=float, default=1.20)

parser.add_argument("--cross-cam-min-transition-sec", type=float, default=0.10)

parser.add_argument("--cross-cam-max-transition-sec", type=float, default=4.00)

parser.add_argument("--cross-cam-transition-margin", type=float, default=0.16)

parser.add_argument("--debug-gait", action="store_true")

parser.add_argument("--prefer-face", action="store_true", help="Prefer face label when

confident (override gait)")

parser.add_argument("--dynamic-alpha", action="store_true", help="Context-aware

alpha for face/gait blending")

parser.add_argument("--fusion-mlp", type=str, default=None, help="Optional PyTorch

model to predict alpha (in [0,1])")

parser.add_argument("--cc-merge", action="store_true", help="Enable correlation-

clustering-style merge of similar global IDs")

parser.add_argument("--cc-merge-sim", type=float, default=0.93, help="Similarity

threshold for cc-merge")

parser.add_argument("--cc-merge-window-sec", type=float, default=3.0, help="Recent

activity window for cc-merge")

parser.add_argument("--cc-merge-period-sec", type=float, default=1.0, help="How often

to run cc-merge")

parser.add_argument(
 "--gait-cfg",

type=str,

default=[Link](core.OPEN_GAIT_ROOT, "configs", "gaitgl", "[Link]"),

parser.add_argument(

"--gait-dist-port",

type=int,

default=0,

help="TCP port for OpenGait single-process [Link] init. Use 0 to auto-pick a

free localhost port (recommended)",

parser.add_argument(

"--gait-db",

type=str,

default=None,

help="Optional gait DB .pkl path (override default output/CASIA-

B/GaitGL/gait_system_db.pkl)",

parser.add_argument("--yolo", type=str, default=[Link](core.ROOT_DIR,

"DA_RobotGuide", "[Link]"))

return parser.parse_args()

if __name__ == "__main__":

args = parse_args()

cams = parse_cameras(str([Link]))

print(f"[INFO] Parsed cameras: {cams}")

 if len(cams) < 2:

print("[WARN] Dang chay duoi 2 camera. De test MTMCT, hay dung --cameras 0,1 hoac
2 nguon video khac nhau.")

# Convenience defaults: when you explicitly request gait debug,

# enable the new fusion/merge behaviors without extra flags.

if bool(getattr(args, "debug_gait", False)):

if not bool(getattr(args, "dynamic_alpha", False)):

args.dynamic_alpha = True

if not bool(getattr(args, "cc_merge", False)):

args.cc_merge = True

run_realtime_mtmct(

alpha=float([Link]),

face_thresh=float(args.face_thresh),

gait_thresh=float(args.gait_thresh),

fuse_thresh=float(args.fuse_thresh),

cameras=cams,

gait_cfg=str(args.gait_cfg),

gait_dist_port=int(getattr(args, 'gait_dist_port', 0)),

gait_db_path=(None if args.gait_db in (None, "") else str(args.gait_db)),

yolo_model_path=str([Link]),

dynamic_alpha=bool(getattr(args, 'dynamic_alpha', False)),

fusion_mlp_path=(None if getattr(args, 'fusion_mlp', None) in (None, "") else

str(getattr(args, 'fusion_mlp'))),

cc_merge=bool(getattr(args, 'cc_merge', False)),

cc_merge_sim=float(getattr(args, 'cc_merge_sim', 0.93)),

cc_merge_window_sec=float(getattr(args, 'cc_merge_window_sec', 3.0)),

cc_merge_period_sec=float(getattr(args, 'cc_merge_period_sec', 1.0)),

det_conf=float(args.det_conf),

min_height=int(args.min_height),

min_area=int(args.min_area),

min_fg_ratio=float(args.min_fg_ratio),

min_aspect=float(args.min_aspect),

face_min_size=int(args.face_min_size),

face_body_overlap=float(args.face_body_overlap),

person_face_region_ratio=float(args.person_face_region_ratio),

display_sync_ms=int(args.display_sync_ms),

display_wait_ms=int(args.display_wait_ms),

bbox_history=int(args.bbox_history),

gait_ready_frames=int(args.gait_ready_frames),

gait_history_maxlen=int(args.gait_history_maxlen),

gait_vote_min=int(args.gait_vote_min),

gait_margin_min=float(args.gait_margin_min),

gait_vote_margin=int(args.gait_vote_margin),

sync_id=bool(args.sync_id),

cross_cam_reid_thresh=float(args.cross_cam_reid_thresh),

cross_cam_overlap_thresh=float(args.cross_cam_overlap_thresh),

cross_cam_active_window_sec=float(args.cross_cam_active_window_sec),

cross_cam_min_transition_sec=float(args.cross_cam_min_transition_sec),

cross_cam_max_transition_sec=float(args.cross_cam_max_transition_sec),

cross_cam_transition_margin=float(args.cross_cam_transition_margin),

debug_gait=bool(args.debug_gait),

prefer_face=bool(getattr(args, 'prefer_face', False)),

 min_contour_ratio=float(args.min_contour_ratio),

max_box_area_ratio=float(args.max_box_area_ratio),

max_box_width_ratio=float(args.max_box_width_ratio),

min_fg_motion=float(args.min_fg_motion),