// scene_analyze — identify actors in a movie using a KPN pipeline // // TRACES: DP-001, DP-002 | PR-004 // One analysis core; the CLI is a front-end over it and must not fork pipeline // logic. Other deployment modes (DP-003, DP-004) wrap this same core. // // KPN topology (release build): // // [frame_source] ──Frame──► [face_detector] ──SceneFrame──► [face_aligner] // ──AlignedSceneFrame──► [embedder] ──EmbeddedSceneFrame──► // [identity_matcher] ──MatchedSceneFrame──► [scene_tracker] // ──SceneAnnotation──► [result_sink] // // Debug build (SAE_DEBUG=1): // [identity_matcher] output fans out to both [scene_tracker] AND [debug_renderer]. // FanoutNode is auto-inserted by make_network(). // // Usage: // scene_analyze --movie --gallery [options] // // Options: // --output output JSON (default: annotations.json) // --fps sample rate in frames/sec (default: 1.0) // --verbosity <0|1|2> 0=minimal, 1=standard, 2=jellyfin-xray (default: 0) // --match-threshold cosine dist threshold (default: 0.45) // --extinction actor extinction window in seconds (default: 5.0) // --detector override SCRFD detector model path // --arcface override ArcFace model path // --scene-detect enable TransNetV2 shot-boundary detection (dense decode; // writes .scenes.json). Off by default. // --scene-detector override TransNetV2 .onnx model path // --scene-detector-engine pre-built TransNetV2 TRT engine (TRT backend) // --scene-threshold boundary sigmoid prob above this → cut (default: 0.60) // --scene-stride frames between TransNetV2 windows (default: 50, ≤100) // --scene-decode-fps dense decode rate in scene-detect mode (default: 12; // 0 = native fps). Lower = faster, coarser boundaries. // --dense-scale downscale decoded frames in scene-detect mode (0 max faces kept per frame (default: 10) // --expand-gallery enable per-film gallery expansion from track continuity // --expand-buffer per-track diversity buffer size (default: 20) // --expand-novelty-sim promote only views with best sim < f (default: 0.55) // --expand-spread-max reject track if buffer spread > f (default: 0.60) // --expand-min-anchor accepted frames before a track confirms (default: 3) // --expand-debug-dir

