Spaces:

ruslanmv
/

HomePilot

Running

HomePilot / backend /app /interactive /playback /render_set.py

HomePilot Deploy Bot

chore(hf): sync HomePilot to HF Space

921d377 6 months ago

12.6 kB

"""Scene-render target selection for the Interactive pre-render pass.
Additive, best-practice replacement for the naive
``[n for n in nodes if n.kind in ("scene","decision","assessment","remediation")]``
filter that ``generator_auto.py`` applies before calling ``render_scene_async``.
Why
---
The naive filter ignores two facts:
1. **Persona Live Play owns its own rendering pipeline.** It anchors on the
   persona portrait and renders edit-recipe variants per user action. The
   scene-graph ``asset_ids[]`` are never read by the Persona Live runtime
   (see ``interactive/routes/persona_live.py::_pick_anchor_image_ref``).
   Pre-rendering scene nodes for a ``persona_live_play`` project is pure
   wasted compute — one SDXL pass per node that nothing ever displays.
2. **Standard projects rarely need every node rendered before Play opens.**
   Only the start node and its first-decision layer need to be ready for
   the initial frames. Deeper nodes can be rendered on-demand as the
   user's cursor approaches them (industry prefetch pattern).
This module exposes two pure functions with zero I/O side effects:
- ``compute_reachable_nodes(nodes, edges, start_id, max_depth)`` — BFS over
  the graph bounded by ``max_depth``.
- ``filter_targets_for_play(targets, experience, edges)`` — takes the
  already-computed naive target list and the experience, returns the
  subset that actually needs pre-rendering.
Both are side-effect-free and typed so they're trivial to unit-test
without standing up a real experience.
Environment knobs
-----------------
- ``INTERACTIVE_PRERENDER_DEPTH`` (int, default 2) — BFS depth limit for
  Standard projects. Higher = more eager, more compute. Lower = lazier,
  faster "Rendering scenes" phase.
- ``INTERACTIVE_PRERENDER_PERSONA_LIVE`` (``true|false``, default ``false``)
  — emergency knob that restores the old behavior (pre-render everything
  for Persona Live too) if a regression is found.
"""
from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Iterable, List, Optional, Sequence, Set

# ── Configuration ────────────────────────────────────────────────────────────

# BFS depth used to decide how many layers of scene/decision nodes to
# pre-render for Standard projects. Default is a large number so existing
# flows pre-render everything (behavior-compatible). Operators opt into
# the depth-limited eager render by setting INTERACTIVE_PRERENDER_DEPTH to
# a small value (e.g. 2 = start + first decision + first choice layer).
DEFAULT_PRERENDER_DEPTH: int = int(os.getenv("INTERACTIVE_PRERENDER_DEPTH", "99") or 99)

# Emergency knob: pre-render everything for Persona Live (old behavior).
PRERENDER_PERSONA_LIVE: bool = (
    os.getenv("INTERACTIVE_PRERENDER_PERSONA_LIVE", "false").lower() == "true"
)

# Kinds that carry their own visual asset. Endings reuse the trailing frame.
RENDERABLE_KINDS: frozenset[str] = frozenset(
    {"scene", "decision", "assessment", "remediation"}
)


# ── Types ────────────────────────────────────────────────────────────────────

@dataclass(frozen=True)
class _NodeLike:
    """Minimal duck-typed view of an interactive Node.
    Kept deliberately narrow so this module doesn't depend on the concrete
    ORM/dataclass in ``branching/graph.py`` — callers can pass any object
    with ``id``, ``kind``, and ``asset_ids`` attributes.
    """
    id: str
    kind: str
    asset_ids: Sequence[str] = ()


@dataclass(frozen=True)
class _EdgeLike:
    """Duck-typed edge view.
    Production ``interactive.models.Edge`` uses ``from_node_id`` /
    ``to_node_id``. Legacy test doubles (and the shim below) use
    ``from_id`` / ``to_id``. ``_edge_endpoints()`` below accepts both —
    this dataclass just documents the "old" shape for reference.
    """
    from_id: str
    to_id: str


