Source code for ska_tango_base.future._component_manager_mixins

#
# This file is part of the SKA Tango Base project
#
# Distributed under the terms of the BSD 3-clause new license.
# See LICENSE.txt for more info.
"""This module defines SKA Tango device mixin classes for using a component manager."""

import typing

import ska_control_model as scm

from ..executor import TaskExecutorComponentManager
from ..long_running_commands.mixin import AbstractLRCMixin
from ..software_bus import SignalBusMixin
from ..type_hints import (
    TaskCallbackType,
    TaskExecutorProtocol,
)

ComponentManagerT = typing.TypeVar("ComponentManagerT")
TaskExecutorComponentManagerT = typing.TypeVar(
    "TaskExecutorComponentManagerT", bound=TaskExecutorComponentManager
)




class ComponentManagerMixin(SignalBusMixin, typing.Generic[ComponentManagerT]):
    """An abstract mixin class for using a component manager with a SKA Tango device."""



    def on_new_shared_bus(self) -> None:
        """
        Create the component manager when the shared bus is initialised.

        The component manager is created here so that the
        :py:class:`~.software_bus.SharingObserver` bus-propagation
        mechanism can automatically share the bus with the component manager (which
        itself should be a subclass of
        :py:class:`~.software_bus.SharingObserver`).  The
        ``SharingObserver.shared_bus`` setter iterates ``vars(self)`` *after* calling
        ``on_new_shared_bus``, so assigning ``_component_manager`` here ensures the
        component manager is found and receives the bus before any signals are emitted.
        """
        self._component_manager = self.create_component_manager()
        super().on_new_shared_bus()




    def delete_device(self) -> None:
        """Cleanup the component manager."""
        # We try to clean up, but it is important we do not
        # throw from delete_device().
        try:
            with self.allow_internal_threads():
                self.component_manager.cleanup()  # type: ignore[attr-defined]
        except Exception:
            pass
        super().delete_device()


    @property
    def component_manager(self) -> ComponentManagerT:
        """Get the component manager."""
        return self._component_manager



    def create_component_manager(self) -> ComponentManagerT:
        """
        Create and return a component manager for this device.

        :raises NotImplementedError: if not overridden by a subclass.
        """
        raise NotImplementedError(
            "'create_component_manager' method must be implemented by "
            f"'{self.__class__.__name__}'."
        )






class ComponentManagerLRCMixin(
    ComponentManagerMixin[TaskExecutorComponentManagerT], AbstractLRCMixin
):
    """
    A mixin class for using a component manager and long running commands.

    This mixin is intended to be used with a SKA Tango device which has a component
    manager that is a subclass of
    :py:class:`~.executor.executor_component_manager.TaskExecutorComponentManager`,
    which in turn provides a task executor following the
    :py:class:`~.type_hints.TaskExecutorProtocol` that can be used to
    execute long running tasks.
    """

    @property
    def task_executor(self) -> TaskExecutorProtocol:
        """
        Get the task executor.

        :return: The initialised task executor.
        """
        return typing.cast(
            TaskExecutorProtocol,
            getattr(self.component_manager, "_task_executor", None),
        )



    def schedule_abort_task(
        self, task_callback: TaskCallbackType
    ) -> tuple[scm.TaskStatus, str]:
        """
        Schedule an Abort task to begin executing immediately.

        Subclasses should override this to change the behaviour of the
        :py:meth:`!Abort()` command.

        :param task_callback: Notified of progress of the abort command.

        :return: A tuple containing `TaskStatus.IN_PROGRESS` and a message.
        """
        return self.component_manager.abort_tasks(task_callback)