Fix #474: FocusedPauliFlow and FocusedGFlow classes

CHANGELOG.md

@@ -45,6 +45,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - #545: Added an amplitude damping noise model. Introduces `amplitude_damping_channel` / `two_qubit_amplitude_damping_channel`, the `AmplitudeDampingNoise` / `TwoQubitAmplitudeDampingNoise` noise elements, and `AmplitudeDampingNoiseModel`.
 
+- #474, #564: Added `FocusedPauliFlow` and `FocusedGFlow` classes to encapsulate operations specific to focused flows. Flow-finding algorithms return focused flows by construction. Having dedicated classes removes the need to dynamically recheck that a flow is focused during circuit extraction.
+
 ### Fixed
 
 - #454, #481: Ensure `Pattern.minimize_space` only reduces max-space and does not increase it.

graphix/circ_ext/extraction.py

@@ -17,7 +17,7 @@
     from collections.abc import Set as AbstractSet
 
     from graphix.command import Node
-    from graphix.flow.core import PauliFlow
+    from graphix.flow.core import FocusedPauliFlow, PauliFlow
     from graphix.opengraph import OpenGraph
     from graphix.transpiler import Circuit
 
@@ -303,12 +303,12 @@ class PauliExponential:
     pauli_string: PauliString
 
     @staticmethod
-    def from_measured_node(flow: PauliFlow[Measurement], node: Node) -> PauliExponential:
+    def from_measured_node(flow: FocusedPauliFlow[Measurement], node: Node) -> PauliExponential:
         """Extract the Pauli exponential of a measured node and its focused correction set.
 
         Parameters
         ----------
-        flow : PauliFlow[Measurement]
+        flow : FocusedPauliFlow[Measurement]
             A focused Pauli flow. The resulting Pauli string is extracted from its correction function.
         node : int
             A measured node whose associated Pauli string is computed.
@@ -364,7 +364,7 @@ class PauliExponentialDAG:
     output_nodes: Sequence[int]
 
     @staticmethod
-    def from_focused_flow(flow: PauliFlow[Measurement]) -> PauliExponentialDAG:
+    def from_focusedflow(flow: FocusedPauliFlow[Measurement]) -> PauliExponentialDAG:
         """Extract a Pauli exponential rotation from a focused Pauli flow.
 
         This routine associates a Pauli exponential to each measured node in ``flow``. The flow's partial order defines a partial order between the Pauli exponentials such that Pauli exponentials in the same layer commute.
@@ -444,14 +444,14 @@ class CliffordMap:
     output_nodes: Sequence[int]
 
     @staticmethod
-    def from_focused_flow(flow: PauliFlow[Measurement]) -> CliffordMap:
+    def from_focusedflow(flow: FocusedPauliFlow[Measurement]) -> CliffordMap:
         """Extract a Clifford map from a focused Pauli flow.
 
         This routine associates two Pauli strings (one per generator of the Pauli group, X and Z) to each input node in ``flow.og``.
 
         Parameters
         ----------
-        flow : PauliFlow[Measurement]
+        flow : FocusedPauliFlow[Measurement]
             A focused Pauli flow.
 
         Returns
@@ -612,14 +612,14 @@ def extend_input(og: OpenGraph[Measurement]) -> tuple[OpenGraph[Measurement], di
     return replace(og, graph=graph, input_nodes=new_input_nodes[::-1], measurements=measurements), ancillary_inputs_map
 
 
-def clifford_z_map_from_focused_flow(flow: PauliFlow[Measurement]) -> tuple[PauliString, ...]:
+def clifford_z_map_from_focused_flow(flow: FocusedPauliFlow[Measurement]) -> tuple[PauliString, ...]:
     r"""Extract the images of the Z generators of a Clifford map from a focused Pauli flow.
 
     If the input node is a measured node, the resulting Pauli string is given by the correction set. If the input node is also an output node, the resulting Pauli string is Z (representing the identity map).
 
     Parameters
     ----------
