Implement Feedforward modules

docs/source/feedforward.rst

@@ -0,0 +1,14 @@
+Feedforward
+===========
+
+:mod:`graphix_zx.feedforward` module
+++++++++++++++++++++++++++++++++++++
+
+.. automodule:: graphix_zx.feedforward
+
+Functions
+---------
+
+.. autofunction:: graphix_zx.feedforward.dag_from_flow
+
+.. autofunction:: graphix_zx.feedforward.check_causality

docs/source/focus_flow.rst

@@ -0,0 +1,13 @@
+Focus Flow
+==========
+
+:mod:`graphix_zx.focus_flow` module
++++++++++++++++++++++++++++++++++++
+
+.. automodule:: graphix_zx.focus_flow
+
+Functions
+---------
+
+.. autofunction:: graphix_zx.focus_flow.is_focused
+.. autofunction:: graphix_zx.focus_flow.focus_gflow

docs/source/graphstate.rst

@@ -24,6 +24,7 @@ Functions
 .. autofunction:: graphix_zx.graphstate.compose_sequentially
 .. autofunction:: graphix_zx.graphstate.compose_in_parallel
 .. autofunction:: graphix_zx.graphstate.bipartite_edges
+.. autofunction:: graphix_zx.graphstate.odd_neighbors
 
 Auxiliary Classes
 ------------------

docs/source/references.rst

@@ -9,3 +9,5 @@ Module reference
     matrix
     graphstate
     random_objects
+    feedforward
+    focus_flow

graphix_zx/feedforward.py

@@ -0,0 +1,139 @@
+"""Feedforward correction functions.
+
+This module provides:
+
+- `dag_from_flow`: Construct a directed acyclic graph (DAG) from a flowlike object.
+- `check_causality`: Check if the flowlike object is causal with respect to the graph state.
+"""
+
+from __future__ import annotations
+
+import sys
+from collections.abc import Iterable, Mapping
+from collections.abc import Set as AbstractSet
+from typing import Any
+
+from graphix_zx.graphstate import BaseGraphState, odd_neighbors
+
+if sys.version_info >= (3, 10):
+    from typing import TypeGuard
+else:
+    from typing_extensions import TypeGuard
+
+
+def _is_flow(flowlike: Mapping[int, Any]) -> TypeGuard[Mapping[int, int]]:
+    r"""Check if the flowlike object is a flow.
+
+    Parameters
+    ----------
+    flowlike : `collections.abc.Mapping`\[`int`, `typing.Any`\]
+        A flowlike object to check
+
+    Returns
+    -------
+    `bool`
+        True if the flowlike object is a flow, False otherwise
+    """
+    return all(isinstance(v, int) for v in flowlike.values())
+
+
+def _is_gflow(flowlike: Mapping[int, Any]) -> TypeGuard[Mapping[int, AbstractSet[int]]]:
+    r"""Check if the flowlike object is a GFlow.
+
+    Parameters
+    ----------
+    flowlike : `collections.abc.Mapping`\[`int`, `typing.Any`\]
+        A flowlike object to check
+
+    Returns
+    -------
+    `bool`
+        True if the flowlike object is a GFlow, False otherwise
+    """
+    return all(isinstance(v, AbstractSet) for v in flowlike.values())
+
+
+def dag_from_flow(
+    flowlike: Mapping[int, int] | Mapping[int, AbstractSet[int]], graph: BaseGraphState, *, check: bool = True
+) -> dict[int, set[int]]:
+    r"""Construct a directed acyclic graph (DAG) from a flowlike object.
+
+    Parameters
+    ----------
+    flowlike : `collections.abc.Mapping`\[`int`, `int`\] | `collections.abc.Mapping`\[`int`, `collections.abc.Set`\[`int`\]`\]
+        A flowlike object
+    graph : `BaseGraphState`
+        The graph state
+    check : `bool`, optional
+        Raise an error if a cycle is detected, by default True
+
+    Returns
+    -------
+    `dict`\[`int`, `set`\[`int`\]\]
+        The directed acyclic graph
+
+    Raises
+    ------
+    TypeError
+        If the flowlike object is not a Flow or GFlow
+    ValueError
+        If a cycle is detected in the graph
+    """  # noqa: E501
+    dag = {}
+    outputs = graph.physical_nodes - set(flowlike)
+    for node in flowlike:
+        if _is_flow(flowlike):
+            target_nodes = {flowlike[node]} | graph.neighbors(node) - {node}
+        elif _is_gflow(flowlike):
+            target_nodes = set(flowlike[node] | odd_neighbors(flowlike[node], graph) - {node})
+        else:
+            msg = "Invalid flowlike object"
+            raise TypeError(msg)
+        dag[node] = target_nodes
+    for output in outputs:
+        dag[output] = set()
+
+    if check and not _check_dag(dag):
+        msg = "Cycle detected in the graph"
+        raise ValueError(msg)
+
+    return dag
+
+
+def _check_dag(dag: Mapping[int, Iterable[int]]) -> bool:
+    r"""Check if a directed acyclic graph (DAG) does not contain a cycle.
+
+    Parameters
+    ----------
+    dag : `collections.abc.Mapping`\[`int`, `collections.abc.Iterable`\[`int`\]\]
+        directed acyclic graph
+
+    Returns
+    -------
+    `bool`
+        True if the graph is valid, False otherwise
+    """
+    for node, children in dag.items():
+        for child in children:
+            if node in dag[child]:
+                return False
+    return True
+
+
+def check_causality(graph: BaseGraphState, flowlike: Mapping[int, int] | Mapping[int, AbstractSet[int]]) -> bool:
+    r"""Check if the flowlike object is causal with respect to the graph state.
+
+    Parameters
+    ----------
+    graph : `BaseGraphState`
+        The graph state
+    flowlike : `collections.abc.Mapping`\[`int`, `int`\] | `collections.abc.Mapping`\[`int`, `collections.abc.Set`\[`int`\]\]
+        The flowlike object
+
+    Returns
+    -------
+    `bool`
+        True if the flowlike object is causal, False otherwise
+    """  # noqa: E501
+    dag = dag_from_flow(flowlike, graph, check=False)
+    return _check_dag(dag)

