resolve_client.py

Go to the documentation of this file.

"""
DaVinci Resolve client wrapper.
 
Provides a clean interface to the DaVinci Resolve API with proper error handling
and logging.
"""
 
import logging
from typing import Any, Dict, List, Optional, Union
 
from .utils.platform import setup_resolve_environment, check_resolve_running
 
 
logger = logging.getLogger(__name__)
 
 
class DaVinciResolveError(Exception):
    """Base exception for DaVinci Resolve related errors."""
    pass
 
 
class DaVinciResolveNotRunningError(DaVinciResolveError):
    """Raised when DaVinci Resolve is not running."""
    pass
 
 
class DaVinciResolveConnectionError(DaVinciResolveError):
    """Raised when connection to DaVinci Resolve fails."""
    pass
 
 
class DaVinciResolveClient:
    """
    A clean interface to the DaVinci Resolve API.
    
    This class handles the connection to DaVinci Resolve and provides
    organized methods for interacting with projects, timelines, media, etc.
    """
    
    def __init__(self) -> None:
        self._resolve: Optional[Any] = None
        self._project_manager: Optional[Any] = None
        self._current_project: Optional[Any] = None
        self._is_connected = False
        
    def connect(self) -> None:
        """Connect to DaVinci Resolve."""
        # Check if Resolve is running
        if not check_resolve_running():
            raise DaVinciResolveNotRunningError(
                "DaVinci Resolve is not running. Please start DaVinci Resolve first."
            )
        
        # Set up environment
        if not setup_resolve_environment():
            raise DaVinciResolveConnectionError(
                "Failed to set up DaVinci Resolve environment variables."
            )
        
        try:
            # Import and connect to Resolve
            import DaVinciResolveScript as dvr_script
            self._resolve = dvr_script.scriptapp("Resolve")
            
            if self._resolve is None:
                raise DaVinciResolveConnectionError(
                    "Failed to get Resolve object. Check that DaVinci Resolve is running."
                )
            
            # Get project manager
            self._project_manager = self._resolve.GetProjectManager()
            if self._project_manager is None:
                raise DaVinciResolveConnectionError("Failed to get Project Manager.")
            
            # Get current project if one is open
            self._current_project = self._project_manager.GetCurrentProject()
            
            self._is_connected = True
            logger.info(f"Connected to {self.get_version()}")
            
        except ImportError as e:
            raise DaVinciResolveConnectionError(
                f"Failed to import DaVinciResolveScript: {e}. "
                "Check environment variables and DaVinci Resolve installation."
            )
        except Exception as e:
            raise DaVinciResolveConnectionError(f"Unexpected error connecting: {e}")
    
    def disconnect(self) -> None:
        """Disconnect from DaVinci Resolve."""
        self._resolve = None
        self._project_manager = None
        self._current_project = None
        self._is_connected = False
        logger.info("Disconnected from DaVinci Resolve")
    
    def is_connected(self) -> bool:
        """Check if connected to DaVinci Resolve."""
        return self._is_connected
    
    def _ensure_connected(self) -> None:
        """Ensure we're connected to Resolve."""
        if not self._is_connected:
            raise DaVinciResolveConnectionError("Not connected to DaVinci Resolve")
    
    def _ensure_project(self) -> Any:
        """Ensure we have a current project."""
        self._ensure_connected()
        
        # Refresh current project
        if self._project_manager:
            self._current_project = self._project_manager.GetCurrentProject()
        
        if self._current_project is None:
            raise DaVinciResolveError("No project is currently open")
        
        return self._current_project
    
    # System Information
    def get_version(self) -> str:
        """Get DaVinci Resolve version."""
        self._ensure_connected()
        if self._resolve:
            return f"{self._resolve.GetProductName()} {self._resolve.GetVersionString()}"
        return "Unknown"
    
    def get_current_page(self) -> str:
        """Get the current page (Edit, Color, Fusion, etc.)."""
        self._ensure_connected()
        if self._resolve:
            return self._resolve.GetCurrentPage()
        return "Unknown"
    
    def switch_page(self, page: str) -> bool:
        """Switch to a specific page."""
        self._ensure_connected()
        
        valid_pages = ['media', 'cut', 'edit', 'fusion', 'color', 'fairlight', 'deliver']
        if page.lower() not in valid_pages:
            raise ValueError(f"Invalid page. Must be one of: {', '.join(valid_pages)}")
        
        if self._resolve:
            return bool(self._resolve.OpenPage(page.lower()))
        return False
    
    # Project Management
    def list_projects(self) -> List[str]:
        """List all projects in the current database."""
        self._ensure_connected()
        
        if self._project_manager:
            projects = self._project_manager.GetProjectListInCurrentFolder()
            return [p for p in projects if p]  # Filter out empty strings
        return []
    
    def get_current_project_name(self) -> Optional[str]:
        """Get the name of the currently open project."""
        try:
            project = self._ensure_project()
            return project.GetName()
        except DaVinciResolveError:
            return None
    
    def open_project(self, name: str) -> bool:
        """Open a project by name."""
        self._ensure_connected()
        
        if not self._project_manager:
            return False
        
        # Check if project exists
        projects = self.list_projects()
        if name not in projects:
            raise ValueError(f"Project '{name}' not found. Available: {', '.join(projects)}")
        
        result = self._project_manager.LoadProject(name)
        if result:
            self._current_project = self._project_manager.GetCurrentProject()
            logger.info(f"Opened project: {name}")
        
        return bool(result)
    
    def create_project(self, name: str) -> bool:
        """Create a new project."""
        self._ensure_connected()
        
        if not self._project_manager:
            return False
        
        # Check if project already exists
        projects = self.list_projects()
        if name in projects:
            raise ValueError(f"Project '{name}' already exists")
        
        result = self._project_manager.CreateProject(name)
        if result:
            self._current_project = self._project_manager.GetCurrentProject()
            logger.info(f"Created project: {name}")
        
        return bool(result)
    
    # Timeline Management
    def list_timelines(self) -> List[str]:
        """List all timelines in the current project."""
        project = self._ensure_project()
        
        timeline_count = project.GetTimelineCount()
        timelines = []
        
        for i in range(1, timeline_count + 1):
            timeline = project.GetTimelineByIndex(i)
            if timeline:
                timelines.append(timeline.GetName())
        
        return timelines
    
    def get_current_timeline_name(self) -> Optional[str]:
        """Get the name of the current timeline."""
        try:
            project = self._ensure_project()
            current_timeline = project.GetCurrentTimeline()
            return current_timeline.GetName() if current_timeline else None
        except DaVinciResolveError:
            return None
    
    def create_timeline(self, name: str) -> bool:
        """Create a new timeline."""
        project = self._ensure_project()
        
        media_pool = project.GetMediaPool()
        if not media_pool:
            raise DaVinciResolveError("Failed to get Media Pool")
        
        timeline = media_pool.CreateEmptyTimeline(name)
        if timeline:
            logger.info(f"Created timeline: {name}")
            return True
        
        return False
    
    def switch_timeline(self, name: str) -> bool:
        """Switch to a timeline by name."""
        project = self._ensure_project()
        
        # Find timeline by name
        timeline_count = project.GetTimelineCount()
        for i in range(1, timeline_count + 1):
            timeline = project.GetTimelineByIndex(i)
            if timeline and timeline.GetName() == name:
                result = project.SetCurrentTimeline(timeline)
                if result:
                    logger.info(f"Switched to timeline: {name}")
                return bool(result)
        
        raise ValueError(f"Timeline '{name}' not found")
    
    # Media Pool Management
    def list_media_clips(self) -> List[Dict[str, Any]]:
        """List all clips in the media pool root folder."""
        project = self._ensure_project()
        
        media_pool = project.GetMediaPool()
        if not media_pool:
            raise DaVinciResolveError("Failed to get Media Pool")
        
        root_folder = media_pool.GetRootFolder()
        if not root_folder:
            raise DaVinciResolveError("Failed to get root folder")
        
        clips = root_folder.GetClipList()
        if not clips:
            return []
        
        result = []
        for clip in clips:
            result.append({
                "name": clip.GetName(),
                "duration": clip.GetDuration(),
                "fps": clip.GetClipProperty("FPS") or "Unknown"
            })
        
        return result
    
    def import_media(self, file_path: str) -> bool:
        """Import a media file into the media pool."""
        project = self._ensure_project()
        
        media_pool = project.GetMediaPool()
        if not media_pool:
            raise DaVinciResolveError("Failed to get Media Pool")
        
        # Import the media file
        imported_clips = media_pool.ImportMedia([file_path])
        
        if imported_clips:
            logger.info(f"Imported media: {file_path}")
            return True
        
        return False