Source code for devctrl.gui.app.task_host

"""Shared, cross-module task system host.

There is a **single** global :class:`~devctrl.gui.task.TaskManagerWidget` owned
by the host. Every module registers its task runners into it, so one task list
can naturally mix tasks from different modules and run them sequentially (e.g.
a TAS measurement right after a PRS spectrum on shared hardware).

No scheduling engine is written: sequential execution of an ordered task list
covers "task B after task A". Safety on shared hardware is enforced by the
app-global blocking signals (§6 of the plan).
"""
import logging
import os
from typing import Optional

from devctrl.gui import task as devtask

log = logging.getLogger(__name__)




class TaskHost:
    """Holds the shared task manager and helper state.

    The widget is created lazily so it is available even when no module uses the
    task system heavily; the host keeps a reference for the console/plugins.
    """



    def __init__(self):
        self._manager: devtask.TaskManagerWidget | None = None
        self._preset_manager: devtask.TaskPresetManager | None = None
        self.task_directory_changed = None  # callable(last_dir)
        self.tasklist_directory_changed = None  # callable(last_dir)




    def get_manager(self) -> devtask.TaskManagerWidget:
        """Return the shared task manager, creating it lazily if needed.

        :raises RuntimeError: If the manager could not be created (usually only
            when called before :meth:`ensure_manager` provided the required dirs).
        """
        self.ensure_manager()
        return self._manager




    def ensure_manager(self, preset_manager: Optional[devtask.TaskPresetManager] = None,
                       initial_task_directory: Optional[str] = None,
                       initial_tasklist_directory: Optional[str] = None,
                       tasklist_path: Optional[str] = None) -> devtask.TaskManagerWidget:
        """Create the single global :class:`TaskManagerWidget` if not present.

        If the manager already exists, the preset manager and directory arguments
        are applied to it, so a module can (re-)configure the manager even when
        the host or another module has already created it. A call without
        arguments never modifies the existing configuration.

        If the manager does not exist yet, missing arguments are replaced by
        sensible defaults below the host cache directory, so the task system
        works even before a module supplies its own config paths.

        :param preset_manager: A :class:`TaskPresetManager`, or None to create one in the default preset directory.
        :param initial_task_directory: Initial "load task settings" directory.
        :param initial_tasklist_directory: Initial "load tasklist" directory.
        :param tasklist_path: Path to load the task list from on creation (``None`` skips). Ignored when the manager
            already exists.
        :return: The shared :class:`TaskManagerWidget`.
        """
        if self._manager is None:
            if preset_manager is None or initial_task_directory is None or initial_tasklist_directory is None:
                from devctrl.gui.app.config import AppConfigBase
                default_dir = os.path.join(AppConfigBase.cache_dir(), "tasks")
                if preset_manager is None:
                    preset_manager = devtask.TaskPresetManager(os.path.join(default_dir, "presets"))
                if initial_task_directory is None:
                    initial_task_directory = default_dir
                if initial_tasklist_directory is None:
                    initial_tasklist_directory = default_dir

            self._manager = devtask.TaskManagerWidget(
                preset_manager=preset_manager,
                initial_task_directory=initial_task_directory,
                initial_tasklist_directory=initial_tasklist_directory,
            )
            if tasklist_path:
                try:
                    self._manager.w_tasklist.load_tasklist(tasklist_path)
                except Exception as e:  # pragma: no cover - defensive
                    log.warning(f"Could not load initial tasklist from '{tasklist_path}': {e}")
        else:
            # Apply the configuration to the already-created manager so a module
            # can replace the defaults (or another module's configuration).
            w_tasklist = self._manager.w_tasklist
            if preset_manager is not None:
                w_tasklist.preset_manager = preset_manager
            if initial_task_directory is not None:
                w_tasklist.initial_task_directory = initial_task_directory
            if initial_tasklist_directory is not None:
                w_tasklist.initial_tasklist_directory = initial_tasklist_directory
        return self._manager