aduc_orchestrator.py

# aduc_orchestrator.py
#
# Copyright (C) August 4, 2025  Carlos Rodrigues dos Santos
#
# Version: 2.2.0
#
# This file contains the core ADUC (Automated Discovery and Orchestration of Complex tasks)
# orchestrator, known as the "Maestro" (Γ). Its responsibility is to manage the high-level
# creative workflow of film production. This version is updated to reflect the final
# refactored project structure with `engineers` and `managers`.


import sys
import os

script_dir = os.path.dirname(os.path.abspath(__file__))
engineers_path = os.path.join(script_dir, "engineers")

print(f"Script directory: {script_dir}")
print(f"Engineers path: {engineers_path}")
print(f"sys.path before import: {sys.path}")

sys.path.insert(0, engineers_path)

print(f"sys.path after insert: {sys.path}")

try:
    from engineers.deformes4D import Deformes4DEngine
    print("Module deformes4D imported successfully!")
except ModuleNotFoundError as e:
    print(f"Error importing deformes4D: {e}")
    sys.exit(1)

try:
    from engineers.deformes3D import deformes3d_engine_singleton
    print("Module deformes3D imported successfully!")
except ModuleNotFoundError as e:
    print(f"Error importing deformes3D: {e}")
    sys.exit(1)


import logging
from typing import List, Dict, Any, Generator, Tuple

import gradio as gr
from PIL import Image, ImageOps

# The logger is configured in app.py; here we just get the instance.
logger = logging.getLogger(__name__)

class AducDirector:
    """
    Represents the Scene Director, responsible for managing the production state.
    Acts as the "score" for the orchestra, keeping track of all generated artifacts
    (script, keyframes, etc.) during the creative process.
    """
    def __init__(self, workspace_dir: str):
        self.workspace_dir = workspace_dir
        os.makedirs(self.workspace_dir, exist_ok=True)
        self.state: Dict[str, Any] = {}
        logger.info(f"The stage is set. Workspace at '{self.workspace_dir}'.")

    def update_state(self, key: str, value: Any) -> None:
        logger.info(f"Notating on the score: State '{key}' updated.")
        self.state[key] = value

    def get_state(self, key: str, default: Any = None) -> Any:
        return self.state.get(key, default)