def _edge_endpoints(edge: object) -> tuple[str, str]:
    """Return ``(from, to)`` for an edge, tolerating both naming styles.
    The production ``Edge`` pydantic model exposes ``from_node_id`` /
    ``to_node_id``; the test doubles in ``test_interactive_render_set``
    use ``from_id`` / ``to_id``. Returning ``("", "")`` for unrecognised
    shapes lets the caller skip the edge without crashing.
    """
    fid = (
        getattr(edge, "from_node_id", None)
        or getattr(edge, "from_id", None)
        or ""
    )
    tid = (
        getattr(edge, "to_node_id", None)
        or getattr(edge, "to_id", None)
        or ""
    )
    return str(fid or ""), str(tid or "")


@dataclass(frozen=True)
class RenderDecision:
    """Why a node was / wasn't selected for pre-render.
    Useful for SSE events and debug logs — lets the UI (and operators)
    see at a glance why "Rendering scenes 0/6" is actually 0/2 or 0/0.
    """
    node_id: str
    selected: bool
    reason: str
    depth: Optional[int] = None


# ── Core functions ───────────────────────────────────────────────────────────

def compute_reachable_nodes(
    nodes: Iterable,
    edges: Iterable,
    *,
    start_id: str,
    max_depth: int = DEFAULT_PRERENDER_DEPTH,
) -> Set[str]:
    """BFS from *start_id* up to (inclusive) *max_depth* edges away.
    Returns the set of reachable node IDs. The start node itself is
    always included at depth 0. Non-existent edges / dangling IDs are
    skipped silently — validation is the planner's job, not ours.
    A ``max_depth`` of 0 returns only the start node. Negative depths
    are clamped to 0.
    """
    if not start_id:
        return set()
    max_depth = max(0, int(max_depth))

    node_ids = {str(getattr(n, "id", "") or "") for n in nodes}
    node_ids.discard("")
    if start_id not in node_ids:
        return set()

    # Forward adjacency only — render set cares about what the user
    # reaches *from* the start, not what reaches them.
    adj: dict[str, List[str]] = {}
    for e in edges:
        fid, tid = _edge_endpoints(e)
        if fid and tid and fid in node_ids and tid in node_ids:
            adj.setdefault(fid, []).append(tid)

    reachable: Set[str] = {start_id}
    frontier: List[tuple[str, int]] = [(start_id, 0)]
    while frontier:
        nid, depth = frontier.pop(0)
        if depth >= max_depth:
            continue
        for child in adj.get(nid, ()):
            if child not in reachable:
                reachable.add(child)
                frontier.append((child, depth + 1))
    return reachable


def _interaction_type(experience: object) -> str:
    """Resolve the interaction flavour, tolerating dict / pydantic shapes.
    Production stores the setting in two places depending on surface:
      * ``experience.audience_profile["interaction_type"]`` — canonical,
        set by the wizard and read by Persona Live routes.
      * ``experience.project_type`` — legacy field; a value of
        ``"persona_live"`` is the project-level marker.
    We also accept a top-level ``interaction_type`` attribute/key so the
    existing unit tests (which pass ``{"interaction_type": ...}``
    literals) keep working. Returned value is normalised to match the
    legacy ``"persona_live_play"`` sentinel used by the filter.
    """
    def _dict_or_none(obj):
        if isinstance(obj, dict):
            return obj
        return None

    # 1. Top-level attribute / key (test-friendly, legacy)
    if isinstance(experience, dict):
        top = str(experience.get("interaction_type") or "").strip()
    else:
        top = str(getattr(experience, "interaction_type", "") or "").strip()
    if top:
        return top

    # 2. audience_profile.interaction_type (canonical runtime location)
    ap = None
    if isinstance(experience, dict):
        ap = _dict_or_none(experience.get("audience_profile"))
    else:
        ap = _dict_or_none(getattr(experience, "audience_profile", None))
    if ap:
        val = str(ap.get("interaction_type") or "").strip()
        if val:
            return val

    # 3. project_type fallback — "persona_live" implies persona-live play
    if isinstance(experience, dict):
        pt = str(experience.get("project_type") or "").strip().lower()
    else:
        pt = str(getattr(experience, "project_type", "") or "").strip().lower()
    if pt == "persona_live":
        return "persona_live_play"
    return ""