dump promoted mugshots + embeddings here (SAE_DEBUG) // (SAE_DEBUG only) // --debug-dir debug frames output dir (default: debug_frames) // --crop-context bbox expansion factor for context crops (default: 1.5) #include "config.hpp" #include "types.hpp" #include "gallery/embedder_stamp.hpp" #include "gallery/gallery_store.hpp" #include "nodes/frame_source_node.hpp" #include "nodes/camera_position_change_detector_node.hpp" #include "nodes/face_detector_node.hpp" #include "nodes/face_aligner_node.hpp" #include "nodes/embedder_node.hpp" #include "nodes/face_tracker_node.hpp" #include "nodes/identity_matcher_node.hpp" #include "nodes/scene_tracker_node.hpp" #include "nodes/scene_detector_node.hpp" #include "scene_boundaries.hpp" #include "nodes/scene_boundary_annotator_node.hpp" #include "nodes/result_sink_node.hpp" #include "nodes/embedding_dump_node.hpp" #ifdef SAE_DEBUG #include "nodes/debug_renderer_node.hpp" #endif #include #include #include #include #include #include #include #include #include #include #include // ── CLI parsing ─────────────────────────────────────────────────────────────── /// TRACES: AR-010, AR-004 | SR-002 /// How deeply the sampled branch is buffered behind the dense one. TransNetV2 /// needs kWindow (100) dense frames before it can score any of them, so the face /// branch must lag by at least that much or it asks about frames nobody has /// looked at yet. Backpressure turns depth into lag: the fanout blocks on the /// slower branch rather than dropping, so the detector simply runs ahead. static constexpr std::size_t kSceneJoinDepth = 256; /// Set when the scene branch is built, so shutdown can report whether the join /// actually worked. static std::shared_ptr scene_stats; static Config parse_args(int argc, char** argv) { Config cfg; cfg.detector_model = kDefaultDetectorModel; cfg.arcface_model = kDefaultArcfaceModel; cfg.scene_model = kDefaultSceneModel; cfg.output_path = "annotations.json"; for (int i = 1; i < argc; ++i) { auto arg = [&](const char* flag) { return std::strcmp(argv[i], flag) == 0; }; auto next = [&]() -> std::string { if (++i >= argc) throw std::runtime_error(std::string("missing arg after ") + argv[i-1]); return argv[i]; }; if (arg("--movie")) cfg.movie_path = next(); else if (arg("--gallery")) cfg.gallery_path = next(); else if (arg("--output")) cfg.output_path = next(); else if (arg("--dump-embeddings")) cfg.dump_embeddings_path = next(); else if (arg("--fps")) cfg.sample_fps = std::stof(next()); else if (arg("--max-decode-fps")) cfg.max_decode_fps = std::stof(next()); else if (arg("--start")) cfg.start_sec = std::stod(next()); else if (arg("--end")) cfg.end_sec = std::stod(next()); else if (arg("--cut-threshold")) cfg.cut_threshold = std::stof(next()); else if (arg("--scene-detect")) cfg.scene_detect = true; else if (arg("--scene-detector")) cfg.scene_model = next(); else if (arg("--scene-detector-engine")) cfg.scene_engine = next(); else if (arg("--scene-threshold")) cfg.scene_threshold = std::stof(next()); else if (arg("--scene-stride")) cfg.scene_stride = std::stoi(next()); else if (arg("--scene-decode-fps")) cfg.scene_decode_fps = std::stof(next()); else if (arg("--dense-scale")) cfg.dense_scale = std::stof(next()); else if (arg("--verbosity")) { int v = std::stoi(next()); cfg.verbosity = v == 2 ? Verbosity::xray : v == 1 ? Verbosity::standard : Verbosity::minimal; } else if (arg("--prior")) cfg.match_prior = std::stof(next()); else if (arg("--prob-threshold")) cfg.prob_threshold = std::stof(next()); else if (arg("--match-threshold")) cfg.match_threshold = std::stof(next()); else if (arg("--extinction")) cfg.extinction_sec = std::stod(next()); else if (arg("--detector")) cfg.detector_model = next(); else if (arg("--detector-engine")) cfg.detector_engine = next(); else if (arg("--arcface")) cfg.arcface_model = next(); else if (arg("--require-gallery-stamp")) cfg.require_gallery_stamp = true; else if (arg("--arcface-engine")) cfg.arcface_engine = next(); else if (arg("--conf")) cfg.detector_conf = std::stof(next()); else if (arg("--max-faces")) cfg.max_faces = std::stoi(next()); else if (arg("--min-face-px")) cfg.min_face_px = std::stof(next()); else if (arg("--ratio")) cfg.match_ratio = std::stof(next()); else if (arg("--ratio-ceil")) cfg.match_ratio_ceil = std::stof(next()); else if (arg("--track-alpha")) cfg.track_alpha = std::stof(next()); else if (arg("--track-min-iou")) cfg.track_min_iou = std::stof(next()); else if (arg("--track-min-prob")) cfg.track_assoc_min_prob = std::stof(next()); else if (arg("--track-extinction")) cfg.track_extinction_sec = std::stod(next()); else if (arg("--anneal")) cfg.anneal_sec = std::stod(next()); else if (arg("--expand-gallery")) cfg.expand_gallery = true; else if (arg("--expand-buffer")) cfg.expand_buffer_size = std::stoi(next()); else if (arg("--expand-novelty-sim")) cfg.expand_novelty_sim = std::stof(next()); else if (arg("--expand-spread-max")) cfg.expand_track_spread_max = std::stof(next()); else if (arg("--expand-min-anchor")) cfg.expand_min_anchor_frames = std::stoi(next()); else if (arg("--expand-debug-dir")) cfg.expand_debug_dir = next(); else if (arg("--trt-cache")) cfg.trt.cache_dir = next(); else if (arg("--trt-fp16")) cfg.trt.fp16 = true; else if (arg("--no-trt-fp16")) cfg.trt.fp16 = false; else if (arg("--trt-int8")) cfg.trt.int8 = true; else if (arg("--embed-batch")) cfg.embed_batch_size = std::stoi(next()); #ifdef SAE_DEBUG else if (arg("--debug-dir")) cfg.debug_dir = next(); else if (arg("--crop-context")) cfg.crop_context = std::stof(next()); #endif else { std::cerr << "[warn] unknown flag: " << argv[i] << "\n"; } } if (cfg.movie_path.empty()) throw std::runtime_error("--movie is required"); if (cfg.gallery_path.empty()) throw std::runtime_error("--gallery is required"); return cfg; } // ── Main ────────────────────────────────────────────────────────────────────── int main(int argc, char** argv) { Config cfg; try { cfg = parse_args(argc, argv); } catch (const std::exception& e) { std::cerr << "Usage error: " << e.what() << "\n"; return 1; } // Load actor gallery ActorGallery gallery; try { gallery = load_gallery(cfg.gallery_path); /// TRACES: GR-004 | SR-001 // Hard startup error before a single frame is decoded: a gallery built // with another embedder yields plausible-looking, meaningless matches. verify_gallery_embedder(gallery, cfg.gallery_path, cfg.arcface_model, cfg.require_gallery_stamp); } catch (const std::exception& e) { std::cerr << "Gallery error: " << e.what() << "\n"; return 1; } std::cerr << "[main] gallery loaded: " << gallery.actors.size() << " actors\n"; // ── Construct node functors ─────────────────────────────────────────────── std::atomic done{false}; // set by result_sink (face branch) std::atomic scene_done{true}; // set by scene_detector; true when disabled FrameSourceFunc source_fn {cfg}; CameraPositionChangeDetectorFunc campos_fn {cfg}; FaceDetectorFunc detector_fn{cfg}; FaceAlignerFunc aligner_fn; EmbedderFunc embedder_fn{cfg}; // Constructed before the tracker: it fits (or loads) the calibration, and // the tracker must decide in that same probability space (AR-024). IdentityMatcherFunc matcher_fn {gallery, cfg}; /// TRACES: AR-007, AR-008, AR-012, AR-024 | SR-002 // The registry is created here and shared, not owned by a node: track state // is not a stage in the stream, it is state several stages read and write, // and its final answer is only known when a track dies. auto same_person = same_person_probability(matcher_fn.calibration()); TrackRegistry::Config reg_cfg; reg_cfg.extinction_sec = cfg.track_extinction_sec; auto registry = std::make_shared( reg_cfg, EvidenceDiscounter(same_person)); matcher_fn.set_registry(registry); FaceTrackerFunc ftracker_fn{cfg, registry, same_person}; SceneTrackerFunc tracker_fn {cfg}; ResultSinkFunc sink_fn {cfg, done}; /// TRACES: AR-012, AR-016, IR-002, IR-003 | SR-002 // A reaped track goes straight to the aggregator, so the registry holds only // live tracks and its size is bounded by concurrent on-screen faces rather // than growing with the film. registry->on_track_dead([&sink_fn](const DeadTrack& d) { sink_fn.add_claim(d); }); // AR-016: a film ends with faces on screen and those tracks have not timed // out. Without this flush the closing scene's cast is silently never // emitted — a loss that reads as a recognition miss, not a bookkeeping bug. sink_fn.set_pre_write_hook([registry](double last_ts) { registry->flush(last_ts); }); #ifdef SAE_DEBUG DebugRendererFunc debug_fn {cfg}; #endif // ── Wrap in KPN ObjectNodes ─────────────────────────────────────────────── // Queue sizes tuned to the pipeline's speed profile: // embedder (16ms) is the slowest GPU node — buffer before it must be largest // to prevent face_aligner pool overflows and frame drops. kpn::ObjectNode, kpn::out<"raw">, "frame_source", 0> source (source_fn, 32); kpn::ObjectNode, kpn::out<"frame">, "camera_pos", 0> campos (campos_fn, 32); kpn::ObjectNode, kpn::out<"scene">, "face_detector", 0> detector (detector_fn, 64); kpn::ObjectNode, kpn::out<"aligned">, "face_aligner", 0> aligner (aligner_fn, 64); kpn::ObjectNode, kpn::out<"embedded">, "embedder", 0> embedder (embedder_fn, 32); kpn::ObjectNode, kpn::out<"tracked">, "face_tracker", 0> ftracker (ftracker_fn, 16); kpn::ObjectNode, kpn::out<"matched">, "identity_matcher", 0> matcher (matcher_fn, 16); kpn::ObjectNode, kpn::out<"annotation">, "scene_tracker", 0> tracker (tracker_fn, 16); kpn::ObjectNode,kpn::out<>, "result_sink", 0> sink (sink_fn, 16); // ── Pipeline observability + run loop (topology-agnostic) ────────────────── // Factored into a lambda so the two topologies (with/without the scene-detect // branch) share identical event handling, wait loop, and teardown. Any // make_network result type binds to `Net&&`. std::mutex event_mtx; std::map overflow_counts; std::atomic node_crashed{false}; auto run_net = [&](auto&& net) -> int { // Tally per-node channel overflow, and treat any non-result_sink Closed // event as a crash so the wait loop below can't hang on `done` forever. net.set_event_handler( [&](std::string_view node_name, kpn::NodeEvent ev, std::chrono::steady_clock::time_point) { if (ev == kpn::NodeEvent::Overflow) { std::lock_guard lk(event_mtx); ++overflow_counts[std::string(node_name)]; } else { // NodeEvent::Closed if (node_name == "result_sink" && done.load(std::memory_order_acquire)) return; std::cerr << "[main] node '" << node_name << "' stopped unexpectedly — aborting pipeline\n"; node_crashed.store(true, std::memory_order_release); } }); std::cerr << "[main] starting pipeline…\n"; net.start(); // Wait until BOTH terminal branches finish: result_sink (face pipeline) // and, when enabled, scene_detector (the dense TransNetV2 branch, which // runs much slower and must not be torn down mid-stream). scene_done is // pre-set true when scene detection is disabled. while ((!done.load(std::memory_order_acquire) || !scene_done.load(std::memory_order_acquire)) && !node_crashed.load(std::memory_order_acquire)) std::this_thread::sleep_for(std::chrono::milliseconds(100)); net.stop(); net.print_diagnostics(); /// TRACES: AR-004 | SR-002 // A dropped frame does not degrade a result, it silently changes one — // the output is a claim about footage that was never analysed, and // nothing in the file says so. Since AR-004 made data pushes block, a // drop can no longer happen on the data path, so any drop here means // either that fix regressed (it lives in the KPN submodule, one line, // easy to lose in an update) or a channel was disabled mid-run. // // Reporting it in a footer and exiting 0 made both invisible: the run // "succeeded" and the truth file looked complete. Fail instead. /// TRACES: AR-010 | SR-002 if (scene_stats) { std::cerr << "[scene_annotate] boundaries=" << scene_stats->count() << " scored_through=" << scene_stats->scored_through() << "s"; // The tail is expected: frames after the detector's last full // window are never covered, and no amount of buffering changes // that. They are counted rather than silently treated as // boundary-free, which is the distinction that matters. if (scene_stats->outran() > 0) std::cerr << " unscored=" << scene_stats->outran() << " frame(s) past the detector's last window — treated as" " boundary-free, which is unverified rather than known"; std::cerr << "\n"; } bool dropped = false; { std::lock_guard lk(event_mtx); if (!overflow_counts.empty()) { dropped = true; std::cerr << "[main] ERROR: frames were dropped (channel overflow):\n"; for (const auto& [name, count] : overflow_counts) std::cerr << " " << name << ": " << count << "\n"; std::cerr << "[main] The output would describe footage that was never " "analysed. Refusing to report success.\n"; } } if (node_crashed.load(std::memory_order_acquire)) return 1; return dropped ? 2 : 0; }; // ── Build static network and run ────────────────────────────────────────── // Common face-analysis chain (campos → … → sink) is identical in all cases; // the scene-detect branch and the debug fanout are spliced on conditionally. // Topology: // plain: source → campos → detector → … → sink // scene-detect: source ─┬→ campos → decimate(filter) → detector → … → sink // └→ scene_detector (TransNetV2 sink → scenes.json) // The fanout after `source` is auto-inserted by make_network when its output // feeds two edges. In dense mode campos still sees native-rate frames (so it // detects angle changes correctly); a FilterNode then thins to sample_fps // before face detection. #ifdef SAE_DEBUG kpn::ObjectNode, kpn::out<>, "debug_renderer", 1> debug_node(debug_fn, 16); #define SAE_DEBUG_EDGE , kpn::edge(matcher.output<"matched">(), debug_node.input<"matched">()) #else #define SAE_DEBUG_EDGE #endif int rc = 0; if (!cfg.dump_embeddings_path.empty()) { // Dump-only topology: run the expensive front half and tee the embedder // output to an HDF5 dump for offline sweep replay (sae_kpn). Downstream // matching is skipped — the sweep re-runs it from the dump. EmbeddingDumpFunc dump_fn{cfg, done}; kpn::ObjectNode, kpn::out<>, "embedding_dump", 0> dump_node(dump_fn, 32); auto net = kpn::make_network( kpn::edge(source.output<"raw">(), campos.input<"raw">()), kpn::edge(campos.output<"frame">(), detector.input<"frame">()), kpn::edge(detector.output<"scene">(), aligner.input<"scene">()), kpn::edge(aligner.output<"aligned">(), embedder.input<"aligned">()), kpn::edge(embedder.output<"embedded">(), dump_node.input<"embedded">()) ); return run_net(std::move(net)); } if (cfg.scene_detect) { scene_done.store(false, std::memory_order_release); // now a real terminal branch SceneDetectorFunc scene_fn{cfg, scene_done}; /// TRACES: AR-010 | SR-002 // The join of the decode butterfly. source fans out to the dense // TransNetV2 branch and the sampled face branch; boundaries found on the // first have to reach the second, and cannot ride the frames because the // branches run in parallel. // // TransNetV2 buffers kWindow frames before it can score any of them, so // the face branch must lag by at least that much or it will ask about // frames nobody has looked at yet. Channel depth is what creates the lag: // with backpressure (AR-004) the fanout blocks on the slower branch, so // a deep face-branch channel lets the detector run ahead by its window // rather than dropping anything. auto boundaries = std::make_shared(); scene_fn.set_boundaries(boundaries); kpn::ObjectNode, kpn::out<>, "scene_detector", 0> scene_node(scene_fn, 128); // Decimator: keep frames on the sample_fps cadence, drop the rest. // eof always passes so downstream shuts down cleanly. Stateful — one // instance, mutable via shared_ptr so the std::function stays copyable. auto decim_state = std::make_shared(-1e18); const double interval = 1.0 / cfg.sample_fps; auto decimate = kpn::make_filter( [decim_state, interval](const Frame& f) { if (f.eof) return true; if (f.timestamp_sec - *decim_state >= interval - 1e-6) { *decim_state = f.timestamp_sec; return true; } return false; }, kSceneJoinDepth); /// TRACES: AR-010 | SR-002 // Stamp is_scene_boundary from the detector's published verdict. tol is // half a sample interval: the two branches sample at different rates, so // a boundary found on a dense frame rarely lands exactly on a sampled // one, and half an interval attributes it to the nearest sampled frame // and no further. // // outran() counts frames that arrived before the detector had scored // them. Nonzero means the join depth is too shallow for the window, and // those frames were annotated from an incomplete verdict — which would // otherwise look exactly like "no boundary here". SceneBoundaryAnnotatorFunc annotate_fn{boundaries, 0.5 / cfg.sample_fps}; kpn::ObjectNode, kpn::out<"frame">, "scene_annotate", 0> annotate(annotate_fn, kSceneJoinDepth); // Reported at shutdown: without this the join is unverifiable, and an // annotator that never fired looks identical to footage with no // boundaries. scene_stats = boundaries; auto net = kpn::make_network( kpn::edge(source.output<"raw">(), campos.input<"raw">()), kpn::edge(source.output<"raw">(), scene_node.input<"dense">()), kpn::edge(campos.output<"frame">(), decimate.input<0>()), kpn::edge(decimate.output<0>(), annotate.input<"frame">()), kpn::edge(annotate.output<"frame">(), detector.input<"frame">()), kpn::edge(detector.output<"scene">(), aligner.input<"scene">()), kpn::edge(aligner.output<"aligned">(), embedder.input<"aligned">()), kpn::edge(embedder.output<"embedded">(), ftracker.input<"embedded">()), kpn::edge(ftracker.output<"tracked">(), matcher.input<"tracked">()), kpn::edge(matcher.output<"matched">(), tracker.input<"matched">()), kpn::edge(tracker.output<"annotation">(), sink.input<"annotation">()) SAE_DEBUG_EDGE ); rc = run_net(std::move(net)); } else { auto net = kpn::make_network( kpn::edge(source.output<"raw">(), campos.input<"raw">()), kpn::edge(campos.output<"frame">(), detector.input<"frame">()), kpn::edge(detector.output<"scene">(), aligner.input<"scene">()), kpn::edge(aligner.output<"aligned">(), embedder.input<"aligned">()), kpn::edge(embedder.output<"embedded">(), ftracker.input<"embedded">()), kpn::edge(ftracker.output<"tracked">(), matcher.input<"tracked">()), kpn::edge(matcher.output<"matched">(), tracker.input<"matched">()), kpn::edge(tracker.output<"annotation">(), sink.input<"annotation">()) SAE_DEBUG_EDGE ); rc = run_net(std::move(net)); } #undef SAE_DEBUG_EDGE return rc; }