class AducOrchestrator:
    """
    Implements the Maestro (Γ), the central orchestration layer of the ADUC architecture.
    It does not execute AI tasks directly but delegates each step of the creative
    process (scriptwriting, art direction, cinematography) to the appropriate Specialists.
    """
    def __init__(self, workspace_dir: str):
        self.director = AducDirector(workspace_dir)
        self.editor = Deformes4DEngine(workspace_dir)
        self.painter = deformes3d_engine_singleton
        logger.info("ADUC Maestro is on the podium. Musicians (specialists) are ready.")

    def process_image_for_story(self, image_path: str, size: int, filename: str) -> str:
        """
        Pre-processes a reference image, standardizing it for use by the Specialists.
        """
        img = Image.open(image_path).convert("RGB")
        img_square = ImageOps.fit(img, (size, size), Image.Resampling.LANCZOS)
        processed_path = os.path.join(self.director.workspace_dir, filename)
        img_square.save(processed_path)
        logger.info(f"Reference image processed and saved to: {processed_path}")
        return processed_path

    # --- PRE-PRODUCTION TASKS ---

    def task_generate_storyboard(self, prompt: str, num_keyframes: int, ref_image_paths: List[str],
                                 progress: gr.Progress) -> Tuple[List[str], str, Any]:
        """
        Delegates the task of creating the storyboard to the Scriptwriter (deformes2D_thinker).
        """
        logger.info(f"Act 1, Scene 1: Script. Instructing Scriptwriter to create {num_keyframes} scenes.")
        progress(0.2, desc="Consulting AI Scriptwriter...")

        storyboard = deformes2d_thinker_singleton.generate_storyboard(prompt, num_keyframes, ref_image_paths)

        logger.info(f"Scriptwriter returned the score: {storyboard}")
        self.director.update_state("storyboard", storyboard)
        self.director.update_state("processed_ref_paths", ref_image_paths)
        return storyboard, ref_image_paths[0], gr.update(visible=True, open=True)

    def task_select_keyframes(self, storyboard: List[str], base_ref_paths: List[str],
                              pool_ref_paths: List[str]) -> List[str]:
        """
        Delegates to the Photographer (deformes2D_thinker) the task of selecting keyframes.
        """
        logger.info(f"Act 1, Scene 2 (Photographer Mode): Instructing Photographer to select {len(storyboard)} keyframes.")
        selected_paths = deformes2d_thinker_singleton.select_keyframes_from_pool(storyboard, base_ref_paths, pool_ref_paths)
        logger.info(f"Photographer selected the following scenes: {[os.path.basename(p) for p in selected_paths]}")
        self.director.update_state("keyframes", selected_paths)
        return selected_paths

    def task_generate_keyframes(self, storyboard: List[str], initial_ref_path: str, global_prompt: str,
                                keyframe_resolution: int, progress_callback_factory=None) -> List[str]:
        """
        Delegates to the Art Director (Deformes3DEngine) the task of generating keyframes.
        """
        logger.info("Act 1, Scene 2 (Art Director Mode): Delegating to Art Director.")
        general_ref_paths = self.director.get_state("processed_ref_paths", [])

        final_keyframes = self.painter.generate_keyframes_from_storyboard(
            storyboard=storyboard,
            initial_ref_path=initial_ref_path,
            global_prompt=global_prompt,
            keyframe_resolution=keyframe_resolution,
            general_ref_paths=general_ref_paths,
            progress_callback_factory=progress_callback_factory
        )
        self.director.update_state("keyframes", final_keyframes)
        logger.info("Maestro: Art Director has completed keyframe generation.")
        return final_keyframes

    # --- PRODUCTION & POST-PRODUCTION TASKS ---

    def task_produce_original_movie(self, keyframes: List[str], global_prompt: str, seconds_per_fragment: float,
                                    trim_percent: int, handler_strength: float,
                                    destination_convergence_strength: float,
                                    guidance_scale: float, stg_scale: float, inference_steps: int,
                                    video_resolution: int, use_continuity_director: bool,
                                    progress: gr.Progress) -> Dict[str, Any]:
        """
        Delegates the production of the original master video to the Deformes4DEngine.
        """
        logger.info("Maestro: Delegating production of the original movie to Deformes4DEngine.")
        storyboard = self.director.get_state("storyboard", [])

        result = self.editor.generate_original_movie(
            keyframes=keyframes,
            global_prompt=global_prompt,
            storyboard=storyboard,
            seconds_per_fragment=seconds_per_fragment,
            trim_percent=trim_percent,
            handler_strength=handler_strength,
            destination_convergence_strength=destination_convergence_strength,
            video_resolution=video_resolution,
            use_continuity_director=use_continuity_director,
            guidance_scale=guidance_scale,
            stg_scale=stg_scale,
            num_inference_steps=inference_steps,
            progress=progress
        )
        
        self.director.update_state("final_video_path", result["final_path"])
        self.director.update_state("latent_paths", result["latent_paths"])
        logger.info("Maestro: Original movie production complete.")
        return result

    def task_run_latent_upscaler(self, latent_paths: List[str], chunk_size: int, progress: gr.Progress) -> Generator[Dict[str, Any], None, None]:
        """
        Orchestrates the latent upscaling task.
        """
        logger.info(f"Maestro: Delegating latent upscaling task for {len(latent_paths)} fragments.")
        for update in self.editor.upscale_latents_and_create_video(
            latent_paths=latent_paths,
            chunk_size=chunk_size,
            progress=progress
        ):
            if "final_path" in update and update["final_path"]:
                self.director.update_state("final_video_path", update["final_path"])
                yield update
                break
        logger.info("Maestro: Latent upscaling complete.")
    
    def task_run_hd_mastering(self, source_video_path: str, model_version: str, steps: int, prompt: str, progress: gr.Progress) -> Generator[Dict[str, Any], None, None]:
        """
        Orchestrates the HD mastering task.
        """
        logger.info(f"Maestro: Delegating HD mastering task using SeedVR {model_version}.")
        for update in self.editor.master_video_hd(
            source_video_path=source_video_path,
            model_version=model_version,
            steps=steps,
            prompt=prompt,
            progress=progress
        ):
            if "final_path" in update and update["final_path"]:
                self.director.update_state("final_video_path", update["final_path"])
                yield update
                break
        logger.info("Maestro: HD mastering complete.")

    def task_run_audio_generation(self, source_video_path: str, audio_prompt: str, progress: gr.Progress) -> Generator[Dict[str, Any], None, None]:
        """
        Orchestrates the audio generation task.
        """
        logger.info(f"Maestro: Delegating audio generation task.")
        for update in self.editor.generate_audio_for_final_video(
            source_video_path=source_video_path,
            audio_prompt=audio_prompt,
            progress=progress
        ):
             if "final_path" in update and update["final_path"]:
                self.director.update_state("final_video_path", update["final_path"])
                yield update
                break
        logger.info("Maestro: Audio generation complete.")