lpcore.abc.process のソースコード

from abc import ABC
from typing import Any, ClassVar, Final

from lpcore.abc.task import Task




class Process(Task, ABC):
    """プロセスを表す抽象クラス．

    Args:
        devices (dict[str, Any], optional): プロセスが使用するデバイス．
        parameters (dict[str, Any], optional): プロセスのパラメータ．
        child_tasks (dict[str, Task], optional): 子タスク．
        name (str, optional): プロセスの名前．
            デフォルト値は None．

    See Also:
        :class:`Task`
    """

    devices_required: ClassVar[dict[str, Any]] = {}
    """プロセスが必要とするデバイスを指定する辞書型．"""

    devices_optional: ClassVar[dict[str, Any]] = {}
    """プロセスがオプションで必要とするデバイスを指定する辞書型．"""

    def __init__(self,
                 devices: dict[str, Any] | None = None,
                 parameters: dict[str, Any] | None = None,
                 child_tasks: dict[str, Task] | None = None,
                 name: str | None = None,
                 *args: Any,
                 **kwargs: Any) -> None:
        devices = devices or {}
        self.validate_parameters(devices, self.devices_required, "devices")
        self._devices: Final[dict[str, Any]] = {
            key: value for key, value in devices.items() if key in self.devices_required
        }

        devices_optional = {
            key: value for key, value in devices.items() if key in self.devices_optional
        }
        self.validate_optional_parameters(devices_optional, self.devices_optional,
                                          "devices_optional")
        self._devices_optional: Final[dict[str, Any]] = devices_optional

        super().__init__(parameters, child_tasks, name, *args, **kwargs)

    @property
    def devices(self) -> dict[str, Any]:
        """プロセスが使用するデバイス（イミュータブル）"""
        return {**self._devices, **self._devices_optional}



    def execute(self, **kwargs: Any) -> dict[str, Any]:
        """プロセスを実行するメソッド．

        Returns:
            dict[str, Any]: プロセスの実行結果．
        """
        result = super().execute(**kwargs)

        return result