script-build-and-framework-tmp/visualization/backend/app/runtime_event_projection.py

from __future__ import annotations

from datetime import datetime, timezone
from typing import Any, Iterable

from .evaluation_event_projection import EvaluationEventProjector
from .creative_event_projection import CreativeEventProjector
from .main_agent_decision_projection import MainAgentDecisionProjector
from .retrieval_event_projection import RetrievalEventProjector
from .runtime_event_index import RuntimeEventIndex


class RuntimeEventProjector:
    """Facade that builds one event index and composes business projections."""

    def __init__(
        self,
        *,
        retrieval_projector: RetrievalEventProjector | None = None,
        evaluation_projector: EvaluationEventProjector | None = None,
        main_agent_projector: MainAgentDecisionProjector | None = None,
        creative_projector: CreativeEventProjector | None = None,
    ):
        self._retrieval_projector = retrieval_projector or RetrievalEventProjector()
        self._evaluation_projector = evaluation_projector or EvaluationEventProjector()
        self._main_agent_projector = main_agent_projector or MainAgentDecisionProjector()
        self._creative_projector = creative_projector or CreativeEventProjector()

    def project(
        self,
        events: list[dict[str, Any]],
        *,
        valid_rounds: set[int],
        valid_branches: set[int],
        captured_at: datetime | None = None,
        run_status: str | None = None,
        script_direction: str | None = None,
        rounds: Iterable[dict[str, Any]] = (),
        multipath_decisions: Iterable[dict[str, Any]] = (),
    ) -> dict[str, Any]:
        captured_at = captured_at or datetime.now(timezone.utc)
        index = RuntimeEventIndex(events)
        retrieval = self._retrieval_projector.project(
            index,
            valid_rounds=valid_rounds,
            valid_branches=valid_branches,
            captured_at=captured_at,
            run_status=run_status,
        )
        evaluations = self._evaluation_projector.project(
            index,
            valid_rounds=valid_rounds,
            valid_branches=valid_branches,
        )
        main_agent = self._main_agent_projector.project(
            index,
            script_direction=script_direction,
            rounds=rounds,
            multipath_decisions=multipath_decisions,
        )
        creative = self._creative_projector.project(
            index,
            valid_rounds=valid_rounds,
            valid_branches=valid_branches,
        )
        unassigned = [
            *retrieval.get("unassigned", []),
            *evaluations.get("unassigned", []),
            *main_agent.get("unassigned", []),
            *creative.get("unassigned", []),
        ]
        return {
            "multipathReviewsByRound": evaluations.get(
                "multipathReviewsByRound", {}
            ),
            "overallEvaluationsByRound": evaluations.get(
                "overallEvaluationsByRound", {}
            ),
            "retrievalStagesByBranch": retrieval.get(
                "retrievalStagesByBranch", {}
            ),
            "mainAgentDecisions": main_agent,
            "creativeByBranch": creative.get("creativeByBranch", {}),
            "unassigned": unassigned,
        }