Skip to content

mis.pipeline.execution

[docs] module mis.pipeline.execution 

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
from __future__ import annotations

import abc
import asyncio
from enum import Enum
from typing import Callable, Generic, TypeVar


class Status(str, Enum):
    IN_PROGRESS = "in-progress"
    SUCCESS = "success"
    FAILURE = "failure"


Result = TypeVar("Result")
Transformed = TypeVar("Transformed")


class Execution(abc.ABC, Generic[Result]):
    """
    A task being sent to a quantum device and which
    may need some time before it is completed.
    """

    @abc.abstractmethod
    def result(self) -> Result: ...

    @abc.abstractmethod
    async def wait(self) -> Result: ...

    @abc.abstractmethod
    def status(self) -> Status: ...

    def map(self, transform: Callable[[Result], Transformed]) -> Execution[Transformed]:
        """
        Apply a transformation to the result once it is
        complete.
        """
        return MappedExecution(self, transform)

    @classmethod
    def success(cls, result: Result) -> Execution[Result]:
        """
        Shortcut to return a result that has already succeeded.
        """
        return SuccessfulExecution(result)


class WaitingExecution(Execution[Result]):
    """
    A task being sent to a quantum device and which
    definitely needs some time before it is completed.

    Unless you're implementing new executors, you're probably
    not interested in this class.
    """

    def __init__(self, sleep_sec: int):
        self._sleep_sec = sleep_sec

    async def wait(self) -> Result:
        while self.status() is Status.IN_PROGRESS:
            await asyncio.sleep(self._sleep_sec)
        return self.result()


class MappedExecution(Execution[Transformed]):
    """
    The result of calling `map` on an `Execution`.

    Unless you're implementing new executors, you're probably
    not interested in this class.
    """

    def __init__(self, origin: Execution[Result], transform: Callable[[Result], Transformed]):
        self._cache_filled = False
        self._origin = origin
        self._transform = transform

    def result(self) -> Transformed:
        original = self._origin.result()
        return self._transform(original)

    def status(self) -> Status:
        return self._origin.status()

    async def wait(self) -> Transformed:
        original = await self._origin.wait()
        return self._transform(original)


class SuccessfulExecution(Execution[Result]):
    """
    An execution that is already completed.

    Unless you're implementing new executors, you're probably
    not interested in this class.
    """

    def __init__(self, result: Result):
        self._result = result

    def status(self) -> Status:
        return Status.SUCCESS

    async def wait(self) -> Result:
        # No need to wait.
        return self.result()

    def result(self) -> Result:
        return self._result

    def map(self, transform: Callable[[Result], Transformed]) -> Execution[Transformed]:
        # Since we know that we're not waiting for anything,
        # we can perform the `map` call immediately.
        mapped = transform(self.result())
        return SuccessfulExecution(result=mapped)