warren/momentry_core
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases 2 Wiki Activity

fix: correct TKG node_type/edge_type mismatches in scan.rs and tkg.rs

Browse Source 
Mismatches fixed (builder vs query layer):
- 'text_region' -> 'text_trace' (3 locations: scan.rs count + tkg.rs LIP_SYNC
  query + tkg.rs LIP_SYNC node ID lookup)
- 'yolo_object' -> 'object' (2 queries in scan.rs)
- 'FACE_TO_FACE' -> CO_OCCURS_WITH + JOIN with tkg_nodes (2 queries in
  scan.rs, so face_face_edges correctly counts only face-track pairs)
- 'HAND_OBJECT' -> 'hand_object' (2 queries in scan.rs)

Impact: ingestion_status and file stats endpoints now return accurate
counts; LIP_SYNC edges now resolve their target text nodes.
This commit is contained in:
Accusys
2026-07-11 08:40:24 +08:00
parent b98a362de5
commit 7635ad8d4d
3 changed files with 165 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/processor/tkg.rs
+137 -40
View File
@@ -37,7 +37,10 @@ async fn scroll_face_points(file_uuid: &str) -> Vec<FacePoint> {
// Retry with exponential backoff (3 attempts)
let mut last_err = None;
for attempt in 1..=3 {
match qdrant.scroll_all_points("_faces", filter.clone(), 500).await {
match qdrant
.scroll_all_points("_faces", filter.clone(), 500)
.await
{
Ok(pts) => {
if attempt > 1 {
tracing::info!(
@@ -150,7 +153,6 @@ async fn build_node_id_map(pool: &PgPool, file_uuid: &str) -> HashMap<(String, S
.collect()
}
// ── Phase 0: Populate trace_id from face.json ───────────────────────────────────────────────────────
async fn populate_face_detections_from_face_json(
@@ -758,7 +760,8 @@ pub async fn build_tkg(
if let Some(r) = redis.as_ref() {
publish_tkg_progress(r, file_uuid, &progress).await;
}
let n_lip = build_lip_track_nodes(pool, file_uuid, output_dir, &pose_data, &frame_index).await?;
let n_lip =
build_lip_track_nodes(pool, file_uuid, output_dir, &pose_data, &frame_index).await?;
progress.stats.lip_track_nodes = n_lip as i64;
progress.update_phase(
TkgPhase::LipTrackNodes,
@@ -881,7 +884,8 @@ pub async fn build_tkg(
if let Some(r) = redis.as_ref() {
publish_tkg_progress(r, file_uuid, &progress).await;
}
let e_co = build_co_occurrence_edges(pool, file_uuid, output_dir, &face_points, &node_id_map).await?;
let e_co =
build_co_occurrence_edges(pool, file_uuid, output_dir, &face_points, &node_id_map).await?;
progress.stats.co_occurrence_edges = e_co as i64;
progress.update_phase(
TkgPhase::CoOccurrenceEdges,
@@ -915,7 +919,8 @@ pub async fn build_tkg(
if let Some(r) = redis.as_ref() {
publish_tkg_progress(r, file_uuid, &progress).await;
}
let e_ff = build_face_face_edges(pool, file_uuid, &pose_data, &face_points, &node_id_map).await?;
let e_ff =
build_face_face_edges(pool, file_uuid, &pose_data, &face_points, &node_id_map).await?;
progress.stats.face_face_edges = e_ff as i64;
progress.update_phase(TkgPhase::FaceFaceEdges, 1.0, &format!("{} face-face", e_ff));
if let Some(r) = redis.as_ref() {
@@ -930,7 +935,8 @@ pub async fn build_tkg(
if let Some(r) = redis.as_ref() {
publish_tkg_progress(r, file_uuid, &progress).await;
}
let e_mg = build_mutual_gaze_edges(pool, file_uuid, &pose_data, &face_points, &node_id_map).await?;
let e_mg =
build_mutual_gaze_edges(pool, file_uuid, &pose_data, &face_points, &node_id_map).await?;
progress.stats.mutual_gaze_edges = e_mg as i64;
progress.update_phase(
TkgPhase::MutualGazeEdges,
@@ -1383,7 +1389,8 @@ async fn build_co_occurrence_edges(
face_points: &[FacePoint],
node_id_map: &HashMap<(String, String), i64>,
) -> Result<usize> {
build_co_occurrence_edges_from_qdrant(pool, file_uuid, output_dir, face_points, node_id_map).await
build_co_occurrence_edges_from_qdrant(pool, file_uuid, output_dir, face_points, node_id_map)
.await
}
async fn build_co_occurrence_edges_from_qdrant(
@@ -1427,13 +1434,17 @@ async fn build_co_occurrence_edges_from_qdrant(
}
// Use pre-built node ID map (no SQL query)
let face_node_id = match node_id_map.get(&("face_track".to_string(), format!("face_track_{}", face.trace_id))) {
let face_node_id = match node_id_map.get(&(
"face_track".to_string(),
format!("face_track_{}", face.trace_id),
)) {
Some(id) => *id,
None => continue,
};
for det in dets {
let obj_node_id = match node_id_map.get(&("object".to_string(), det.class_name.clone())) {
let obj_node_id = match node_id_map.get(&("object".to_string(), det.class_name.clone()))
{
Some(id) => *id,
None => continue,
};
@@ -1443,7 +1454,13 @@ async fn build_co_occurrence_edges_from_qdrant(
"object_confidence": det.confidence,
});
edges_to_insert.push(("CO_OCCURS_WITH", face_node_id, obj_node_id, file_uuid, serde_json::to_string(&edge_props)?));
edges_to_insert.push((
"CO_OCCURS_WITH",
face_node_id,
obj_node_id,
file_uuid,
serde_json::to_string(&edge_props)?,
));
}
}
@@ -1458,7 +1475,8 @@ async fn build_co_occurrence_edges_from_qdrant(
if i > 0 {
query.push(", ");
}
query.push("(")
query
.push("(")
.push_bind(*edge_type)
.push_bind(*src)
.push_bind(*tgt)
@@ -1535,10 +1553,11 @@ async fn build_speaker_face_edges_from_pg(
for (tid, sf, ef) in &traces {
// Use pre-built node ID map (no SQL query)
let face_node_id = match node_id_map.get(&("face_track".to_string(), format!("face_track_{}", tid))) {
Some(id) => *id,
None => continue,
};
let face_node_id =
match node_id_map.get(&("face_track".to_string(), format!("face_track_{}", tid))) {
Some(id) => *id,
None => continue,
};
let face_start_sec = *sf as f64 / fps;
let face_end_sec = *ef as f64 / fps;
@@ -1565,10 +1584,11 @@ async fn build_speaker_face_edges_from_pg(
}
// Use pre-built node ID map (no SQL query)
let speaker_node_id = match node_id_map.get(&("speaker".to_string(), seg.speaker_id.clone())) {
Some(id) => *id,
None => continue,
};
let speaker_node_id =
match node_id_map.get(&("speaker".to_string(), seg.speaker_id.clone())) {
Some(id) => *id,
None => continue,
};
let edge_props = serde_json::json!({
"overlap_ratio": (overlap_ratio * 1000.0).round() / 1000.0,
@@ -2410,7 +2430,7 @@ async fn build_lip_sync_edges(
(properties->>'start_frame')::bigint,
(properties->>'end_frame')::bigint,
properties->>'speaker_id'
FROM {} WHERE file_uuid = $1 AND node_type = 'text_region'
FROM {} WHERE file_uuid = $1 AND node_type = 'text_trace'
"#,
nodes_table
))
@@ -2472,7 +2492,7 @@ async fn build_lip_sync_edges(
Some(id) => *id,
None => {
if let Some((id,)) = sqlx::query_as::<_, (i64,)>(&format!(
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='text_region' AND external_id=$2",
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='text_trace' AND external_id=$2",
nodes_table
))
.bind(file_uuid).bind(text_ext).fetch_optional(pool).await?
@@ -2560,9 +2580,9 @@ async fn build_appearance_trace_nodes(
if let Some(nose) = nose {
let nx = nose.get("x").and_then(|v| v.as_f64()).unwrap_or(0.0);
let ny = nose.get("y").and_then(|v| v.as_f64()).unwrap_or(0.0);
// Find closest face trace in Qdrant for this frame (in-memory lookup)
if let Some(trace_id) =
find_trace_in_frame(frame_index, frame_num, nx, ny, 1.0, 1.0)
// Find closest face trace in Qdrant for this frame (in-memory lookup)
if let Some(trace_id) =
find_trace_in_frame(frame_index, frame_num, nx, ny, 1.0, 1.0)
{
trace_data
.entry(trace_id)
@@ -2689,7 +2709,11 @@ async fn build_accessory_nodes(pool: &PgPool, file_uuid: &str, output_dir: &str)
// ── HAS_APPEARANCE Edges ──────────────────────────────────────────
async fn build_has_appearance_edges(pool: &PgPool, file_uuid: &str, _node_id_map: &HashMap<(String, String), i64>) -> Result<usize> {
async fn build_has_appearance_edges(
pool: &PgPool,
file_uuid: &str,
_node_id_map: &HashMap<(String, String), i64>,
) -> Result<usize> {
let nodes_table = t("tkg_nodes");
let edges_table = t("tkg_edges");
@@ -3181,10 +3205,38 @@ mod tests {
#[test]
fn test_build_frame_index() {
let points = vec![
FacePoint { trace_id: 1, frame: 0, x: 100.0, y: 100.0, w: 50.0, h: 50.0 },
FacePoint { trace_id: 1, frame: 0, x: 200.0, y: 200.0, w: 50.0, h: 50.0 },
FacePoint { trace_id: 2, frame: 1, x: 150.0, y: 150.0, w: 50.0, h: 50.0 },
FacePoint { trace_id: 1, frame: 2, x: 120.0, y: 120.0, w: 50.0, h: 50.0 },
FacePoint {
trace_id: 1,
frame: 0,
x: 100.0,
y: 100.0,
w: 50.0,
h: 50.0,
},
FacePoint {
trace_id: 1,
frame: 0,
x: 200.0,
y: 200.0,
w: 50.0,
h: 50.0,
},
FacePoint {
trace_id: 2,
frame: 1,
x: 150.0,
y: 150.0,
w: 50.0,
h: 50.0,
},
FacePoint {
trace_id: 1,
frame: 2,
x: 120.0,
y: 120.0,
w: 50.0,
h: 50.0,
},
];
let index = build_frame_index(&points);
@@ -3205,8 +3257,22 @@ mod tests {
#[test]
fn test_find_trace_in_frame_exact_match() {
let points = vec![
FacePoint { trace_id: 1, frame: 0, x: 100.0, y: 100.0, w: 100.0, h: 100.0 },
FacePoint { trace_id: 2, frame: 0, x: 300.0, y: 300.0, w: 100.0, h: 100.0 },
FacePoint {
trace_id: 1,
frame: 0,
x: 100.0,
y: 100.0,
w: 100.0,
h: 100.0,
},
FacePoint {
trace_id: 2,
frame: 0,
x: 300.0,
y: 300.0,
w: 100.0,
h: 100.0,
},
];
let index = build_frame_index(&points);
@@ -3218,8 +3284,22 @@ mod tests {
#[test]
fn test_find_trace_in_frame_closest_match() {
let points = vec![
FacePoint { trace_id: 1, frame: 0, x: 100.0, y: 100.0, w: 100.0, h: 100.0 },
FacePoint { trace_id: 2, frame: 0, x: 500.0, y: 500.0, w: 100.0, h: 100.0 },
FacePoint {
trace_id: 1,
frame: 0,
x: 100.0,
y: 100.0,
w: 100.0,
h: 100.0,
},
FacePoint {
trace_id: 2,
frame: 0,
x: 500.0,
y: 500.0,
w: 100.0,
h: 100.0,
},
];
let index = build_frame_index(&points);
@@ -3230,9 +3310,14 @@ mod tests {
#[test]
fn test_find_trace_in_frame_no_matching_frame() {
let points = vec![
FacePoint { trace_id: 1, frame: 0, x: 100.0, y: 100.0, w: 100.0, h: 100.0 },
];
let points = vec![FacePoint {
trace_id: 1,
frame: 0,
x: 100.0,
y: 100.0,
w: 100.0,
h: 100.0,
}];
let index = build_frame_index(&points);
// Query for frame 1 which doesn't exist
@@ -3256,9 +3341,21 @@ mod tests {
map.insert(("speaker".to_string(), "SPEAKER_0".to_string()), 200);
map.insert(("object".to_string(), "person".to_string()), 300);
assert_eq!(map.get(&("face_track".to_string(), "face_track_1".to_string())), Some(&100));
assert_eq!(map.get(&("speaker".to_string(), "SPEAKER_0".to_string())), Some(&200));
assert_eq!(map.get(&("object".to_string(), "person".to_string())), Some(&300));
assert_eq!(map.get(&("face_track".to_string(), "face_track_99".to_string())), None);
assert_eq!(
map.get(&("face_track".to_string(), "face_track_1".to_string())),
Some(&100)
);
assert_eq!(
map.get(&("speaker".to_string(), "SPEAKER_0".to_string())),
Some(&200)
);
assert_eq!(
map.get(&("object".to_string(), "person".to_string())),
Some(&300)
);
assert_eq!(
map.get(&("face_track".to_string(), "face_track_99".to_string())),
None
);
}
}