def _entry_node_id(experience: object, nodes: Sequence) -> str:
    """Best-effort lookup of the start node.
    Prefers an explicit ``entry_node_id`` / ``start_node_id`` field on
    the experience; falls back to the first scene node.
    """
    eid = ""
    if isinstance(experience, dict):
        eid = str(
            experience.get("entry_node_id")
            or experience.get("start_node_id")
            or ""
        )
    else:
        eid = str(
            getattr(experience, "entry_node_id", "")
            or getattr(experience, "start_node_id", "")
            or ""
        )
    if eid:
        return eid
    for n in nodes:
        if str(getattr(n, "kind", "") or "") == "scene":
            return str(getattr(n, "id", "") or "")
    return ""


def filter_targets_for_play(
    naive_targets: Sequence,
    *,
    experience: object,
    edges: Iterable,
    nodes: Optional[Sequence] = None,
    max_depth: int = DEFAULT_PRERENDER_DEPTH,
) -> tuple[List, List[RenderDecision]]:
    """Scope the naive target list to what Play will actually display.
    Returns ``(targets, decisions)`` where:
      * ``targets`` is the filtered list (same objects as ``naive_targets``),
        preserving their original order.
      * ``decisions`` is one ``RenderDecision`` per *original* target so
        callers can emit SSE events for both selected and skipped nodes.
    Policy
    ------
    1. ``interaction_type == "persona_live_play"`` → return ``[]``. Persona
       Live anchors on the persona portrait and renders edit-recipe variants
       on demand; scene-graph assets are never read by its runtime.
       Set ``INTERACTIVE_PRERENDER_PERSONA_LIVE=true`` to opt back in.
    2. Otherwise → BFS reachable set from the entry node, bounded by
       ``max_depth``. Nodes outside the set are marked ``deferred``; the
       on-demand render pass can fill them in when the user cursor lands.
    3. Nodes that already carry ``asset_ids`` are always kept — idempotency
       matters more than scope. (The existing generator_auto.py also
       skips these during the render loop.)
    """
    itype = _interaction_type(experience)

    # Persona Live: hand back zero targets.
    if itype == "persona_live_play" and not PRERENDER_PERSONA_LIVE:
        decisions = [
            RenderDecision(
                node_id=str(getattr(n, "id", "") or ""),
                selected=False,
                reason="persona_live_own_pipeline",
            )
            for n in naive_targets
        ]
        return [], decisions

    # Standard: BFS-bounded reachable set.
    node_list: Sequence = list(nodes) if nodes is not None else list(naive_targets)
    start_id = _entry_node_id(experience, node_list)
    reachable: Set[str] = set()
    if start_id:
        reachable = compute_reachable_nodes(
            node_list, edges, start_id=start_id, max_depth=max_depth
        )

    selected: List = []
    decisions: List[RenderDecision] = []
    for n in naive_targets:
        nid = str(getattr(n, "id", "") or "")
        has_assets = bool(getattr(n, "asset_ids", None))
        if has_assets:
            selected.append(n)
            decisions.append(RenderDecision(nid, True, "already_rendered"))
            continue
        if not reachable:
            # No entry node resolved — fall back to the original behavior.
            selected.append(n)
            decisions.append(RenderDecision(nid, True, "reachable_unknown"))
            continue
        if nid in reachable:
            selected.append(n)
            decisions.append(RenderDecision(nid, True, "reachable"))
        else:
            decisions.append(
                RenderDecision(nid, False, "deferred_unreachable_at_depth")
            )
    return selected, decisions


