#pragma once /// TRACES: IR-001 | SR-003 #include "types.hpp" #include "config.hpp" #include "track_registry.hpp" #include #include #include #include #include #include #include #include #include #include #include using json = nlohmann::json; // ── ResultSinkFunc ──────────────────────────────────────────────────────────── // KPN sink node: accumulates SceneAnnotations and writes the final JSON on EOF. // // Verbosity::minimal — merges per-frame presence into contiguous time windows. // Output: { // "schema_version": 1, "movie": "...", "sample_fps": ..., "anneal_sec": ..., // "actors": [{ "name", "imdb_id", "tmdb_id", "jellyfin_id", "scenes": [[t0,t1], ...] }] // } // An optional top-level "jellyfin_item_id" (the analysed title's Jellyfin item // GUID) may also be present: scene_analyze doesn't know it, so it's stamped in // by run_from_jellyfin.py after analysis. Downstream tools (cameo detection) // use it to check cast membership in Jellyfin's own id space — see // scripts/cameo_jellyfin.py. // This is the spec consumed by the Jellyfin plugin: each actor carries every // identity key the gallery knows (empty string if not resolved). The plugin // should prefer "jellyfin_id" (direct Person item GUID) when non-empty, and // otherwise resolve "imdb_id"/"tmdb_id" against the item's People ProviderIds. // To find who's on screen at timestamp t, scan each actor's "scenes" for a // window where start <= t <= end. // // Verbosity::standard — per-frame detail including bboxes, similarity, unknowns. // Output: { "frames": [{ "t", "identified": [...], "unknowns": [...] }] } // // eof signal: sets done_ = true so the main thread can call net.stop(). struct ResultSinkFunc { static constexpr std::string_view label() { return "result_sink"; } /// TRACES: AR-012, AR-017, IR-002 | SR-002, SR-003 /// A finished presence claim from the registry. Called from inside the /// registry's reap while it holds its own lock, so this must stay a cheap /// push and must never re-enter the registry. void add_claim(const DeadTrack& d) { if (d.actor_idx < 0) return; // never owned: nothing to claim std::lock_guard g(claims_mu_); claims_.push_back(d); } ResultSinkFunc(const Config& cfg, std::atomic& done) : cfg_(cfg), done_(done) {} /// TRACES: AR-016 | SR-002 /// Runs immediately before the output is written, with the last timestamp /// seen. Used to flush tracks still live at EOF, which have not timed out /// and would otherwise never be emitted. void set_pre_write_hook(std::function fn) { pre_write_ = std::move(fn); } void operator()(SceneAnnotation sa) { if (sa.eof) { flush(); return; } // Progress to stderr std::cerr << "\r[result_sink] t=" << sa.timestamp_sec << "s" << " active=" << count_known(sa.visible_actors) << " unknowns=" << count_unknown(sa.visible_actors) << std::flush; for (const auto& ia : sa.visible_actors) { if (ia.actor_idx < 0) continue; auto& m = actor_meta_[ia.actor_idx]; if (m.name.empty()) m = {ia.name, ia.imdb_id, ia.tmdb_id, ia.jellyfin_id}; } last_ts_ = sa.timestamp_sec; frames_.push_back(std::move(sa)); } // Write accumulated results and signal done. Safe to call more than once. void flush() { if (written_.exchange(true)) return; if (pre_write_) pre_write_(last_ts_); write_output(); done_.store(true, std::memory_order_release); } private: // Bump when the minimal/standard output JSON structure changes in a way // the Jellyfin plugin needs to detect. static constexpr int kSchemaVersion = 2; // SR-003 coordinated bump static int count_known(const std::vector& v) { int n = 0; for (const auto& a : v) if (a.actor_idx >= 0) ++n; return n; } static int count_unknown(const std::vector& v) { int n = 0; for (const auto& a : v) if (a.actor_idx < 0) ++n; return n; } void write_output() { std::cerr << "\n[result_sink] writing " << cfg_.output_path << "\n"; json root; if (cfg_.verbosity == Verbosity::xray) { root = build_xray(); } else { /// TRACES: IR-002 | SR-003 /// schema_version 2, per jRay/SPEC.md JR-002. anneal_sec is REMOVED /// rather than zeroed: a field naming a mechanism the pipeline no /// longer has is actively misleading, and would outlive everyone who /// remembers why it reads 0. extinction_sec succeeds it as the /// parameter that actually shapes window extent. root["schema_version"] = kSchemaVersion; root["movie"] = cfg_.movie_path; root["extraction"] = { {"sample_fps", cfg_.sample_fps}, {"extinction_sec", cfg_.track_extinction_sec}, {"gallery_scope", cfg_.gallery_scope}, }; root["actors"] = build_epochs(); if (cfg_.verbosity == Verbosity::standard) root["frames"] = build_standard(); } std::ofstream f(cfg_.output_path); if (!f.is_open()) { std::cerr << "[result_sink] ERROR: cannot write " << cfg_.output_path << "\n"; return; } f << root.dump(2) << "\n"; std::cerr << "[result_sink] done.\n"; } struct Window { double start{0.0}; double end{0.0}; float belief{0.f}; ///< the posterior that justified the claim (AR-017) }; struct ActorWindow { std::string name, imdb_id, tmdb_id, jellyfin_id; std::vector scenes; }; struct ActorMeta { std::string name, imdb_id, tmdb_id, jellyfin_id; }; // Core logic: merge per-frame detections into annealed [start, end] windows. /// TRACES: AR-012, IR-002 | SR-002 /// A claim already IS a window — `[first_seen, last_seen]` of a track the /// actor owned. There is no annealing pass: `anneal_sec` existed to bridge /// gaps between isolated accepted frames, and a track that survives its own /// gaps leaves it nothing to do (see the AR-012 withdrawal note). std::vector build_actor_windows() { std::lock_guard g(claims_mu_); std::map by_actor; for (const auto& c : claims_) { auto& aw = by_actor[c.actor_idx]; if (aw.name.empty()) { auto it = actor_meta_.find(c.actor_idx); if (it != actor_meta_.end()) { aw.name = it->second.name; aw.imdb_id = it->second.imdb_id; aw.tmdb_id = it->second.tmdb_id; aw.jellyfin_id = it->second.jellyfin_id; } } aw.scenes.push_back({c.first_seen, c.last_seen, c.belief}); } std::vector result; for (auto& [idx, aw] : by_actor) { std::sort(aw.scenes.begin(), aw.scenes.end(), [](const Window& a, const Window& b) { return a.start < b.start; }); result.push_back(std::move(aw)); } return result; } json build_epochs() { json actors = json::array(); for (const auto& aw : build_actor_windows()) { // Objects, not float pairs: a window carries the belief that // justified it and the route by which it was identified (AR-017), // so a consumer can caveat or filter rather than treating every // window as equally certain. json windows = json::array(); for (const auto& w : aw.scenes) windows.push_back({{"start", w.start}, {"end", w.end}, {"belief", w.belief}, {"route", "live"}}); json ja; ja["name"] = aw.name; ja["imdb_id"] = aw.imdb_id; ja["tmdb_id"] = aw.tmdb_id; ja["jellyfin_id"] = aw.jellyfin_id; ja["scenes"] = std::move(windows); actors.push_back(std::move(ja)); } return actors; } // Jellyfin-Xray format: { "second": ["Actor", ...] } // Expands each annealed window into every integer second so coverage is dense // regardless of sample rate. Seconds between scenes have no key → overlay clears. json build_xray() { std::map> xray; for (const auto& aw : build_actor_windows()) { for (const auto& w : aw.scenes) { int t0 = static_cast(std::floor(w.start)); int t1 = static_cast(std::ceil(w.end)); for (int t = t0; t <= t1; ++t) xray[t].push_back(aw.name); } } json root = json::object(); for (const auto& [t, names] : xray) root[std::to_string(t)] = names; return root; } json build_standard() { json frames = json::array(); for (const auto& frame : frames_) { json jf; jf["t"] = frame.timestamp_sec; jf["identified"] = json::array(); jf["unknowns"] = json::array(); for (const auto& ia : frame.visible_actors) { const auto& b = ia.bbox; json jbox = {b.x, b.y, b.width, b.height}; if (ia.actor_idx >= 0) { json ja; ja["name"] = ia.name; ja["imdb_id"] = ia.imdb_id; ja["tmdb_id"] = ia.tmdb_id; ja["jellyfin_id"] = ia.jellyfin_id; ja["similarity"] = ia.similarity; ja["track_id"] = ia.track_id; ja["bbox"] = jbox; jf["identified"].push_back(std::move(ja)); } else { json ju; ju["bbox"] = jbox; ju["track_id"] = ia.track_id; ju["confidence"] = ia.similarity; // reuse field; 0 for unknowns jf["unknowns"].push_back(std::move(ju)); } } frames.push_back(std::move(jf)); } return frames; } const Config& cfg_; std::atomic& done_; std::atomic written_{false}; std::vector frames_; std::function pre_write_; double last_ts_{0.0}; std::mutex claims_mu_; std::vector claims_; std::map actor_meta_; ///< actor_idx → identity keys };