kami
ff6a512630
Working tree (including .git) was lost to an rm. Rebuilt by replaying Write/Edit/ Read/attachment events from 25 Claude sessions and 22 successful codex apply_patch blocks into one timestamp-ordered timeline. Verified against ground truth recorded in the transcripts: wc -l on 10 files and ls -l on 5 files at 2026-07-18T13:13:44Z both match exactly; 18 files are byte-identical to their newest ~/.claude/file-history blob. See HANDOFF.md for sources, gaps, and how to rebuild .venv. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
171 lines
8.6 KiB
Python
171 lines
8.6 KiB
Python
# worker_scene.py — stage 6 scene graph. FastAPI :8004. no GPU, no session (cheap join).
|
|
# joins vision + identity into a named-character scene graph. speaker attribution comes straight from
|
|
# the vision/dialogue model (tail direction + turn-taking); the old OCR nearest-bbox heuristic is gone.
|
|
import json, logging
|
|
from fastapi import FastAPI
|
|
from pydantic import BaseModel
|
|
import transport
|
|
|
|
log = logging.getLogger("scene")
|
|
|
|
|
|
def _describe(appearance) -> str:
|
|
"""A: a stable descriptive phrase for an unnamed character, from its FIRST-seen appearance
|
|
(frozen in the registry). "the one with black hair and glasses" — reused across panels so the
|
|
label never drifts; empty string if appearance is bare (script then falls back to Person X)."""
|
|
if isinstance(appearance, str):
|
|
try:
|
|
appearance = json.loads(appearance or "{}")
|
|
except ValueError:
|
|
appearance = {}
|
|
if not isinstance(appearance, dict):
|
|
return ""
|
|
# animals/creatures get named as their species, not by clothes ("the black cat", not "the one
|
|
# with a gold bell"). spec A: story-state tracks "props/entities seen (the cat)".
|
|
species = (appearance.get("species") or "human").strip().lower()
|
|
if species not in ("", "human", "person", "man", "woman"):
|
|
color = (appearance.get("hair") or "").strip()
|
|
feats = [str(f).strip() for f in (appearance.get("features") or []) if str(f).strip()]
|
|
color = color or (feats[0] if feats else "")
|
|
return f"the {color} {species}".replace(" ", " ").strip()
|
|
hair = (appearance.get("hair") or "").strip()
|
|
clothing = (appearance.get("clothing") or "").strip()
|
|
feats = [str(f).strip() for f in (appearance.get("features") or []) if str(f).strip()]
|
|
bits = []
|
|
if hair:
|
|
# vision sometimes already includes the word "hair" ("brown hair") -> don't double it.
|
|
bits.append(hair if hair.lower().endswith("hair") else f"{hair} hair")
|
|
if feats:
|
|
bits.append(feats[0])
|
|
elif clothing:
|
|
# vision may still return a compound garment ("yellow shirt and yellow jacket");
|
|
# keep only the first clause so the label stays "the one with a yellow shirt".
|
|
bits.append(clothing.split(" and ")[0].strip())
|
|
return "the one with " + " and ".join(bits) if bits else ""
|
|
|
|
|
|
class SceneInput(BaseModel):
|
|
panel_id: str = ""
|
|
panel_uri: str = ""
|
|
vision_result: dict = {}
|
|
identity_assignments: list = []
|
|
characters_registry: list = [] # [{"character_id","name"}]
|
|
|
|
|
|
def build_scene(data: SceneInput):
|
|
name_by_id = {c["character_id"]: c.get("name", "") for c in data.characters_registry}
|
|
desc_by_id = {c["character_id"]: c.get("description") for c in data.characters_registry}
|
|
id_by_local = {a["local_id"]: a["character_id"] for a in data.identity_assignments}
|
|
|
|
characters, present = [], []
|
|
for vc in data.vision_result.get("characters", []):
|
|
cid = id_by_local.get(vc.get("local_id"))
|
|
if not cid:
|
|
continue
|
|
name = name_by_id.get(cid, "")
|
|
# frozen registry appearance first (stable across panels); fall back to this panel's vision.
|
|
label = name or _describe(desc_by_id.get(cid)) or _describe(vc.get("appearance"))
|
|
characters.append({"id": cid, "name": name, "label": label})
|
|
present.append({"character_id": cid, "name": name, "bbox": vc.get("bbox", [0, 0, 0, 0])})
|
|
|
|
# Speaker attribution is the vision/dialogue model's job now (it reads bubble tails + turn-taking).
|
|
# speaker is a local_id -> map to character_id via identity; None for narration/sfx/off-panel.
|
|
dialogue = []
|
|
for d in data.vision_result.get("dialogue", []):
|
|
raw = d.get("speaker")
|
|
# "unknown" is a DELIBERATE off-panel/indeterminate speaker (narrated as "someone"); it maps
|
|
# to None just like narration. id_by_local.get already yields None for it.
|
|
dialogue.append({"speaker": id_by_local.get(raw), "text": d.get("text", ""),
|
|
"type": d.get("type", "speech"),
|
|
"confidence": d.get("confidence"),
|
|
"speaker_method": d.get("speaker_method", "unknown")})
|
|
|
|
vchars = data.vision_result.get("characters", [])
|
|
action = "; ".join(c.get("action", "") for c in vchars if c.get("action")) or ""
|
|
return {"panel_id": data.panel_id, "characters": characters, "dialogue": dialogue,
|
|
"action": action, "entities": data.vision_result.get("entities", []),
|
|
"camera": data.vision_result.get("camera", {}), # #8 direction passthrough -> render
|
|
"transition": data.vision_result.get("transition", "cut")} # #6 transition out
|
|
|
|
|
|
app = FastAPI()
|
|
transport.install_logging(app, "scene")
|
|
|
|
|
|
@app.post("/scene/build")
|
|
async def scene(data: SceneInput):
|
|
return build_scene(data)
|
|
|
|
|
|
@app.get("/health")
|
|
async def health():
|
|
return {"status": "ok"}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
out = build_scene(SceneInput(
|
|
panel_id="p001",
|
|
vision_result={"characters": [{"local_id": "person_1", "bbox": [0, 0, 10, 10], "action": "waving"}]},
|
|
identity_assignments=[{"local_id": "person_1", "character_id": "c1"}],
|
|
characters_registry=[{"character_id": "c1", "name": "Teto"}],
|
|
))
|
|
assert out["characters"] == [{"id": "c1", "name": "Teto", "label": "Teto"}]
|
|
|
|
# A: unnamed character -> stable descriptive label from frozen registry appearance
|
|
outu = build_scene(SceneInput(
|
|
panel_id="p003",
|
|
vision_result={"characters": [{"local_id": "person_1", "bbox": [0, 0, 10, 10]}]},
|
|
identity_assignments=[{"local_id": "person_1", "character_id": "c9"}],
|
|
characters_registry=[{"character_id": "c9", "name": "",
|
|
"description": '{"hair":"black","features":["glasses"]}'}],
|
|
))
|
|
assert outu["characters"][0]["name"] == "" and outu["characters"][0]["label"] == "the one with black hair and glasses"
|
|
assert _describe({}) == "" # bare appearance -> empty, script falls back to Person X
|
|
# animals are named as their species (color + species), never by clothing/"the one with..."
|
|
assert _describe({"species": "cat", "hair": "black"}) == "the black cat"
|
|
assert _describe({"species": "cat", "features": ["gold bell"]}) == "the gold bell cat"
|
|
assert _describe({"species": "cat"}) == "the cat"
|
|
assert _describe({"species": "human", "hair": "black"}) == "the one with black hair"
|
|
# compound garment collapses to the first clause (no "shirt and jacket" redundancy)
|
|
assert _describe({"clothing": "yellow shirt and yellow jacket"}) == "the one with yellow shirt"
|
|
assert out["action"] == "waving"
|
|
|
|
# vision-owned dialogue: typed, proper-cased, speaker mapped local_id -> character_id
|
|
out2 = build_scene(SceneInput(
|
|
panel_id="p002",
|
|
vision_result={
|
|
"characters": [{"local_id": "person_1", "bbox": [0, 0, 10, 10]}],
|
|
"dialogue": [{"speaker": "person_1", "type": "thought", "text": "So this is it."},
|
|
{"speaker": "", "type": "narration", "text": "Three years later."}],
|
|
"entities": [{"name": "Everyday", "kind": "shop"}],
|
|
},
|
|
identity_assignments=[{"local_id": "person_1", "character_id": "c1"}],
|
|
characters_registry=[{"character_id": "c1", "name": "Teto"}],
|
|
))
|
|
assert out2["dialogue"][0]["speaker"] == "c1" and out2["dialogue"][0]["type"] == "thought"
|
|
assert out2["dialogue"][1]["speaker"] is None and out2["dialogue"][1]["type"] == "narration"
|
|
assert out2["entities"][0]["name"] == "Everyday"
|
|
|
|
# explicit "unknown" (off-panel/indeterminate) maps to None -> narrated as "someone", never
|
|
# pinned to a visible character. (the "line defaults to the MC" bug.)
|
|
out4 = build_scene(SceneInput(
|
|
panel_id="p005b",
|
|
vision_result={
|
|
"characters": [{"local_id": "person_1", "bbox": [0, 0, 10, 10]}],
|
|
"dialogue": [{"speaker": "unknown", "type": "speech", "text": "A Teto and Egen test?"}],
|
|
},
|
|
identity_assignments=[{"local_id": "person_1", "character_id": "c1"}],
|
|
characters_registry=[{"character_id": "c1", "name": "MC"}],
|
|
))
|
|
assert out4["dialogue"][0]["speaker"] is None, out4["dialogue"][0]
|
|
# Attribution provenance is load-bearing: a weak guess must remain distinguishable downstream.
|
|
out5 = build_scene(SceneInput(
|
|
panel_id="p006",
|
|
vision_result={"dialogue": [{"speaker": "person_1", "type": "speech", "text": "Maybe.",
|
|
"confidence": 0.3, "speaker_method": "turn_taking"}]},
|
|
identity_assignments=[{"local_id": "person_1", "character_id": "c1"}],
|
|
))
|
|
assert out5["dialogue"][0]["confidence"] == 0.3
|
|
assert out5["dialogue"][0]["speaker_method"] == "turn_taking"
|
|
print("worker_scene self-check ok")
|