__all__ = [
    "DEFAULT_PRERENDER_DEPTH",
    "PRERENDER_PERSONA_LIVE",
    "RENDERABLE_KINDS",
    "RenderDecision",
    "compute_reachable_nodes",
    "filter_targets_for_play",
]

	"""Scene-render target selection for the Interactive pre-render pass.

	Additive, best-practice replacement for the naive
	``[n for n in nodes if n.kind in ("scene","decision","assessment","remediation")]``
	filter that ``generator_auto.py`` applies before calling ``render_scene_async``.

	Why
	---
	The naive filter ignores two facts:

	1. Persona Live Play owns its own rendering pipeline. It anchors on the
	persona portrait and renders edit-recipe variants per user action. The
	scene-graph ``asset_ids[]`` are never read by the Persona Live runtime
	(see ``interactive/routes/persona_live.py::_pick_anchor_image_ref``).
	Pre-rendering scene nodes for a ``persona_live_play`` project is pure
	wasted compute — one SDXL pass per node that nothing ever displays.

	2. Standard projects rarely need every node rendered before Play opens.
	Only the start node and its first-decision layer need to be ready for
	the initial frames. Deeper nodes can be rendered on-demand as the
	user's cursor approaches them (industry prefetch pattern).

	This module exposes two pure functions with zero I/O side effects:

	- ``compute_reachable_nodes(nodes, edges, start_id, max_depth)`` — BFS over
	the graph bounded by ``max_depth``.
	- ``filter_targets_for_play(targets, experience, edges)`` — takes the
	already-computed naive target list and the experience, returns the
	subset that actually needs pre-rendering.

	Both are side-effect-free and typed so they're trivial to unit-test
	without standing up a real experience.

	Environment knobs
	-----------------
	- ``INTERACTIVE_PRERENDER_DEPTH`` (int, default 2) — BFS depth limit for
	Standard projects. Higher = more eager, more compute. Lower = lazier,
	faster "Rendering scenes" phase.
	- ``INTERACTIVE_PRERENDER_PERSONA_LIVE`` (``true\|false``, default ``false``)
	— emergency knob that restores the old behavior (pre-render everything
	for Persona Live too) if a regression is found.
	"""
	from __future__ import annotations

	import os
	from dataclasses import dataclass
	from typing import Iterable, List, Optional, Sequence, Set

	# ── Configuration ────────────────────────────────────────────────────────────

	# BFS depth used to decide how many layers of scene/decision nodes to
	# pre-render for Standard projects. Default is a large number so existing
	# flows pre-render everything (behavior-compatible). Operators opt into
	# the depth-limited eager render by setting INTERACTIVE_PRERENDER_DEPTH to
	# a small value (e.g. 2 = start + first decision + first choice layer).
	DEFAULT_PRERENDER_DEPTH: int = int(os.getenv("INTERACTIVE_PRERENDER_DEPTH", "99") or 99)

	# Emergency knob: pre-render everything for Persona Live (old behavior).
	PRERENDER_PERSONA_LIVE: bool = (
	os.getenv("INTERACTIVE_PRERENDER_PERSONA_LIVE", "false").lower() == "true"
	)

	# Kinds that carry their own visual asset. Endings reuse the trailing frame.
	RENDERABLE_KINDS: frozenset[str] = frozenset(
	{"scene", "decision", "assessment", "remediation"}
	)


	# ── Types ────────────────────────────────────────────────────────────────────

	@dataclass(frozen=True)
	class _NodeLike:
	"""Minimal duck-typed view of an interactive Node.

	Kept deliberately narrow so this module doesn't depend on the concrete
	ORM/dataclass in ``branching/graph.py`` — callers can pass any object
	with ``id``, ``kind``, and ``asset_ids`` attributes.
	"""
	id: str
	kind: str
	asset_ids: Sequence[str] = ()


	@dataclass(frozen=True)
	class _EdgeLike:
	"""Duck-typed edge view.

	Production ``interactive.models.Edge`` uses ``from_node_id`` /
	``to_node_id``. Legacy test doubles (and the shim below) use
	``from_id`` / ``to_id``. ``_edge_endpoints()`` below accepts both —
	this dataclass just documents the "old" shape for reference.
	"""
	from_id: str
	to_id: str


	def _edge_endpoints(edge: object) -> tuple[str, str]:
	"""Return ``(from, to)`` for an edge, tolerating both naming styles.

	The production ``Edge`` pydantic model exposes ``from_node_id`` /
	``to_node_id``; the test doubles in ``test_interactive_render_set``
	use ``from_id`` / ``to_id``. Returning ``("", "")`` for unrecognised
	shapes lets the caller skip the edge without crashing.
	"""
	fid = (
	getattr(edge, "from_node_id", None)
	or getattr(edge, "from_id", None)
	or ""
	)
	tid = (
	getattr(edge, "to_node_id", None)
	or getattr(edge, "to_id", None)
	or ""
	)
	return str(fid or ""), str(tid or "")


	@dataclass(frozen=True)
	class RenderDecision:
	"""Why a node was / wasn't selected for pre-render.

	Useful for SSE events and debug logs — lets the UI (and operators)
	see at a glance why "Rendering scenes 0/6" is actually 0/2 or 0/0.
	"""
	node_id: str
	selected: bool
	reason: str
	depth: Optional[int] = None


	# ── Core functions ───────────────────────────────────────────────────────────

	def compute_reachable_nodes(
	nodes: Iterable,
	edges: Iterable,
	*,
	start_id: str,
	max_depth: int = DEFAULT_PRERENDER_DEPTH,
	) -> Set[str]:
	"""BFS from start_id up to (inclusive) max_depth edges away.

	Returns the set of reachable node IDs. The start node itself is
	always included at depth 0. Non-existent edges / dangling IDs are
	skipped silently — validation is the planner's job, not ours.

	A ``max_depth`` of 0 returns only the start node. Negative depths
	are clamped to 0.
	"""
	if not start_id:
	return set()
	max_depth = max(0, int(max_depth))

	node_ids = {str(getattr(n, "id", "") or "") for n in nodes}
	node_ids.discard("")
	if start_id not in node_ids:
	return set()

	# Forward adjacency only — render set cares about what the user
	# reaches from the start, not what reaches them.
	adj: dict[str, List[str]] = {}
	for e in edges:
	fid, tid = _edge_endpoints(e)
	if fid and tid and fid in node_ids and tid in node_ids:
	adj.setdefault(fid, []).append(tid)

	reachable: Set[str] = {start_id}
	frontier: List[tuple[str, int]] = [(start_id, 0)]
	while frontier:
	nid, depth = frontier.pop(0)
	if depth >= max_depth:
	continue
	for child in adj.get(nid, ()):
	if child not in reachable:
	reachable.add(child)
	frontier.append((child, depth + 1))
	return reachable


	def _interaction_type(experience: object) -> str:
	"""Resolve the interaction flavour, tolerating dict / pydantic shapes.

	Production stores the setting in two places depending on surface:
	* ``experience.audience_profile["interaction_type"]`` — canonical,
	set by the wizard and read by Persona Live routes.
	* ``experience.project_type`` — legacy field; a value of
	``"persona_live"`` is the project-level marker.

	We also accept a top-level ``interaction_type`` attribute/key so the
	existing unit tests (which pass ``{"interaction_type": ...}``
	literals) keep working. Returned value is normalised to match the
	legacy ``"persona_live_play"`` sentinel used by the filter.
	"""
	def _dict_or_none(obj):
	if isinstance(obj, dict):
	return obj
	return None

	# 1. Top-level attribute / key (test-friendly, legacy)
	if isinstance(experience, dict):
	top = str(experience.get("interaction_type") or "").strip()
	else:
	top = str(getattr(experience, "interaction_type", "") or "").strip()
	if top:
	return top

	# 2. audience_profile.interaction_type (canonical runtime location)
	ap = None
	if isinstance(experience, dict):
	ap = _dict_or_none(experience.get("audience_profile"))
	else:
	ap = _dict_or_none(getattr(experience, "audience_profile", None))
	if ap:
	val = str(ap.get("interaction_type") or "").strip()
	if val:
	return val

	# 3. project_type fallback — "persona_live" implies persona-live play
	if isinstance(experience, dict):
	pt = str(experience.get("project_type") or "").strip().lower()
	else:
	pt = str(getattr(experience, "project_type", "") or "").strip().lower()
	if pt == "persona_live":
	return "persona_live_play"
	return ""


	def _entry_node_id(experience: object, nodes: Sequence) -> str:
	"""Best-effort lookup of the start node.

	Prefers an explicit ``entry_node_id`` / ``start_node_id`` field on
	the experience; falls back to the first scene node.
	"""
	eid = ""
	if isinstance(experience, dict):
	eid = str(
	experience.get("entry_node_id")
	or experience.get("start_node_id")
	or ""
	)
	else:
	eid = str(
	getattr(experience, "entry_node_id", "")
	or getattr(experience, "start_node_id", "")
	or ""
	)
	if eid:
	return eid
	for n in nodes:
	if str(getattr(n, "kind", "") or "") == "scene":
	return str(getattr(n, "id", "") or "")
	return ""


	def filter_targets_for_play(
	naive_targets: Sequence,
	*,
	experience: object,
	edges: Iterable,
	nodes: Optional[Sequence] = None,
	max_depth: int = DEFAULT_PRERENDER_DEPTH,
	) -> tuple[List, List[RenderDecision]]:
	"""Scope the naive target list to what Play will actually display.

	Returns ``(targets, decisions)`` where:
	* ``targets`` is the filtered list (same objects as ``naive_targets``),
	preserving their original order.
	* ``decisions`` is one ``RenderDecision`` per original target so
	callers can emit SSE events for both selected and skipped nodes.

	Policy
	------
	1. ``interaction_type == "persona_live_play"`` → return ``[]``. Persona
	Live anchors on the persona portrait and renders edit-recipe variants
	on demand; scene-graph assets are never read by its runtime.
	Set ``INTERACTIVE_PRERENDER_PERSONA_LIVE=true`` to opt back in.
	2. Otherwise → BFS reachable set from the entry node, bounded by
	``max_depth``. Nodes outside the set are marked ``deferred``; the
	on-demand render pass can fill them in when the user cursor lands.
	3. Nodes that already carry ``asset_ids`` are always kept — idempotency
	matters more than scope. (The existing generator_auto.py also
	skips these during the render loop.)
	"""
	itype = _interaction_type(experience)

	# Persona Live: hand back zero targets.
	if itype == "persona_live_play" and not PRERENDER_PERSONA_LIVE:
	decisions = [
	RenderDecision(
	node_id=str(getattr(n, "id", "") or ""),
	selected=False,
	reason="persona_live_own_pipeline",
	)
	for n in naive_targets
	]
	return [], decisions

	# Standard: BFS-bounded reachable set.
	node_list: Sequence = list(nodes) if nodes is not None else list(naive_targets)
	start_id = _entry_node_id(experience, node_list)
	reachable: Set[str] = set()
	if start_id:
	reachable = compute_reachable_nodes(
	node_list, edges, start_id=start_id, max_depth=max_depth
	)

	selected: List = []
	decisions: List[RenderDecision] = []
	for n in naive_targets:
	nid = str(getattr(n, "id", "") or "")
	has_assets = bool(getattr(n, "asset_ids", None))
	if has_assets:
	selected.append(n)
	decisions.append(RenderDecision(nid, True, "already_rendered"))
	continue
	if not reachable:
	# No entry node resolved — fall back to the original behavior.
	selected.append(n)
	decisions.append(RenderDecision(nid, True, "reachable_unknown"))
	continue
	if nid in reachable:
	selected.append(n)
	decisions.append(RenderDecision(nid, True, "reachable"))
	else:
	decisions.append(
	RenderDecision(nid, False, "deferred_unreachable_at_depth")
	)
	return selected, decisions


	__all__ = [
	"DEFAULT_PRERENDER_DEPTH",
	"PRERENDER_PERSONA_LIVE",
	"RENDERABLE_KINDS",
	"RenderDecision",
	"compute_reachable_nodes",
	"filter_targets_for_play",
	]