graphix_zx/focus_flow.py

@@ -0,0 +1,204 @@
+"""Focus flow algorithm.
+
+This module provides:
+
+- `is_focused`: Check if a flowlike object is focused.
+- `focus_gflow`: Focus a flowlike object.
+"""
+
+from __future__ import annotations
+
+from graphlib import TopologicalSorter
+from typing import TYPE_CHECKING
+
+from graphix_zx.common import Plane
+from graphix_zx.feedforward import _is_flow, _is_gflow, check_causality, dag_from_flow
+from graphix_zx.graphstate import odd_neighbors
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable, Mapping, Sequence
+    from collections.abc import Set as AbstractSet
+
+    from graphix_zx.graphstate import BaseGraphState
+
+
+def is_focused(flowlike: Mapping[int, int] | Mapping[int, AbstractSet[int]], graph: BaseGraphState) -> bool:
+    r"""Check if a flowlike object is focused.
+
+    Parameters
+    ----------
+    flowlike : `collections.abc.Mapping`\[`int`, `int`\] | `collections.abc.Mapping`\[`int`, `collections.abc.Set`\[`int`\]\]`
+        flowlike object
+    graph : `BaseGraphState`
+        graph state
+
+    Returns
+    -------
+    `bool`
+        True if the flowlike object is focused, False otherwise
+
+    Raises
+    ------
+    TypeError
+        If the flowlike object is not a Flow or GFlow
+    """  # noqa: E501
+    meas_bases = graph.meas_bases
+    outputs = set(graph.output_node_indices)
+
+    focused = True
+    for node in set(flowlike) - outputs:
+        if _is_flow(flowlike):
+            for child in graph.neighbors(flowlike[node]) - outputs:
+                focused &= node == child
+        elif _is_gflow(flowlike):
+            for child in flowlike[node]:
+                if child in outputs:
+                    continue
+                focused &= (meas_bases[child].plane == Plane.XY) or (node == child)
+
+            for child in odd_neighbors(flowlike[node], graph):
+                if child in outputs:
+                    continue
+                focused &= (meas_bases[child].plane != Plane.XY) or (node == child)
+        else:
+            msg = "Invalid flowlike object"
+            raise TypeError(msg)
+
+    return focused
+
+
+def focus_gflow(
+    flowlike: Mapping[int, int] | Mapping[int, AbstractSet[int]], graph: BaseGraphState
+) -> dict[int, set[int]]:
+    r"""Focus a flowlike object.
+
+    Parameters
+    ----------
+    flowlike : `collections.abc.Mapping`\[`int`, `int`\] | `collections.abc.Mapping`\[`int`, `collections.abc.Set`\[`int`\]\]
+        flowlike object
+    graph : `BaseGraphState`
+        graph state
+
+    Returns
+    -------
+    `dict`\[`int`, `set`\[`int`\]\]
+        focused flowlike object
+
+    Raises
+    ------
+    TypeError
+        If the flowlike object is not a Flow or GFlow
+    ValueError
+        if the flowlike object is not causal with respect to the graph state
+    """  # noqa: E501
+    if _is_flow(flowlike):
+        flowlike = {key: {value} for key, value in flowlike.items()}
+    elif _is_gflow(flowlike):
+        flowlike = {key: set(value) for key, value in flowlike.items()}
+    else:
+        msg = "Invalid flowlike object"
+        raise TypeError(msg)
+    if not check_causality(graph, flowlike):
+        msg = "The flowlike object is not causal with respect to the graph state"
+        raise ValueError(msg)
+    outputs = graph.physical_nodes - set(flowlike)
+    dag = dag_from_flow(flowlike, graph)
+    topo_order = list(TopologicalSorter(dag).static_order())
+
+    for output in outputs:
+        topo_order.remove(output)
+
+    for target in topo_order:
+        flowlike = _focus(target, flowlike, graph, topo_order)
+
+    return flowlike
+
+
+def _focus(
+    target: int, gflow: dict[int, set[int]], graph: BaseGraphState, topo_order: Sequence[int]
+) -> dict[int, set[int]]:
+    r"""Subroutine of the focus_gflow function.
+
+    Parameters
+    ----------
+    target : `int`
+        target node to be focused
+    gflow : `dict`\[`int`, `set`\[`int`\]\]
+        gflow object
+    graph : `BaseGraphState`
+        graph state
+    topo_order : `collections.abc.Sequence`\[`int`\]
+        topological order of the graph state
+
+    Returns
+    -------
+    `dict`\[`int`, `set`\[`int`\]
+        flowlike object after focusing the target node
+    """
+    k = 0
+    s_k = _find_unfocused_corrections(target, gflow, graph)
+    while s_k:
+        gflow = _update_gflow(target, gflow, s_k, topo_order)
+        s_k = _find_unfocused_corrections(target, gflow, graph)
+
+        k += 1
+
+    return gflow
+
+
+def _find_unfocused_corrections(target: int, gflow: dict[int, set[int]], graph: BaseGraphState) -> set[int]:
+    r"""Subroutine of the _focus function.
+
+    Parameters
+    ----------
+    target : `int`
+        target node
+    gflow : `dict`\[`int`, `set`\[`int`\]
+        flowlike object
+    graph : `BaseGraphState`
+        graph state
+
+    Returns
+    -------
+    `set`\[`int`\]
+        set of unfocused corrections
+    """
+    meas_bases = graph.meas_bases
+    non_outputs = set(gflow) - set(graph.output_node_indices)
+
+    s_xy_candidate = odd_neighbors(gflow[target], graph) & non_outputs - {target}
+    s_xz_candidate = gflow[target] & non_outputs - {target}
+    s_yz_candidate = gflow[target] & non_outputs - {target}
+
+    s_xy = {node for node in s_xy_candidate if meas_bases[node].plane == Plane.XY}
+    s_xz = {node for node in s_xz_candidate if meas_bases[node].plane == Plane.XZ}
+    s_yz = {node for node in s_yz_candidate if meas_bases[node].plane == Plane.YZ}
+
+    return s_xy | s_xz | s_yz
+
+
+def _update_gflow(
+    target: int, gflow: dict[int, set[int]], s_k: Iterable[int], topo_order: Sequence[int]
+) -> dict[int, set[int]]:
+    r"""Subroutine of the _focus function.
+
+    Parameters
+    ----------
+    target : `int`
+        target node
+    gflow : `dict`\[`int`, `set`\[`int`\]
+        flowlike object
+    s_k : `Iterable`\[`int`\]
+        unfocused correction
+    topo_order : `collections.abc.Sequence`\[`int`\]
+        topological order of the graph state
+
+    Returns
+    -------
+    `dict`\[`int`, `set`\[`int`\]
+        gflow object after updating the target node
+    """
+    minimal_in_s_k = min(s_k, key=topo_order.index)
+    gflow[target] ^= gflow[minimal_in_s_k]
+
+    return gflow