-    flow : PauliFlow[Measurement]
+    flow : FocusedPauliFlow[Measurement]
         A focused Pauli flow.
 
     Returns

graphix/flow/core.py

@@ -594,35 +594,6 @@ class PauliFlow(Generic[_AM_co]):
 
     _CF_PREFIX: str = "p"  # Correction function prefix for printing
 
-    @classmethod
-    def try_from_correction_matrix(cls, correction_matrix: CorrectionMatrix[_AM_co]) -> Self | None:
-        """Initialize a ``PauliFlow`` object from a matrix encoding a correction function.
-
-        Parameters
-        ----------
-        correction_matrix : CorrectionMatrix[_AM_co]
-            Algebraic representation of the correction function.
-
-        Returns
-        -------
-        Self | None
-            A Pauli flow if it exists, ``None`` otherwise.
-
-        Notes
-        -----
-        This method verifies if there exists a partial measurement order on the input open graph compatible with the input correction matrix. See Lemma 3.12, and Theorem 3.1 in Ref. [1]. Failure to find a partial order implies the non-existence of a Pauli flow if the correction matrix was calculated by means of Algorithms 2 and 3 in [1].
-
-        References
-        ----------
-        [1] Mitosek and Backens, 2024 (arXiv:2410.23439).
-        """
-        correction_function = correction_matrix.to_correction_function()
-        partial_order_layers = compute_partial_order_layers(correction_matrix)
-        if partial_order_layers is None:
-            return None
-
-        return cls(correction_matrix.aog.og, correction_function, partial_order_layers)
-
     def to_corrections(self) -> XZCorrections[_AM_co]:
         """Compute the X and Z corrections induced by the Pauli flow encoded in ``self``.
 
@@ -935,6 +906,80 @@ def is_focused(self) -> bool:
                     return False
         return True
 
+    def to_focused(self) -> FocusedPauliFlow[_AM_co]:
+        """Return the Pauli flow as a focused Pauli flow after verifying that it is focused.
+
+        Notes
+        -----
+        If ``self`` is already an instance of :class:`FocusedPauliFlow`,
+        the check is bypassed and ``self`` is returned.
+
+        Raises
+        ------
+        FlowGenericError
+            Raised if the Pauli flow is not focused, with the reason ``NotFocused``.
+
+        Returns
+        -------
+        FocusedPauliFlow[_AM_co]
+            The Pauli flow, checked to be focused.
+        """
+        if not self.is_focused():
+            raise FlowGenericError(FlowGenericErrorReason.NotFocused)
+        return FocusedPauliFlow(self.og, self.correction_function, self.partial_order_layers)
+
+
+class FocusedPauliFlow(PauliFlow[_AM_co], Generic[_AM_co]):
+    """An immutable dataclass representing a focused Pauli flow.
+
+    An instance of ``FocusedPauliFlow`` is a :class:`PauliFlow` that
+    satisfies the :meth:`PauliFlow.is_focused` predicate. Once a
+    ``FocusedPauliFlow`` instance has been obtained, this property
+    does not need to be checked again.
+    """
+
+    @classmethod
+    def try_from_correction_matrix(cls, correction_matrix: CorrectionMatrix[_AM_co]) -> Self | None:
+        """Initialize a ``PauliFlow`` object from a matrix encoding a correction function.
+
+        Parameters
+        ----------
+        correction_matrix : CorrectionMatrix[_AM_co]
+            Algebraic representation of the correction function.
+
+        Returns
+        -------
+        Self | None
+            A Pauli flow if it exists, ``None`` otherwise.
+
+        Notes
+        -----
+        This method verifies if there exists a partial measurement order on the input open graph compatible with the input correction matrix. See Lemma 3.12, and Theorem 3.1 in Ref. [1]. Failure to find a partial order implies the non-existence of a Pauli flow if the correction matrix was calculated by means of Algorithms 2 and 3 in [1].
+
+        References
+        ----------
+        [1] Mitosek and Backens, 2024 (arXiv:2410.23439).
+        """
+        correction_function = correction_matrix.to_correction_function()
+        partial_order_layers = compute_partial_order_layers(correction_matrix)
+        if partial_order_layers is None:
+            return None
+
+        return cls(correction_matrix.aog.og, correction_function, partial_order_layers)
+
+    def check_well_formed(self) -> None:
+        """Verify if the Pauli flow is well formed and focused.
+
+        See :meth:`PauliFlow.check_well_formed` and :meth:`PauliFlow.is_focused` for details.
+        """
+        super().check_well_formed()
+        if not self.is_focused():
+            raise FlowGenericError(FlowGenericErrorReason.NotFocused)
+
+    @override
+    def to_focused(self) -> FocusedPauliFlow[_AM_co]:
+        return self
+
     @cached_property
     def extraction_pauli_strings(self: PauliFlow[Measurement]) -> dict[int, PauliString]:
         """Compute the extraction Pauli strings associated with each node in the correction function.
