feat: add qsystem op for measure leaked

guppylang/std/_internal/compiler/futures.py

@@ -0,0 +1,27 @@
+from collections.abc import Callable, Sequence
+
+from hugr import ops
+from hugr import tys as ht
+
+from guppylang.error import InternalGuppyError
+from guppylang.std._internal.compiler.tket2_exts import FUTURES_EXTENSION
+from guppylang.tys.arg import Argument, TypeArg
+from guppylang.tys.subst import Inst
+
+
+def future_to_hugr(type_args: Sequence[Argument]) -> ht.Type:
+    match type_args:
+        case [TypeArg(ty)]:
+            type_def = FUTURES_EXTENSION.get_type("Future")
+            return type_def.instantiate([ht.TypeTypeArg(ty.to_hugr())])
+        case _:
+            raise InternalGuppyError("Invalid type args for Future type")
+
+
+def future_op(op_name: str) -> Callable[[ht.FunctionType, Inst], ops.DataflowOp]:
+    op_def = FUTURES_EXTENSION.get_op(op_name)
+
+    def op(concrete: ht.FunctionType, args: Inst) -> ops.DataflowOp:
+        return op_def.instantiate([arg.to_hugr() for arg in args], concrete)
+
+    return op

guppylang/std/futures.py

@@ -0,0 +1,30 @@
+from typing import Generic, no_type_check
+
+from guppylang.decorator import guppy
+from guppylang.std._internal.compiler.futures import future_op, future_to_hugr
+from guppylang.std.lang import owned
+from guppylang.tys.param import TypeParam
+
+T = guppy.type_var("T", copyable=False, droppable=False)
+
+_future_params = [TypeParam(0, "T", must_be_copyable=False, must_be_droppable=False)]
+
+
+@guppy.type(future_to_hugr, copyable=False, droppable=False, params=_future_params)
+class Future(Generic[T]):  # type: ignore[misc]
+    """A value of type `T` that is computed asynchronously."""
+
+    @guppy.hugr_op(future_op("Read"))
+    @no_type_check
+    def read(self: "Future[T]" @ owned) -> T:
+        """Reads a value from a future, consuming it."""
+
+    @guppy.hugr_op(future_op("Dup"))
+    @no_type_check
+    def copy(self: "Future[T]") -> "Future[T]":
+        """Duplicate a future."""
+
+    @guppy.hugr_op(future_op("Free"))
+    @no_type_check
+    def discard(self: "Future[T]" @ owned) -> None:
+        """Discards a future without reading it."""

guppylang/std/qsystem/__init__.py

@@ -11,6 +11,8 @@
 from guppylang.std._internal.util import quantum_op
 from guppylang.std.angles import angle, pi
 from guppylang.std.builtins import owned
+from guppylang.std.futures import Future
+from guppylang.std.option import Option, nothing, some
 from guppylang.std.quantum import qubit
 
 
@@ -74,6 +76,52 @@ def reset(q: qubit) -> None: ...
 def qfree(q: qubit @ owned) -> None: ...
 
 
+@guppy.hugr_op(quantum_op("LazyMeasureLeaked", ext=QSYSTEM_EXTENSION))
+@no_type_check
+def _measure_leaked(q: qubit @ owned) -> Future[int]:
+    """Measure the quibit or return 2 if it is leaked."""
+
+
+@guppy
+@no_type_check
+def measure_leaked(q: qubit @ owned) -> "MaybeLeaked":
+    fm = _measure_leaked(q)
+    return MaybeLeaked(fm)
+
+
+@guppy.struct
+@no_type_check
+class MaybeLeaked:
+    """A class representing a measurement that may have leaked.
+
+    This is used to represent the result of `measure_leaked`, which can either
+    return a boolean measurement result or indicate that the qubit has leaked.
+    """
+
+    _measurement: Future[int]  # type: ignore[type-arg]
+
+    @guppy
+    @no_type_check
+    def is_leaked(self: "MaybeLeaked") -> bool:
+        """Check if the measurement indicates a leak."""
+        return self._measurement.copy().read() == 2
+
+    @guppy
+    @no_type_check
+    def to_result(self: "MaybeLeaked @ owned") -> Option[bool]:
+        """Get the measurement result if not leaked."""
+        int_value: int = self._measurement.read()
+        if int_value == 2:
+            return nothing()
+        measurement = int_value == 1
+        return some(measurement)
+
+    @guppy
+    @no_type_check
+    def discard(self: "MaybeLeaked @ owned") -> None:
+        self._measurement.discard()
+
+
 # ------------------------------------------------------
 # --------- Internal definitions -----------------------
 # ------------------------------------------------------

pyproject.toml

@@ -88,7 +88,7 @@ miette-py = { workspace = true }
 
 # Uncomment these to test the latest dependency version during development
 # hugr = { git = "https://github.com/CQCL/hugr", subdirectory = "hugr-py", tag = "hugr-v0.20.2" }
-# tket2-exts = { git = "https://github.com/CQCL/tket2", subdirectory = "tket2-exts", rev = "652a7d0" }
+tket2-exts = { git = "https://github.com/CQCL/tket2", subdirectory = "tket2-exts", rev = "38d1c6f" }
 # tket2 = { git = "https://github.com/CQCL/tket2", subdirectory = "tket2-py", rev = "fcb2131" }
 
 

tests/integration/std/test_futures.py

@@ -0,0 +1,15 @@
+from guppylang.decorator import guppy
+from guppylang.std.futures import Future
+from guppylang.std.lang import owned
+
+
+def test_read(validate):
+    @guppy
+    def main(x: Future[int], y: Future[bool] @ owned) -> int:
+        z = x.copy()
+        if y.read():
+            return z.read()
+        z.discard()
+        return 0
+
+    validate(guppy.compile(main))

tests/integration/test_qsystem.py

@@ -3,7 +3,10 @@
 from guppylang.decorator import guppy
 from guppylang.std.angles import angle
 from guppylang.std.builtins import owned, array
+from guppylang.std.option import Option
 from guppylang.std.qsystem.random import make_discrete_distribution, RNG
+
+from guppylang.std.qsystem import MaybeLeaked, measure_leaked
 from guppylang.std.qsystem.utils import get_current_shot
 from guppylang.std.quantum import qubit, measure_array
 from guppylang.std.qsystem.functional import (
@@ -59,3 +62,20 @@ def test() -> tuple[int, float, int, int, angle, angle]:
         return rint, rfloat, rint_bnd, rint_discrete, rangle, rcangle
 
     validate(guppy.compile(test))
+
+
+def test_measure_leaked(validate):  # type: ignore[no-untyped-def]
+    """Compile the measure_leaked operation."""
+
+    @guppy
+    def test(q: qubit @ owned) -> bool:
+        ml: MaybeLeaked = measure_leaked(q)
+        if ml.is_leaked():
+                ml.discard()
+                q = qubit()
+                m = measure(q)
+                return m
+        b: bool = ml.to_result().unwrap()
+        return b
+
+    validate(guppy.compile(test))