"""Class for open graph states."""
from __future__ import annotations
import dataclasses
from dataclasses import dataclass
from typing import TYPE_CHECKING, Generic, TypeVar
from warnings import warn
import networkx as nx
# override introduced in Python 3.12
from typing_extensions import override
from graphix.clifford import Clifford
from graphix.flow._find_cflow import find_cflow
from graphix.flow._find_gpflow import AlgebraicOpenGraph, PlanarAlgebraicOpenGraph, compute_correction_matrix
from graphix.flow.core import GFlow, PauliFlow
from graphix.fundamentals import AbstractMeasurement, AbstractPlanarMeasurement
from graphix.measurements import BlochMeasurement, Measurement
from graphix.parameter import Parameterizable
from graphix.simulator import Simulable
if TYPE_CHECKING:
from collections.abc import Callable, Collection, Iterable, Mapping, Sequence
# Unpack introduced in Python 3.12
from typing_extensions import Unpack
from graphix import CausalFlow, Pattern
from graphix.circ_ext.extraction import CliffordMap, PauliExponentialDAG
from graphix.parameter import ExpressionOrSupportsFloat, Parameter
from graphix.transpiler import Circuit
from graphix.visualization import DrawKwargs
# TODO: Maybe move these definitions to graphix.fundamentals and graphix.measurements ? Now they are redefined in graphix.flow._find_gpflow, not very elegant.
_AM_co = TypeVar("_AM_co", bound=AbstractMeasurement, covariant=True)
_PM_co = TypeVar("_PM_co", bound=AbstractPlanarMeasurement, covariant=True)
_M = TypeVar("_M", bound=Measurement)
_A = TypeVar("_A", bound=AbstractMeasurement)
_B = TypeVar("_B", bound=AbstractMeasurement)
[docs]
@dataclass(frozen=True)
class OpenGraph(Parameterizable, Simulable[_AM_co], Generic[_AM_co]):
"""An unmutable dataclass providing a representation of open graph states.
Attributes
----------
graph : networkx.Graph[int]
The underlying resource-state graph. Nodes represent qubits and edges represent the application of :math:`CZ` gate on the linked nodes.
input_nodes : Sequence[int]
An ordered sequence of node labels corresponding to the open graph inputs.
output_nodes : Sequence[int]
An ordered sequence of node labels corresponding to the open graph outputs.
measurements : Mapping[int, _AM_co]
A mapping between the non-output nodes of the open graph (``key``) and their corresponding measurement label (``value``). Measurement labels can be specified as `Measurement`, `Plane` or `Axis` instances.
output_cliffords: Mapping[int, Clifford]
A mapping between output nodes of the open graph (``key``) and Clifford operators. This attribute allows to preserve the semantics of (deterministic) patterns with local-clifford commands when extracting the open graph. See :meth:`Pattern.to_opengraph`.
Notes
-----
The inputs and outputs of `OpenGraph` instances in Graphix are defined as ordered sequences of node labels. This contrasts the usual definition of open graphs in the literature, where inputs and outputs are unordered sets of nodes labels. This restriction facilitates the interplay with `Pattern` objects, where the order of input and output nodes represents a choice of Hilbert space basis.
Example
-------
>>> import networkx as nx
>>> from graphix.opengraph import OpenGraph
>>> from graphix.measurements import Measurement
>>>
>>> graph = nx.Graph([(0, 1), (1, 2)])
>>> measurements = {i: Measurement.XY(0.5 * i) for i in range(2)}
>>> input_nodes = [0]
>>> output_nodes = [2]
>>> og = OpenGraph(graph, input_nodes, output_nodes, measurements)
"""
graph: nx.Graph[int]
input_nodes: Sequence[int]
output_nodes: Sequence[int]
measurements: Mapping[int, _AM_co]
output_cliffords: Mapping[int, Clifford] = dataclasses.field(default_factory=dict)
def __post_init__(self) -> None:
"""Validate the correctness of the open graph."""
all_nodes = set(self.graph.nodes)
inputs = set(self.input_nodes)
outputs = set(self.output_nodes)
if not set(self.measurements).issubset(all_nodes):
raise OpenGraphError("All measured nodes must be part of the graph's nodes.")
if not inputs.issubset(all_nodes):
raise OpenGraphError("All input nodes must be part of the graph's nodes.")
if not outputs.issubset(all_nodes):
raise OpenGraphError("All output nodes must be part of the graph's nodes.")
if outputs & self.measurements.keys():
raise OpenGraphError("Output nodes cannot be measured.")
if all_nodes - outputs != self.measurements.keys():
raise OpenGraphError("All non-output nodes must be measured.")
if len(inputs) != len(self.input_nodes):
raise OpenGraphError("Input nodes contain duplicates.")
if len(outputs) != len(self.output_nodes):
raise OpenGraphError("Output nodes contain duplicates.")
if not outputs.issuperset(self.output_cliffords.keys()):
raise OpenGraphError("Cliffords in `output_cliffords` mapping can only act on output nodes.")
[docs]
def to_pattern(self: OpenGraph[Measurement], *, stacklevel: int = 1) -> Pattern:
"""Extract a deterministic pattern from an ``OpenGraph[Measurement]`` if it exists.
Parameters
----------
stacklevel : int, optional
Stack level to use for warnings. Defaults to 1, meaning that warnings
are reported at this function's call site.
Returns
-------
Pattern
A deterministic pattern on the open graph.
Raises
------
OpenGraphError
If the open graph does not have flow.
Notes
-----
- The open graph instance must be of parametric type `Measurement` to allow for a pattern extraction, otherwise it does not contain information about the measurement angles.
- This method proceeds by searching a flow on the open graph and converting it into a pattern as prescripted in Ref. [1_to_pattern]_.
It first attempts to find a causal flow because the corresponding flow-finding algorithm has lower complexity. If it fails, it attemps to find a Pauli flow because this property is more general than a generalised flow, and the corresponding flow-finding algorithms have the same complexity in the current implementation.
References
----------
.. [1_to_pattern] Browne et al., NJP 9, 250 (2007)
"""
try:
bloch_case = self.downcast_bloch()
except TypeError:
flow: PauliFlow[Measurement] | None = None
else:
flow = bloch_case.to_causalflow_or_none()
if flow is None:
flow = self.to_pauliflow_or_none(stacklevel=stacklevel + 1)
if flow is not None:
return flow.to_xzcorrections().to_pattern()
raise OpenGraphError("The open graph does not have flow. It does not support a deterministic pattern.")
[docs]
def draw(self, **options: Unpack[DrawKwargs]) -> None:
"""Visualize the opengraph.
Parameters
----------
options : Unpack[DrawKwargs]
Options controlling graph visualization. See :class:`VisualizationOptions`.
"""
from graphix.visualization import GraphVisualizer # noqa: PLC0415 Avoid circular imports
gv = GraphVisualizer.from_opengraph(og=self, **options)
gv.visualize()
[docs]
def map(self: OpenGraph[_A], f: Callable[[_A], _B]) -> OpenGraph[_B]:
"""Apply a function to every measurement of the open graph and return the resulting open graph.
Parameters
----------
f: Callable[[_A], _B]
Function applied to each measurement.
Returns
-------
OpenGraph[_B]
The resulting open graph.
Notes
-----
The types of measurements can be changed by ``f``.
Example
-------
>>> import networkx as nx
>>> from graphix.opengraph import OpenGraph
>>> from graphix.measurements import Measurement
>>> graph = nx.Graph([(0, 1), (1, 2)])
>>> measurements = {0: Measurement.XY(0.5), 1: Measurement.YZ(1)}
>>> og = OpenGraph(graph, [0], [2], measurements)
>>> og.map(lambda m: m.plane).measurements
{0: Plane.XY, 1: Plane.YZ}
"""
measurements = {node: f(meas) for node, meas in self.measurements.items()}
return OpenGraph(self.graph, self.input_nodes, self.output_nodes, measurements, self.output_cliffords)
[docs]
def to_bloch(self: OpenGraph[Measurement]) -> OpenGraph[BlochMeasurement]:
"""Return the open graph where all measurements are converted to Bloch.
Example
-------
>>> import networkx as nx
>>> from graphix.opengraph import OpenGraph
>>> from graphix.measurements import Measurement
>>> graph = nx.Graph([(0, 1), (1, 2)])
>>> measurements = {0: Measurement.X, 1: Measurement.Y}
>>> og = OpenGraph(graph, [0], [2], measurements)
>>> og.to_bloch().measurements
{0: Measurement.XY(0), 1: Measurement.XY(0.5)}
"""
return self.map(lambda meas: meas.to_bloch())
[docs]
def downcast_bloch(self: OpenGraph[Measurement]) -> OpenGraph[BlochMeasurement]:
"""Return the open graph if all measurements are described as Bloch measurements; raise `TypeError` otherwise.
If this method succeeds, the returned graph is identical to
the original one, but this function narrows the static type of
the measurements. The returned graph can be used with
functions that require all measurements are described as Bloch
measurements, such as :func:`OpenGraph.to_causalflow`.
Example
-------
>>> import networkx as nx
>>> from graphix.opengraph import OpenGraph
>>> from graphix.measurements import Measurement
>>> graph = nx.Graph([(0, 1), (1, 2)])
>>> measurements = {0: Measurement.XY(0.5), 1: Measurement.YZ(1)}
>>> og = OpenGraph(graph, [0], [2], measurements)
>>> og.downcast_bloch().measurements
{0: Measurement.XY(0.5), 1: Measurement.YZ(1)}
>>> measurements = {0: Measurement.X, 1: Measurement.Y}
>>> og = OpenGraph(graph, [0], [2], measurements)
>>> og.downcast_bloch()
Traceback (most recent call last):
...
TypeError: Bloch measurement expected, but Pauli measurement was found.
"""
return self.map(lambda meas: meas.downcast_bloch())
[docs]
def infer_pauli_measurements(
self: OpenGraph[Measurement], rel_tol: float = 1e-09, abs_tol: float = 0.0
) -> OpenGraph[Measurement]:
r"""Return an equivalent open graph in which Bloch measurements close to a Pauli measurement are replaced by Pauli measurements.
Pauli measurements are measurements with a Pauli angle,
i.e. an integer multiple of :math:`\pi/2`.
This method can be used before calling methods such as
`OpenGraph.to_pauliflow_or_none`, that deal with Pauli measurements
in a special way.
Parameters
----------
pattern : pattern
Source pattern.
rel_tol : float, optional
Relative tolerance for comparing angles, passed to :func:`math.isclose`.
Default is ``1e-9``.
abs_tol : float, optional
Absolute tolerance for comparing angles, passed to :func:`math.isclose`.
Default is ``0.0``.
Returns
-------
Pattern
An equivalent open graph in which Bloch measurements close to a Pauli measurement are replaced by Pauli measurements.
Example
-------
>>> import networkx as nx
>>> from graphix.opengraph import OpenGraph
>>> from graphix.measurements import Measurement
>>> graph = nx.Graph([(0, 1), (1, 2)])
>>> measurements = {0: Measurement.XY(0.5), 1: Measurement.YZ(1)}
>>> og = OpenGraph(graph, [0], [2], measurements)
>>> og.infer_pauli_measurements().measurements
{0: Measurement.Y, 1: -Measurement.Z}
"""
return self.map(lambda meas: meas.to_pauli_or_bloch(rel_tol, abs_tol))
[docs]
def isclose(self, other: OpenGraph[_AM_co], rel_tol: float = 1e-09, abs_tol: float = 0.0) -> bool:
"""Check if two open graphs are equal within a given tolerance.
Parameters
----------
other : OpenGraph[_AM_co]
rel_tol : float
Relative tolerance. Optional, defaults to ``1e-09``.
abs_tol : float
Absolute tolerance. Optional, defaults to ``0.0``.
Returns
-------
bool
``True`` if the two open graphs are approximately equal.
Notes
-----
This method verifies the open graphs have:
- Truly equal underlying graphs (not up to an isomorphism).
- Equal input and output nodes.
- Same measurement planes or axes and approximately equal measurement angles if the open graph is of parametric type `Measurement`.
- Equal ``output_cliffords`` mappings.
The static typer does not allow an ``isclose`` comparison of two open graphs with different parametric type. For a structural comparison, see :func:`OpenGraph.is_equal_structurally`.
"""
return self.is_equal_structurally(other) and all(
m.isclose(other.measurements[node], rel_tol=rel_tol, abs_tol=abs_tol)
for node, m in self.measurements.items()
)
[docs]
def is_equal_structurally(self, other: OpenGraph[AbstractMeasurement]) -> bool:
"""Compare the underlying structure of two open graphs.
Parameters
----------
other : OpenGraph[AbstractMeasurement]
Returns
-------
bool
``True`` if ``self`` and ``og`` have the same structure.
Notes
-----
This method verifies the open graphs have:
- Truly equal underlying graphs (not up to an isomorphism).
- Equal input and output nodes. This assumes equal types as well, i.e., if ``self.input_nodes`` is a ``list`` and ``other.input_nodes`` is a ``tuple``, this method will return ``False``.
- Equal ``output_cliffords`` mappings.
It assumes the open graphs are well formed.
The static typer allows comparing the structure of two open graphs with different parametric type.
"""
return (
nx.utils.graphs_equal(self.graph, other.graph)
and self.input_nodes == other.input_nodes
and self.output_nodes == other.output_nodes
and self.output_cliffords == other.output_cliffords
)
[docs]
def neighbors(self, nodes: Collection[int]) -> set[int]:
"""Return the set containing the neighborhood of a set of nodes in the open graph.
Parameters
----------
nodes : Collection[int]
Set of nodes whose neighborhood is to be found
Returns
-------
neighbors_set : set[int]
Neighborhood of set `nodes`.
"""
neighbors_set: set[int] = set()
for node in nodes:
neighbors_set |= set(self.graph.neighbors(node))
return neighbors_set
[docs]
def odd_neighbors(self, nodes: Collection[int]) -> set[int]:
"""Return the set containing the odd neighborhood of a set of nodes in the open graph.
Parameters
----------
nodes : Collection[int]
Set of nodes whose odd neighborhood is to be found
Returns
-------
odd_neighbors_set : set[int]
Odd neighborhood of set `nodes`.
"""
odd_neighbors_set: set[int] = set()
for node in nodes:
odd_neighbors_set ^= self.neighbors([node])
return odd_neighbors_set
[docs]
def to_causalflow(self: OpenGraph[_PM_co]) -> CausalFlow[_PM_co]:
"""Try to extract a causal flow on the open graph.
This method is a wrapper over :func:`OpenGraph.to_causalflow_or_none` with a single return type.
This method requires all measurements to be planar: to extract
the causal flow from a graph that contains Pauli measurements,
you can apply :func:`OpenGraph.to_bloch()` first.
Returns
-------
CausalFlow[_PM_co]
A causal flow object if the open graph has causal flow.
Raises
------
OpenGraphError
If the open graph does not have a causal flow.
See Also
--------
:func:`OpenGraph.to_causalflow_or_none`
"""
cf = self.to_causalflow_or_none()
if cf is None:
raise OpenGraphError("The open graph does not have a causal flow.")
return cf
[docs]
def to_gflow(self: OpenGraph[_PM_co]) -> GFlow[_PM_co]:
r"""Try to extract a maximally delayed generalised flow (gflow) on the open graph.
This method is a wrapper over :func:`OpenGraph.to_gflow_or_none` with a single return type.
This method requires all measurements to be planar: to extract
a gflow from a graph that contains Pauli measurements, you
can apply :func:`OpenGraph.to_bloch()` first.
Returns
-------
GFlow[_PM_co]
A gflow object if the open graph has gflow.
Raises
------
OpenGraphError
If the open graph does not have a gflow.
See Also
--------
:func:`OpenGraph.to_gflow_or_none`
"""
gf = self.to_gflow_or_none()
if gf is None:
raise OpenGraphError("The open graph does not have a gflow.")
return gf
[docs]
def to_pauliflow(self: OpenGraph[_AM_co], *, stacklevel: int = 1) -> PauliFlow[_AM_co]:
r"""Try to extract a maximally delayed Pauli on the open graph.
This method is a wrapper over :func:`OpenGraph.to_pauliflow_or_none` with a single return type.
Parameters
----------
stacklevel : int, optional
Stack level to use for warnings. Defaults to 1, meaning that warnings
are reported at this function's call site.
Returns
-------
PauliFlow[_AM_co]
A Pauli flow object if the open graph has Pauli flow.
Raises
------
OpenGraphError
If the open graph does not have a Pauli flow.
See Also
--------
:func:`OpenGraph.to_pauliflow_or_none`, :func:`OpenGraph.infer_pauli_measurements`
"""
pf = self.to_pauliflow_or_none(stacklevel=stacklevel + 1)
if pf is None:
raise OpenGraphError("The open graph does not have a Pauli flow.")
return pf
[docs]
def to_causalflow_or_none(self: OpenGraph[_PM_co]) -> CausalFlow[_PM_co] | None:
"""Return a causal flow on the open graph if it exists.
This method requires all measurements to be planar: to find
the causal flow from a graph that contains Pauli measurements,
you can apply :func:`OpenGraph.to_bloch()` first.
Returns
-------
CausalFlow[_PM_co] | None
A causal flow object if the open graph has causal flow or ``None`` otherwise.
See Also
--------
:func:`OpenGraph.to_causalflow`
Notes
-----
- The open graph instance must be of parametric type `Measurement` or `Plane` since the causal flow is only defined on open graphs with :math:`XY` measurements.
- This function implements the algorithm presented in Ref. [1] with polynomial complexity on the number of nodes, :math:`O(N^2)`.
References
----------
[1] Mhalla and Perdrix, (2008), Finding Optimal Flows Efficiently, doi.org/10.1007/978-3-540-70575-8_70
"""
return find_cflow(self)
[docs]
def to_gflow_or_none(self: OpenGraph[_PM_co]) -> GFlow[_PM_co] | None:
r"""Return a maximally delayed Pauli on the open graph if it exists.
This method requires all measurements to be planar: to find
a gflow in a graph that contains Pauli measurements, you
can apply :func:`OpenGraph.to_bloch()` first.
Returns
-------
GFlow[_PM_co] | None
A gflow object if the open graph has gflow or ``None`` otherwise.
See Also
--------
:func:`OpenGraph.to_gflow`
Notes
-----
- The open graph instance must be of parametric type `Measurement` or `Plane` since the gflow is only defined on open graphs with planar measurements. Measurement instances with a Pauli angle (integer multiple of :math:`\pi/2`) are interpreted as `Plane` instances, in contrast with :func:`OpenGraph.to_pauliflow_or_none`.
- This function implements the algorithm presented in Ref. [1] with polynomial complexity on the number of nodes, :math:`O(N^3)`.
References
----------
[1] Mitosek and Backens, 2024 (arXiv:2410.23439).
"""
aog = PlanarAlgebraicOpenGraph(self)
correction_matrix = compute_correction_matrix(aog)
if correction_matrix is None:
return None
return GFlow.from_correctionmatrix_or_none(
correction_matrix
) # The constructor returns `None` if the correction matrix is not compatible with any partial order on the open graph.
[docs]
def to_pauliflow_or_none(self: OpenGraph[_AM_co], *, stacklevel: int = 1) -> PauliFlow[_AM_co] | None:
r"""Return a maximally delayed Pauli on the open graph if it exists.
Parameters
----------
stacklevel : int, optional
Stack level to use for warnings. Defaults to 1, meaning that warnings
are reported at this function's call site.
Returns
-------
PauliFlow[_AM_co] | None
A Pauli flow object if the open graph has Pauli flow or ``None`` otherwise.
See Also
--------
:func:`OpenGraph.to_pauliflow`, :func:`OpenGraph.infer_pauli_measurements`
Notes
-----
- Measurements are considered to be Pauli only if they are represented as :class:`graphix.measurements.PauliMeasurement`. The method :func:`OpenGraph.infer_pauli_measurements` can be called first to convert all measurements with a Pauli angle (integer multiple of :math:`\pi/2`) into Pauli measurements.
- This function implements the algorithm presented in Ref. [1] with polynomial complexity on the number of nodes, :math:`O(N^3)`.
References
----------
[1] Mitosek and Backens, 2024 (arXiv:2410.23439).
Example
-------
>>> import warnings
>>> import networkx as nx
>>> from graphix.opengraph import OpenGraph
>>> from graphix.measurements import Measurement
>>> graph = nx.Graph([(0, 1), (1, 2)])
>>> measurements = {0: Measurement.XZ(0.5), 1: Measurement.XZ(0.5)}
>>> og = OpenGraph(graph, [0], [2], measurements)
>>> with warnings.catch_warnings():
... warnings.filterwarnings("ignore", message="Open graph with non-inferred Pauli measurements.")
... og.to_pauliflow()
Traceback (most recent call last):
...
graphix.opengraph.OpenGraphError: The open graph does not have a Pauli flow.
>>> og.infer_pauli_measurements().measurements
{0: Measurement.X, 1: Measurement.X}
>>> str(og.infer_pauli_measurements().to_pauliflow())
'p(0) = {1}, p(1) = {2}; {0, 1} < {2}'
"""
self._warn_non_inferred_pauli_measurements(stacklevel=stacklevel + 1)
aog = AlgebraicOpenGraph(self)
correction_matrix = compute_correction_matrix(aog)
if correction_matrix is None:
return None
return PauliFlow.from_correctionmatrix_or_none(
correction_matrix
) # The constructor returns `None` if the correction matrix is not compatible with any partial order on the open graph.
[docs]
def to_circuit(
self: OpenGraph[Measurement],
pexp_cp: Callable[[PauliExponentialDAG, Circuit], None] | None = None,
cm_cp: Callable[[CliffordMap, Circuit], None] | None = None,
*,
stacklevel: int = 1,
) -> Circuit:
"""Extract a unitary in the form of a circuit from an open graph resource state.
This method acts as a wrapper around the circuit extraction routine, simplifying
its usage. It first attempts to extract the Pauli flow of the open graph, then
applies the circuit extraction procedure described in Ref. [1_to_circuit]_, and finally compiles
the resulting circuit using the provided passes.
To obtain the open graph's unitary in the form of a Pauli exponential DAG along with
a Clifford transformation, as presented in Ref. [1_to_circuit]_, one should instead operate
directly on the flow object using :meth:`PauliFlow.extract_circuit`.
Parameters
----------
pexp_cp : Callable[[PauliExponentialDAG, Circuit], None] | None
Compilation pass to synthesize a Pauli exponential DAG.
If ``None`` (default), :func:`graphix.circ_ext.compilation.pexp_ladder_pass` is
employed.
cm_cp : Callable[[CliffordMap, Circuit], None] | None
Compilation pass to synthesize a Clifford map. If ``None`` (default),
:func:`graphix.circ_ext.compilation.cm_berg_pass` is employed. This pass
only handles unitaries so far (Clifford maps with the same number of input
and ouptut nodes).
stacklevel : int, optional
Stack level to use for warnings. Defaults to 1, meaning that warnings
are reported at this function's call site.
Returns
-------
Circuit
Quantum circuit represented as a set of instructions.
Notes
-----
- The open graph instance must be of parametric type ``Measurement`` to allow
for a circuit extraction, otherwise it does not contain information about the
measurement angles.
- This wrapper extracts a Pauli flow rather than a gflow, as the former is more
general while the underlying extraction algorithms have the same computational
complexity in both cases. The resulting unitary is identical whether it is
obtained from a ``GFlow`` or from a ``PauliFlow`` with inferred Pauli measurements.
However, compilation passes that simultaneously diagonalize Pauli exponentials
within the same layer of the Pauli exponential DAG may benefit from flows of
lower depth, which is often the case for Pauli flow. The pass
:func:`graphix.circ_ext.compilation.pexp_ladder_pass` does not take into account
the flow's depth.
References
----------
.. [1_to_circuit] Simmons, 2021 (arXiv:2109.05654).
Examples
--------
>>> import warnings
>>> import networkx as nx
>>> from graphix.opengraph import OpenGraph
>>> from graphix.measurements import Measurement
>>> og = OpenGraph(
... graph=nx.Graph([(0, 1), (1, 2), (3, 4), (4, 5), (6, 7), (7, 8), (1, 3), (4, 6)]),
... input_nodes=(0, 3, 6),
... output_nodes=(2, 5, 8),
... measurements=dict.fromkeys((0, 1, 3, 4, 6, 7), Measurement.XY(angle=0)),
... )
>>> with warnings.catch_warnings():
... warnings.filterwarnings("ignore", message="Open graph with non-inferred Pauli measurements.")
... og.to_circuit()
Circuit(width=3, instr=[H(2), H(1), CNOT(2, 1), H(1), H(1), H(0), CNOT(2, 0), CNOT(1, 0), H(2), H(1), H(0)])
>>> # The default compilation passes do not exploit the lower depth of the Pauli flow
>>> # compared the gflow.
>>> og.infer_pauli_measurements().to_circuit()
Circuit(width=3, instr=[H(2), H(1), CNOT(2, 1), H(1), H(1), H(0), CNOT(2, 0), CNOT(1, 0), H(2), H(1), H(0)])
"""
return (
self.to_pauliflow(stacklevel=stacklevel + 1)
.extract_circuit(stacklevel=stacklevel + 1)
.to_circuit(pexp_cp=pexp_cp, cm_cp=cm_cp)
)
[docs]
def compose(self, other: OpenGraph[_AM_co], mapping: Mapping[int, int]) -> tuple[OpenGraph[_AM_co], dict[int, int]]:
r"""Compose two open graphs by merging subsets of nodes from ``self`` and ``other``, and relabeling the nodes of ``other`` that were not merged.
Parameters
----------
other : OpenGraph[_AM_co]
Open graph to be composed with ``self``.
mapping : dict[int, int]
Partial relabelling of the nodes in ``other``, with ``keys`` and ``values`` denoting the old and new node labels, respectively.
Returns
-------
og : OpenGraph[_AM_co]
Composed open graph.
mapping_complete : dict[int, int]
Complete relabelling of the nodes in ``other``, with ``keys`` and ``values`` denoting the old and new node label, respectively.
Notes
-----
Let's denote :math:`\{G(V_1, E_1), I_1, O_1\}` the open graph ``self``, :math:`\{G(V_2, E_2), I_2, O_2\}` the open graph ``other``, :math:`\{G(V, E), I, O\}` the resulting open graph ``og`` and ``{v:u}`` an element of ``mapping``.
We define :math:`V, U` the set of nodes in ``mapping.keys()`` and ``mapping.values()``, and :math:`M = U \cap V_1` the set of merged nodes.
The open graph composition requires that
- :math:`V \subseteq V_2`.
- If both ``v`` and ``u`` are measured, the corresponding measurements must have the same plane and angle.
The returned open graph follows this convention:
- :math:`I = (I_1 \cup I_2) \setminus M \cup (I_1 \cap I_2 \cap M)`,
- :math:`O = (O_1 \cup O_2) \setminus M \cup (O_1 \cap O_2 \cap M)`,
- If only one node of the pair ``{v:u}`` is measured, this measure is assigned to :math:`u \in V` in the resulting open graph.
- Input (and, respectively, output) nodes in the returned open graph have the order of the open graph ``self`` followed by those of the open graph ``other``. Merged nodes are removed, except when they are input (or output) nodes in both open graphs, in which case, they appear in the order they originally had in the graph ``self``.
- Clifford operations on output nodes:
- if two output nodes :math:`o_1` and :math:`o_2` are merged, then :math:`C(o_1) = C_2(o_2) \circ C_1(o_1)`;
- if an output node :math:`o_1` is merged with a measured node :math:`m_2`, then :math:`M(o_1) = M_2(m_2) \circ C_1(o_1)`;
- if a measured node :math:`m_1` is merged with an output node :math:`o_2`, then :math:`M(m_1) = M_1(m_1) \circ C_2(o_2)`.
"""
if not (mapping.keys() <= other.graph.nodes):
raise ValueError("Keys of mapping must be correspond to nodes of other.")
if len(mapping) != len(set(mapping.values())):
raise ValueError("Values in mapping contain duplicates.")
measurements = {**self.measurements}
measurements_other = {**other.measurements}
for v, u in mapping.items():
vm = other.measurements.get(v)
um = self.measurements.get(u)
# Validate that shared measurements are compatible
if vm is not None and um is not None and not vm.isclose(um):
raise OpenGraphError(f"Cannot merge nodes with different measurements: {v, vm} -> {u, um}.")
# Apply other's output Cliffords onto self's measurement
if um is not None and (vc := other.output_cliffords.get(v)) is not None:
measurements[u] = um.clifford(vc)
# Apply self's output Cliffords onto other's measurement
if vm is not None and (uc := self.output_cliffords.get(u)) is not None:
measurements_other[v] = vm.clifford(uc)
shift = max((*self.graph.nodes, *mapping.values())) + 1
mapping_sequential = {
node: i for i, node in enumerate(sorted(other.graph.nodes - mapping.keys()), start=shift)
} # assigns new labels to nodes in other not specified in mapping
mapping_complete = {**mapping, **mapping_sequential}
g2_shifted = nx.relabel_nodes(other.graph, mapping_complete)
g = nx.compose(self.graph, g2_shifted)
merged = set(mapping_complete.values()) & self.graph.nodes
def merge_ports(p1: Iterable[int], p2: Iterable[int]) -> list[int]:
p2_mapped = [mapping_complete[node] for node in p2]
p2_set = set(p2_mapped)
part1 = [node for node in p1 if node not in merged or node in p2_set]
part2 = [node for node in p2_mapped if node not in merged]
return part1 + part2
inputs = merge_ports(self.input_nodes, other.input_nodes)
outputs = merge_ports(self.output_nodes, other.output_nodes)
measurements_shifted = {mapping_complete[i]: meas for i, meas in measurements_other.items()}
measurements.update(measurements_shifted)
oset = set(outputs)
output_cliffords = {node: clifford for node, clifford in self.output_cliffords.items() if node in oset}
for node, clifford in other.output_cliffords.items():
if (mapped_node := mapping_complete[node]) in oset:
# if a clifford `C` has been already applied to a node,
# applying a clifford `C'` to the same node is equivalent
# to applying `C'C` to a fresh node.
output_cliffords[mapped_node] = clifford @ output_cliffords.get(mapped_node, Clifford.I)
return OpenGraph(g, inputs, outputs, measurements, output_cliffords), mapping_complete
[docs]
@override
def with_parameter(
self: OpenGraph[_M], variable: Parameter, substitute: ExpressionOrSupportsFloat
) -> OpenGraph[_M]:
"""Substitute a parameter with a value or expression in all measurement angles.
Creates a new open graph where every measurement angle containing the specified variable is updated using the provided substitution. The original open graph instance remains unmodified.
Parameters
----------
variable : Parameter
The symbolic expression to be replaced within the measurement angles.
substitute : ExpressionOrSupportsFloat
The value or symbolic expression to substitute in place of ``variable``.
Returns
-------
OpenGraph[Measurement]
A new instance of OpenGraph with the updated measurement parameters.
Notes
-----
Substitution relies on object identity. You must provide the exact parameter object instance currently stored in the measurements. Passing a new object with the same name will not trigger a substitution if it is not the same instance in memory.
Examples
--------
>>> import networkx as nx
>>> from graphix.fundamentals import Plane
>>> from graphix.measurements import Measurement
>>> from graphix.opengraph import OpenGraph
>>> from graphix.parameter import Placeholder
>>> # Initialize placeholders and open graph
>>> parametric_angles = [Placeholder(f"alpha{i}") for i in range(2)]
>>> measurements = {node: Measurement.XY(angle) for node, angle in enumerate(parametric_angles)}
>>> og = OpenGraph(
... graph=nx.Graph([(0, 1), (1, 2)]),
... input_nodes=[0],
... output_nodes=[2],
... measurements=measurements,
... )
>>> # To perform substitution, use the actual object in memory
>>> new_og = og.with_parameter(parametric_angles[0], 0.3)
>>> # Note: og.with_parameter(Placeholder("alpha0"), 0.3) would not trigger any substitution.
"""
return self.map(lambda meas: meas.with_parameter(variable, substitute))
[docs]
@override
def with_parameters(
self: OpenGraph[_M], assignment: Mapping[Parameter, ExpressionOrSupportsFloat]
) -> OpenGraph[_M]:
"""Perform parallel substitution of multiple parameters in measurement angles.
Creates a new open graph where occurrences of parameters defined in the assignment mapping are replaced by their corresponding values. The original open graph instance remains unmodified.
Parameters
----------
assignment : Mapping[Parameter, ExpressionOrSupportsFloat]
A dictionary-like mapping where keys are the `Parameter` objects to be replaced and values are the new expressions or numerical values.
Returns
-------
OpenGraph[Measurement]
A new instance of OpenGraph with the updated measurement angles.
Notes
-----
The notes provided in :func:`self.subs` apply here as well.
Examples
--------
>>> import networkx as nx
>>> from graphix.fundamentals import Plane
>>> from graphix.measurements import Measurement
>>> from graphix.opengraph import OpenGraph
>>> from graphix.parameter import Placeholder
>>> # Initialize placeholders
>>> alpha = Placeholder("alpha")
>>> beta = Placeholder("beta")
>>> measurements = {0: Measurement.XY(alpha), 1: Measurement.XY(beta)}
>>> og = OpenGraph(nx.Graph([(0, 1)]), [0], [], measurements)
>>> # Substitute multiple parameters at once
>>> subs_map = {alpha: 0.5, beta: 1.2}
>>> new_og = og.with_parameters(subs_map)
"""
return self.map(lambda meas: meas.with_parameters(assignment))
def _warn_non_inferred_pauli_measurements(self, stacklevel: int) -> None:
for m in self.measurements.values():
if isinstance(m, BlochMeasurement) and m.to_pauli_or_none() is not None:
warn("Open graph with non-inferred Pauli measurements.", stacklevel=stacklevel + 1)
return
[docs]
class OpenGraphError(Exception):
"""Exception subclass to handle open graphs errors."""