@@ -956,11 +1001,9 @@ def extraction_pauli_strings(self: PauliFlow[Measurement]) -> dict[int, PauliStr
         This property is cached; the dictionary is computed only once upon the first access and stored for subsequent calls.
         See notes in ``PauliString.from_measured_node`` for additional information.
         """
-        if not self.is_focused():
-            raise ValueError("Flow is not focused.")
         return {node: extraction_ps_from_corrected_node(self, node) for node in self.correction_function}
 
-    def extract_circuit(self: PauliFlow[Measurement]) -> ExtractionResult:
+    def extract_circuit(self: FocusedPauliFlow[Measurement]) -> ExtractionResult:
         """Extract a circuit from a flow.
 
         This routine assumes that the flow ``self`` is focused (see Notes).
@@ -983,8 +1026,8 @@ def extract_circuit(self: PauliFlow[Measurement]) -> ExtractionResult:
         """
         if self.og.output_cliffords:
             raise NotImplementedError("Circuit extraction is not supported for open graphs with Clifford decorations.")
-        pexp_dag = PauliExponentialDAG.from_focused_flow(self)
-        clifford_map = CliffordMap.from_focused_flow(self)
+        pexp_dag = PauliExponentialDAG.from_focusedflow(self)
+        clifford_map = CliffordMap.from_focusedflow(self)
 
         return ExtractionResult(pexp_dag=pexp_dag, clifford_map=clifford_map)
 
@@ -1006,31 +1049,6 @@ class GFlow(PauliFlow[_PM_co], Generic[_PM_co]):
 
     _CF_PREFIX: str = "g"  # Correction function prefix for printing
 
-    @override
-    @classmethod
-    def try_from_correction_matrix(cls, correction_matrix: CorrectionMatrix[_PM_co]) -> Self | None:
-        """Initialize a ``GFlow`` object from a matrix encoding a correction function.
-
-        Parameters
-        ----------
-        correction_matrix : CorrectionMatrix[_PM_co]
-            Algebraic representation of the correction function.
-
-        Returns
-        -------
-        Self | None
-            A gflow if it exists, ``None`` otherwise.
-
-        Notes
-        -----
-        This method verifies if there exists a partial measurement order on the input open graph compatible with the input correction matrix. See Lemma 3.12, and Theorem 3.1 in Ref. [1]. Failure to find a partial order implies the non-existence of a generalised flow if the correction matrix was calculated by means of Algorithms 2 and 3 in [1].
-
-        References
-        ----------
-        [1] Mitosek and Backens, 2024 (arXiv:2410.23439).
-        """
-        return super().try_from_correction_matrix(correction_matrix)
-
     @override
     def to_corrections(self) -> XZCorrections[_PM_co]:
         r"""Compute the XZ-corrections induced by the generalised flow encoded in ``self``.
@@ -1168,6 +1186,68 @@ def node_measurement_label(self, node: int) -> Plane:
         """
         return self.og.measurements[node].to_plane()
 
+    @override
+    def to_focused(self) -> FocusedGFlow[_PM_co]:
+        """Return the gflow as a focused gflow after verifying that it is focused.
+
+        Notes
+        -----
+        If ``self`` is already an instance of :class:`FocusedGFlow`,
+        the check is bypassed and ``self`` is returned.
+
+        Raises
+        ------
+        FlowGenericError
+            Raised if the gflow is not focused, with the reason ``NotFocused``.
+
+        Returns
+        -------
+        FocusedGFlow[_PM_co]
+            The gflow, checked to be focused.
+        """
+        if not self.is_focused():
+            raise FlowGenericError(FlowGenericErrorReason.NotFocused)
+        return FocusedGFlow(self.og, self.correction_function, self.partial_order_layers)
+
+
+class FocusedGFlow(GFlow[_PM_co], FocusedPauliFlow[_PM_co], Generic[_PM_co]):
+    """An immutable dataclass representing a focused gflow.
+
+    An instance of ``FocusedGFlow`` is a :class:`GFlow` that satisfies
+    the :meth:`PauliFlow.is_focused` predicate. Once a
+    ``FocusedGFlow`` instance has been obtained, this property does
+    not need to be checked again.
+    """
+
+    @override
+    @classmethod
+    def try_from_correction_matrix(cls, correction_matrix: CorrectionMatrix[_PM_co]) -> Self | None:
+        """Initialize a ``GFlow`` object from a matrix encoding a correction function.
+
+        Parameters
+        ----------
+        correction_matrix : CorrectionMatrix[_PM_co]
+            Algebraic representation of the correction function.
+
+        Returns
+        -------
+        Self | None
+            A gflow if it exists, ``None`` otherwise.
+
+        Notes
+        -----
+        This method verifies if there exists a partial measurement order on the input open graph compatible with the input correction matrix. See Lemma 3.12, and Theorem 3.1 in Ref. [1]. Failure to find a partial order implies the non-existence of a generalised flow if the correction matrix was calculated by means of Algorithms 2 and 3 in [1].
+
+        References
+        ----------
+        [1] Mitosek and Backens, 2024 (arXiv:2410.23439).
+        """
+        return super().try_from_correction_matrix(correction_matrix)
+
+    @override
+    def to_focused(self) -> FocusedGFlow[_PM_co]:
+        return self
+
 
 @dataclass(frozen=True)
 class CausalFlow(GFlow[_PM_co], Generic[_PM_co]):
@@ -1185,11 +1265,6 @@ class CausalFlow(GFlow[_PM_co], Generic[_PM_co]):
 
     _CF_PREFIX: str = "c"  # Correction function prefix for printing
 
-    @override
-    @classmethod
-    def try_from_correction_matrix(cls, correction_matrix: CorrectionMatrix[_PM_co]) -> None:
-        raise NotImplementedError("Initialization of a causal flow from a correction matrix is not supported.")
-
     @override
     def to_corrections(self) -> XZCorrections[_PM_co]:
         r"""Compute the XZ-corrections induced by the causal flow encoded in ``self``.

graphix/flow/exceptions.py

@@ -91,6 +91,9 @@ class FlowGenericErrorReason(Enum):
     NoPauliFlow = enum.auto()
     """No Pauli flow is compatible with the given XZ-corrections: the GF(2) system that reconstructs the correction function has no solution for at least one measured node."""
 
+    NotFocused = enum.auto()
+    """The Pauli flow is not focused."""
+
 
 class XZCorrectionsOrderErrorReason(Enum):
     """Describe the reason of an `XZCorrectionsOrderError` exception."""
@@ -227,6 +230,8 @@ def __str__(self) -> str:
                 return "Causal flow is only defined on open graphs with XY measurements."
             case FlowGenericErrorReason.NoPauliFlow:
                 return "No Pauli flow is compatible with the XZ-corrections: the GF(2) system that reconstructs the correction function has no solution for at least one measured node."
+            case FlowGenericErrorReason.NotFocused:
+                return "The Pauli flow is not focused."
             case _:
                 assert_never(self.reason)