Source code for ewokscore.engine_interface

from pathlib import Path
from abc import ABC, abstractmethod
from typing import Any, Optional, Union, List

from .graph import TaskGraph
from .events.contexts import RawExecInfoType




class WorkflowEngine(ABC):
    """Python projects that provide Ewoks engines for deserializing, serializing and executing
    computational Ewoks graphs can implement this interface.

    To make it discoverable it can be added as an entry-point the the project. For
    example in a `pyproject.toml` file:

    .. code-block: toml

        [project.entry-points."ewoks.engines"]
        "<engine-name>" = "<project-name>.engine:MyWorkflowEngine"
    """



    @abstractmethod
    def execute_graph(
        self,
        graph: TaskGraph,
        *,
        inputs: Optional[List[dict]] = None,
        load_options: Optional[dict] = None,
        varinfo: Optional[dict] = None,
        execinfo: Optional[RawExecInfoType] = None,
        task_options: Optional[dict] = None,
        outputs: Optional[List[dict]] = None,
        merge_outputs: Optional[bool] = True,
        # Engine specific:
        **execute_options,
    ) -> Optional[dict]:
        """Execute a computional Ewoks graph."""
        pass






class WorkflowEngineWithSerialization(WorkflowEngine):
    """Ewoks engines with graph serialization capabilities."""



    @abstractmethod
    def deserialize_graph(
        self,
        graph: Any,
        *,
        inputs: Optional[List[dict]] = None,
        representation: Optional[str] = None,
        root_dir: Optional[Union[str, Path]] = None,
        root_module: Optional[str] = None,
        # Serializer specific:
        **deserialize_options,
    ) -> TaskGraph:
        """Convert a computational graph representation to the canonical in-memory representation `TaskGraph`."""
        pass




    @abstractmethod
    def serialize_graph(
        self,
        graph: TaskGraph,
        destination: Any,
        *,
        representation: Optional[str] = None,
        # Serializer specific:
        **serialize_options,
    ) -> Any:
        """Convert the canonical computational graph representation `TaskGraph` to another representation."""
        pass




    @abstractmethod
    def get_graph_representation(self, graph: Any) -> Optional[str]:
        """Return the representation if the engine recognizes the graph object."""
        pass