"""MBQC pattern according to Measurement Calculus.
ref: V. Danos, E. Kashefi and P. Panangaden. J. ACM 54.2 8 (2007)
"""
from __future__ import annotations
import copy
import dataclasses
import enum
import itertools
import warnings
from collections.abc import Iterable, Iterator, Mapping
from collections.abc import Set as AbstractSet
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import TYPE_CHECKING, SupportsFloat
from warnings import warn
import networkx as nx
# override introduced in Python 3.12
from typing_extensions import assert_never, override
from graphix import command, optimization
from graphix.command import CommandKind, Node
from graphix.flow.exceptions import FlowError
from graphix.fundamentals import Plane
from graphix.measurements import BlochMeasurement, Measurement, Outcome, toggle_outcome
from graphix.parameter import InplaceParameterizable
from graphix.pretty_print import OutputFormat, pattern_to_str
from graphix.qasm3_exporter import pattern_to_qasm3_lines
from graphix.sim import DensityMatrix, MBQCTensorNet, Statevector
from graphix.simulator import Simulable
from graphix.space_minimization import pattern_max_space
from graphix.states import BasicStates
from graphix.visualization import GraphVisualizer
if TYPE_CHECKING:
from collections.abc import Callable, Collection, Container, Iterator
from typing import TypeVar
# Unpack introduced in Python 3.12
from typing_extensions import Unpack
from graphix.clifford import Clifford
from graphix.command import CommandType
from graphix.flow.core import CausalFlow, GFlow, PauliFlow, XZCorrections
from graphix.opengraph import OpenGraph
from graphix.parameter import ExpressionOrSupportsFloat, Parameter
from graphix.space_minimization import SpaceMinimizationHeuristic
from graphix.states import State
from graphix.visualization import DrawKwargs
K = TypeVar("K")
V = TypeVar("V")
_BuiltinBackendState = DensityMatrix | Statevector | MBQCTensorNet
[docs]
class DrawPatternAnnotations(Enum):
"""Enumeration to indicate the possible annotations for `Pattern.draw`."""
Flow = enum.auto()
XZCorrections = enum.auto()
[docs]
class Pattern(InplaceParameterizable, Simulable[Measurement]):
"""
MBQC pattern class.
Pattern holds a sequence of commands to operate the MBQC (Pattern.seq),
and provide modification strategies to improve the structure and simulation
efficiency of the pattern accoring to measurement calculus.
ref: V. Danos, E. Kashefi and P. Panangaden. J. ACM 54.2 8 (2007)
"""
__seq: list[CommandType]
[docs]
def __init__(
self,
input_nodes: Iterable[int] | None = None,
cmds: Iterable[CommandType] | None = None,
output_nodes: Iterable[int] | None = None,
) -> None:
"""
Construct a pattern.
Parameters
----------
input_nodes : Iterable[int] | None
Optional. List of input qubits.
cmds : Iterable[CommandType] | None
Optional. List of initial commands.
output_nodes : Iterable[int] | None
Optional. List of output qubits.
"""
if input_nodes is None:
self.__input_nodes = []
else:
self.__input_nodes = list(input_nodes) # input nodes (list() makes our own copy of the list)
self.__n_node = len(self.__input_nodes) # total number of nodes in the graph state
self.__seq = []
# output nodes are initially a copy input nodes, since none are measured yet
self.__output_nodes = list(self.__input_nodes)
if cmds is not None:
self.extend(cmds)
if output_nodes is not None:
self.reorder_output_nodes(output_nodes)
[docs]
def add(self, cmd: CommandType) -> None:
"""Add command to the end of the pattern.
An MBQC command is an instance of :class:`graphix.command.CommandType`.
Parameters
----------
cmd : :class:`graphix.command.CommandType`
MBQC command.
"""
match cmd.kind:
case CommandKind.N:
self.__n_node += 1
self.__output_nodes.append(cmd.node)
case CommandKind.M:
if cmd.node in self.__output_nodes:
self.__output_nodes.remove(cmd.node)
self.__seq.append(cmd)
[docs]
def extend(self, *cmds: CommandType | Iterable[CommandType]) -> None:
"""Add sequences of commands.
:param cmds: sequences of commands
"""
for item in cmds:
if isinstance(item, Iterable):
for cmd in item:
self.add(cmd)
else:
self.add(item)
[docs]
def clear(self) -> None:
"""Clear the sequence of pattern commands."""
self.__n_node = len(self.__input_nodes)
self.__seq = []
self.__output_nodes = list(self.__input_nodes)
[docs]
def replace(self, cmds: list[CommandType], input_nodes: list[int] | None = None) -> None:
"""Replace pattern with a given sequence of pattern commands.
:param cmds: list of commands
:param input_nodes: optional, list of input qubits (by default, keep the same input nodes as before)
"""
if input_nodes is not None:
self.__input_nodes = list(input_nodes)
self.clear()
self.extend(cmds)
[docs]
def node_mapping(
self,
mapping: Mapping[Node, Node] | None = None,
*,
start: int = 0,
avoids: Collection[Node] | None = None,
preserves: Collection[Node] | None = None,
) -> dict[Node, Node]:
"""Compute an injective mapping of node indices.
The resulting mapping can be passed to :meth:`reindex`.
Fresh indices are allocated in increasing order, starting at
``start`` and skipping any values already used.
Parameters
----------
mapping: Mapping[Node, Node] | None, optional
An initial (partial) mapping that will be extended.
A ``ValueError`` is raised if the supplied mapping is not
injective.
start: int, optional
The starting index from which fresh indices are allocated.
avoids: Collection[Node] | None, optional
Indices that must not be assigned to a node.
preserves: Collection[Node] | None, optional
Nodes that should not appear in the mapping.
Their index will not be changed by :meth:`reindex`.
Returns
-------
dict[Node, Node]
A dictionary that maps each node of ``self`` (except those
listed in ``preserves``) to a distinct integer index.
Notes
-----
The resulting dictionary is guaranteed to be an extension of
``mapping``, i.e., all entries from ``mapping`` appear in the
resulting dictionary. In particular, the constraints specified
in ``avoids`` and ``preserves`` only affect nodes that are
freshly generated by the function; they do not apply to the
nodes already present in ``mapping``.
"""
nodes = self.nodes()
if mapping is None:
result = {}
else:
result = dict(mapping)
unknown_nodes = [node for node in mapping if node not in nodes]
if unknown_nodes:
unknown_nodes_str = ", ".join(map(str, unknown_nodes))
raise ValueError(f"Some keys in the initial mapping are not nodes: {unknown_nodes_str}")
duplicates = _duplicates_in_mapping(result)
if duplicates:
reason = ", ".join(f"{antecedents} are mapped to {value}" for value, antecedents in duplicates.items())
raise ValueError(f"Initial mapping is not injective: {reason}")
used_indices = set(avoids or ())
used_indices.update(result.values())
if preserves:
used_indices.update(preserves)
preserves_set = preserves if isinstance(preserves, AbstractSet) else set(preserves)
else:
preserves_set = set()
candidate = start
for node in sorted(nodes):
if node in result or node in preserves_set:
continue
while candidate in used_indices:
candidate += 1
result[node] = candidate
candidate += 1
return result
[docs]
def reindex(
self, mapping: Callable[[Node], Node] | Mapping[Node, Node] | None = None, *, copy: bool = False
) -> Pattern:
"""Return a pattern whose nodes have been re-indexed using ``f``.
This method does not verify that ``f`` is injective. The
semantic of the pattern is only preserved when ``f`` is
injective. A non-injective mapping can even break the
runnability of the resulting pattern.
Parameters
----------
mapping : Callable[[Node], Node] | Mapping[Node, Node] | None, optional
A function or a mapping that translates the current node
indices to new ones. Indices that are not present in the
mapping are left unchanged. If ``f`` is omitted, a
default mapping is computed by using :meth:`node_mapping`.
copy : bool, optional
If ``True``, the current pattern remains unchanged, and a
new pattern is returned. The default is ``False``, meaning
that changes are performed in place.
Returns
-------
Pattern
The re-indexed pattern. Equal to ``self`` if ``copy`` is ``False``.
"""
if mapping is None:
# Suggested in issue #519
mapping = self.node_mapping()
func: Callable[[Node], Node] = (
(lambda node: mapping.get(node, node)) if isinstance(mapping, Mapping) else mapping
)
new_pattern = Pattern(input_nodes=map(func, self.input_nodes))
for cmd in self:
new_pattern.add(cmd.reindex(func))
new_pattern.reorder_output_nodes(map(func, self.output_nodes))
if copy:
return new_pattern
self.__input_nodes = new_pattern.__input_nodes
self.__seq = new_pattern.__seq
self.__output_nodes = new_pattern.__output_nodes
return self
[docs]
def compose(
self, other: Pattern, mapping: Mapping[int, int], preserve_mapping: bool = False
) -> tuple[Pattern, dict[int, int]]:
r"""Compose two patterns by merging subsets of outputs from ``self`` and a subset of inputs of ``other``, and relabeling the nodes of ``other`` that were not merged.
Parameters
----------
other : Pattern
Pattern to be composed with ``self``.
mapping: Mapping[int, int]
Partial relabelling of the nodes in ``other``, with ``keys`` and ``values`` denoting the old and new node labels, respectively.
preserve_mapping: bool
Boolean flag controlling the ordering of the output nodes in the returned pattern.
Returns
-------
p: Pattern
composed pattern
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:`(I_j, O_j, V_j, S_j)` the ordered set of inputs and outputs, the computational space and the sequence of commands of pattern :math:`P_j`, respectively, with :math:`j = 1` for pattern ``self`` and :math:`j = 2` for pattern ``other``. Let's denote :math:`P` the resulting pattern with :math:`(I, O, V, S)`.
Let's denote :math:`K, U` the sets of ``keys`` and ``values`` of ``mapping``, :math:`M_1 = O_1 \cap U` the set of merged outputs, and :math:`M_2 = \{k \in I_2 \cap K | k \rightarrow v, v \in M_1 \}` the set of merged inputs.
The pattern composition requires that
- :math:`K \subseteq V_2`.
- For a pair :math:`(k, v) \in (K, U)`
- :math:`U \cap V_1 \setminus O_1 = \emptyset`. If :math:`v \in O_1`, then :math:`k \in I_2`, otherwise an error is raised.
- :math:`v` can always satisfy :math:`v \notin V_1`, thereby allowing a custom relabelling.
The returned pattern follows this convention:
- Nodes of pattern ``other`` not specified in ``mapping`` (i.e., :math:`V_2 \cap K^c`) are relabelled in ascending order.
- The sequence of the resulting pattern is :math:`S = S_2 S_1`, where nodes in :math:`S_2` are relabelled according to ``mapping``.
- :math:`I = I_1 \cup (I_2 \setminus M_2)`.
- :math:`O = (O_1 \setminus M_1) \cup O_2`.
- Input (and, respectively, output) nodes in the returned pattern have the order of the pattern ``self`` followed by those of the pattern ``other``. Merged nodes are removed.
- If ``preserve_mapping = True`` and :math:`|M_1| = |I_2| = |O_2|`, then the outputs of the returned pattern are the outputs of pattern ``self``, where the nth merged output is replaced by the output of pattern ``other`` corresponding to its nth input instead.
"""
nodes_p1 = self.nodes() # Results contain preprocessed Pauli nodes
nodes_p2 = other.nodes()
if not mapping.keys() <= nodes_p2:
raise PatternError("Keys of `mapping` must correspond to the nodes of `other`.")
# Cast to set for improved performance in membership test
mapping_values_set = set(mapping.values())
o1_set = set(self.__output_nodes)
i2_set = set(other.input_nodes)
if len(mapping) != len(mapping_values_set):
raise PatternError("Values of `mapping` contain duplicates.")
if mapping_values_set & nodes_p1 - o1_set:
raise PatternError("Values of `mapping` must not contain measured nodes of pattern `self`.")
for k, v in mapping.items():
if v in o1_set and k not in i2_set:
raise PatternError(
f"Mapping {k} -> {v} is not valid. {v} is an output of pattern `self` but {k} is not an input of pattern `other`."
)
# Check if resulting pattern will have C commands before E commands
if any(cmd.kind == CommandKind.C for cmd in self.__seq) and any(cmd.kind == CommandKind.E for cmd in other):
warnings.warn(
r"Pattern `self` contains Clifford commands and pattern `other` contains E commands. Standardization might not be possible for the resulting composed pattern.",
stacklevel=2,
)
shift = max((*nodes_p1, *mapping.values())) + 1
mapping_complete = other.node_mapping(mapping, start=shift)
other = other.reindex(mapping_complete, copy=True)
merged = mapping_values_set.intersection(self.__output_nodes)
inputs = self.__input_nodes + [n for n in other.input_nodes if n not in merged]
if preserve_mapping and not (len(merged) == len(other.input_nodes) == len(other.output_nodes)):
warnings.warn(
"`preserve_mapping = True` ignored because the number of merged nodes, inputs, and outputs of pattern `other` are different.",
stacklevel=2,
)
preserve_mapping = False
if preserve_mapping:
io_mapping = dict(zip(other.input_nodes, other.output_nodes, strict=True))
outputs = [io_mapping[n] if n in merged else n for n in self.__output_nodes]
else:
outputs = [n for n in self.__output_nodes if n not in merged] + other.output_nodes
seq = [*self.__seq, *other]
p = Pattern(input_nodes=inputs, output_nodes=outputs, cmds=seq)
return p, mapping_complete
@property
def input_nodes(self) -> list[int]:
"""List input nodes."""
return list(self.__input_nodes) # copy for preventing modification
@property
def output_nodes(self) -> list[int]:
"""List all nodes that are either ``input_nodes`` or prepared with ``N`` commands and that have not been measured with an ``M`` command."""
return list(self.__output_nodes) # copy for preventing modification
def __len__(self) -> int:
"""Return the length of command sequence."""
return len(self.__seq)
def __iter__(self) -> Iterator[CommandType]:
"""Iterate over commands."""
return iter(self.__seq)
def __getitem__(self, index: int) -> CommandType:
"""Get the command at a given index."""
return self.__seq[index]
@property
def n_node(self) -> int:
"""Count of nodes that are either `input_nodes` or prepared with `N` commands."""
return self.__n_node
[docs]
def reorder_output_nodes(self, output_nodes: Iterable[int]) -> None:
"""Arrange the order of output_nodes.
Parameters
----------
output_nodes: iterable of int
output nodes order determined by user. each index corresponds to that of logical qubits.
"""
output_nodes = list(output_nodes) # make our own copy (allow iterators to be passed)
assert_permutation(self.__output_nodes, output_nodes)
self.__output_nodes = output_nodes
def __repr__(self) -> str:
"""Return a representation string of the pattern."""
arguments = []
if self.__input_nodes:
arguments.append(f"input_nodes={self.__input_nodes}")
if self.__seq:
arguments.append(f"cmds={self.__seq}")
if self.__output_nodes:
arguments.append(f"output_nodes={self.__output_nodes}")
return f"Pattern({', '.join(arguments)})"
def __str__(self) -> str:
"""Return a human-readable string of the pattern."""
return pattern_to_str(self)
def __eq__(self, other: object) -> bool:
"""Return `True` if the two patterns are equal, `False` otherwise."""
if not isinstance(other, Pattern):
return NotImplemented
return (
self.__seq == other.__seq
and self.__input_nodes == other.__input_nodes
and self.__output_nodes == other.__output_nodes
)
[docs]
def to_ascii(
self, left_to_right: bool = False, limit: int | None = 40, target: Container[command.CommandKind] | None = None
) -> str:
"""Return the ASCII string representation of the pattern.
Parameters
----------
left_to_right: bool, optional
If ``True``, the first command will appear at the beginning of
the resulting string. If ``False`` (the default), the first command will
appear at the end of the string.
limit: int | None, optional
If set to an int (default: 40), only first ``limit`` commands are printed,
and an ellipsis is added at the end to indicate that some commands have been elided.
If ``limit=None``, there is no limit on the number of printed commands.
target: Container[command.CommandKind], optional
If set, only commands of kinds specified in ``target`` are printed.
Examples
--------
>>> from graphix import Circuit
>>> circuit = Circuit(1)
>>> circuit.h(0)
>>> pattern = circuit.transpile().pattern
>>> pattern.to_ascii()
'X(1,{0}) M(0,0) E(0,1) N(1)'
>>> pattern.to_ascii(left_to_right=True)
'N(1) E(0,1) M(0,0) X(1,{0})'
>>> from graphix.command import CommandKind
>>> pattern.to_ascii(target={CommandKind.M})
'M(0,0)'
>>> pattern.to_ascii(target={CommandKind.X, CommandKind.Z})
'X(1,{0})'
>>> circuit = Circuit(3)
>>> circuit.ccx(0, 1, 2)
>>> pattern = circuit.transpile().pattern
>>> pattern.to_ascii(limit=5)
'X(24,{21}) Z(24,{20}) X(32,{31}) Z(32,{28})...(98 more commands)'
>>> pattern.to_ascii(limit=5, left_to_right=True)
'N(3) E(2,3) M(2,0) N(4)...(98 more commands)'
>>> pattern.to_ascii(limit=None)
'X(24,{21}) Z(24,{20}) X(32,{31}) Z(32,{28}) X(30,...E(4,1) [M(3,0)]{2} E(3,4) N(4) M(2,0) E(2,3) N(3)'
"""
return pattern_to_str(self, OutputFormat.ASCII, left_to_right, limit, target)
[docs]
def to_latex(
self, left_to_right: bool = False, limit: int | None = 40, target: Container[command.CommandKind] | None = None
) -> str:
r"""Return a string containing the LaTeX representation of the pattern.
Parameters
----------
left_to_right: bool, optional
If ``True``, the first command will appear at the beginning of
the resulting string. If ``False`` (the default), the first command will
appear at the end of the string.
limit: int | None, optional
If set to an int (default: 40), only first ``limit`` commands are printed,
and an ellipsis is added at the end to indicate that some commands have been elided.
If ``limit=None``, there is no limit on the number of printed commands.
target: Container[command.CommandKind], optional
If set, only commands of kinds specified in ``target`` are printed.
Examples
--------
>>> from graphix import Circuit
>>> circuit = Circuit(1)
>>> circuit.h(0)
>>> pattern = circuit.transpile().pattern
>>> pattern.to_latex()
'\\(X_{1}^{0}\\,M_{0}^{0}\\,E_{0,1}\\,N_{1}\\)'
>>> pattern.to_latex(left_to_right=True)
'\\(N_{1}\\,E_{0,1}\\,M_{0}^{0}\\,X_{1}^{0}\\)'
>>> from graphix.command import CommandKind
>>> pattern.to_latex(target={CommandKind.M})
'\\(M_{0}^{0}\\)'
>>> pattern.to_latex(target={CommandKind.X, CommandKind.Z})
'\\(X_{1}^{0}\\)'
"""
return pattern_to_str(self, OutputFormat.LaTeX, left_to_right, limit, target)
[docs]
def to_unicode(
self, left_to_right: bool = False, limit: int | None = 40, target: Container[command.CommandKind] | None = None
) -> str:
"""Return the Unicode string representation of the pattern.
Parameters
----------
left_to_right: bool, optional
If ``True``, the first command will appear at the beginning of
the resulting string. If ``False`` (the default), the first command will
appear at the end of the string.
limit: int | None, optional
If set to an int (default: 40), only first ``limit`` commands are printed,
and an ellipsis is added at the end to indicate that some commands have been elided.
If ``limit=None``, there is no limit on the number of printed commands.
target: Container[command.CommandKind], optional
If set, only commands of kinds specified in ``target`` are printed.
Examples
--------
>>> from graphix import Circuit
>>> circuit = Circuit(1)
>>> circuit.h(0)
>>> pattern = circuit.transpile().pattern
>>> pattern.to_unicode()
'X₁⁰ M₀(0) E₀₋₁ N₁'
>>> pattern.to_unicode(left_to_right=True)
'N₁ E₀₋₁ M₀(0) X₁⁰'
>>> from graphix.command import CommandKind
>>> pattern.to_unicode(target={CommandKind.M})
'M₀(0)'
>>> pattern.to_unicode(target={CommandKind.X, CommandKind.Z})
'X₁⁰'
"""
return pattern_to_str(self, OutputFormat.Unicode, left_to_right, limit, target)
[docs]
def standardize(self) -> None:
"""Execute standardization of the pattern.
'standard' pattern is one where commands are sorted in the
order of 'N', 'E', 'M' and then byproduct commands ('X' and
'Z') and finally Clifford commands ('C').
"""
self.__seq = optimization.standardize(self).__seq
[docs]
def is_standard(self, strict: bool = False) -> bool:
"""Determine whether the command sequence is standard.
Parameters
----------
strict : bool, optional
If True, ensures that C commands are the last ones.
Returns
-------
is_standard : bool
True if the pattern is standard
"""
it = iter(self)
try:
kind = next(it).kind
while kind == CommandKind.N:
kind = next(it).kind
while kind == CommandKind.E:
kind = next(it).kind
while kind == CommandKind.M:
kind = next(it).kind
if strict:
xz = {CommandKind.X, CommandKind.Z}
while kind in xz:
kind = next(it).kind
while kind == CommandKind.C:
kind = next(it).kind
else:
xzc = {CommandKind.X, CommandKind.Z, CommandKind.C}
while kind in xzc:
kind = next(it).kind
except StopIteration:
return True
else:
return False
[docs]
def shift_signals(self, method: str = "direct") -> dict[int, set[int]]:
"""Perform signal shifting procedure.
Extract the t-dependence of the measurement into 'S' commands
and commute them to the end of the command sequence where it can be removed.
This procedure simplifies the dependence structure of the pattern.
Ref for the original 'mc' method:
V. Danos, E. Kashefi and P. Panangaden. J. ACM 54.2 8 (2007)
Parameters
----------
method : str, optional
'direct' shift_signals is executed on a conventional Pattern sequence.
'mc' shift_signals is done using the original algorithm in the measurement calculus paper.
Returns
-------
signal_dict : dict[int, set[int]]
For each node, the signal that have been shifted.
"""
# Shifting signals could turn non-runnable patterns into
# runnable ones, so we check runnability first to avoid hiding
# code-logic errors.
# For example, the non-runnable pattern {1}[M(0)] N(0) would
# become M(0) N(0), which is runnable.
self.check_runnability()
match method:
case "direct":
return self.shift_signals_direct()
case "mc":
signal_dict = self.signals()
target = self._find_op_to_be_moved(CommandKind.S, rev=True)
while target is not None:
if target == len(self.__seq) - 1:
self.__seq.pop(target)
target = self._find_op_to_be_moved(CommandKind.S, rev=True)
continue
cmd = self.__seq[target + 1]
kind = cmd.kind
match kind:
case CommandKind.X:
self._commute_xs(target)
case CommandKind.Z:
self._commute_zs(target)
case CommandKind.M:
self._commute_ms(target)
case CommandKind.S:
self._commute_ss(target)
case _:
self._commute_with_following(target)
target += 1
return signal_dict
case _:
raise PatternError("Invalid method")
[docs]
def shift_signals_direct(self) -> dict[int, set[int]]:
"""Perform signal shifting procedure."""
signal_dict: dict[int, set[int]] = {}
def expand_domain(domain: set[command.Node]) -> None:
"""Expand ``domain`` with previously shifted signals.
Parameters
----------
domain : set[int]
Set of nodes representing the current domain. This set is
modified in place by XORing any previously shifted domains.
"""
for node in domain & signal_dict.keys():
domain ^= signal_dict[node]
for i, cmd in enumerate(self):
match cmd.kind:
case CommandKind.M:
s_domain = set(cmd.s_domain)
t_domain = set(cmd.t_domain)
expand_domain(s_domain)
expand_domain(t_domain)
plane = cmd.measurement.to_bloch().plane
match plane:
case Plane.XY:
# M^{XY,α} X^s Z^t = M^{XY,(-1)^s·α+tπ}
# = S^t M^{XY,(-1)^s·α}
# = S^t M^{XY,α} X^s
if t_domain:
signal_dict[cmd.node] = t_domain
t_domain = set()
case Plane.XZ:
# M^{XZ,α} X^s Z^t = M^{XZ,(-1)^t((-1)^s·α+sπ)}
# = M^{XZ,(-1)^{s+t}·α+(-1)^t·sπ}
# = M^{XZ,(-1)^{s+t}·α+sπ} (since (-1)^t·π ≡ π (mod 2π))
# = S^s M^{XZ,(-1)^{s+t}·α}
# = S^s M^{XZ,α} Z^{s+t}
if s_domain:
signal_dict[cmd.node] = s_domain
t_domain ^= s_domain
s_domain = set()
case Plane.YZ if s_domain:
# M^{YZ,α} X^s Z^t = M^{YZ,(-1)^t·α+sπ)}
# = S^s M^{YZ,(-1)^t·α}
# = S^s M^{YZ,α} Z^t
signal_dict[cmd.node] = s_domain
s_domain = set()
if s_domain != cmd.s_domain or t_domain != cmd.t_domain:
self.__seq[i] = dataclasses.replace(cmd, s_domain=s_domain, t_domain=t_domain)
case CommandKind.X | CommandKind.Z:
domain = set(cmd.domain)
expand_domain(domain)
if domain != cmd.domain:
self.__seq[i] = dataclasses.replace(cmd, domain=domain)
return signal_dict
def _find_op_to_be_moved(self, op: CommandKind, rev: bool = False, skipnum: int = 0) -> int | None:
"""Find a command.
Parameters
----------
op : CommandKind, N, E, M, X, Z, S
command types to be searched
rev : bool
search from the end (true) or start (false) of seq
skipnum : int
skip the detected command by specified times
"""
if not rev: # Search from the start
start_index, end_index, step = 0, len(self.__seq), 1
else: # Search from the end
start_index, end_index, step = len(self.__seq) - 1, -1, -1
num_ops = 0
for index in range(start_index, end_index, step):
if self.__seq[index].kind == op:
num_ops += 1
if num_ops == skipnum + 1:
return index
# If no target found
return None
def _commute_ex(self, target: int) -> bool:
"""Perform the commutation of E and X.
Parameters
----------
target : int
target command index. this must point to
a X command followed by E command
"""
x = self.__seq[target]
e = self.__seq[target + 1]
assert x.kind == CommandKind.X
assert e.kind == CommandKind.E
if e.nodes[0] == x.node:
z = command.Z(node=e.nodes[1], domain=x.domain)
self.__seq.pop(target + 1) # del E
self.__seq.insert(target, z) # add Z in front of X
self.__seq.insert(target, e) # add E in front of Z
return True
if e.nodes[1] == x.node:
z = command.Z(node=e.nodes[0], domain=x.domain)
self.__seq.pop(target + 1) # del E
self.__seq.insert(target, z) # add Z in front of X
self.__seq.insert(target, e) # add E in front of Z
return True
self._commute_with_following(target)
return False
def _commute_mx(self, target: int) -> bool:
"""Perform the commutation of M and X.
Parameters
----------
target : int
target command index. this must point to
a X command followed by M command
"""
x = self.__seq[target]
m = self.__seq[target + 1]
assert x.kind == CommandKind.X
assert m.kind == CommandKind.M
if x.node == m.node:
m.s_domain ^= x.domain
self.__seq.pop(target) # del X
return True
self._commute_with_following(target)
return False
def _commute_mz(self, target: int) -> bool:
"""Perform the commutation of M and Z.
Parameters
----------
target : int
target command index. this must point to
a Z command followed by M command
"""
z = self.__seq[target]
m = self.__seq[target + 1]
assert z.kind == CommandKind.Z
assert m.kind == CommandKind.M
if z.node == m.node:
m.t_domain ^= z.domain
self.__seq.pop(target) # del Z
return True
self._commute_with_following(target)
return False
def _commute_xs(self, target: int) -> None:
"""Perform the commutation of X and S.
Parameters
----------
target : int
target command index. this must point to
a S command followed by X command
"""
s = self.__seq[target]
x = self.__seq[target + 1]
assert s.kind == CommandKind.S
assert x.kind == CommandKind.X
if s.node in x.domain:
x.domain ^= s.domain
self._commute_with_following(target)
def _commute_zs(self, target: int) -> None:
"""Perform the commutation of Z and S.
Parameters
----------
target : int
target command index. this must point to
a S command followed by Z command
"""
s = self.__seq[target]
z = self.__seq[target + 1]
assert s.kind == CommandKind.S
assert z.kind == CommandKind.Z
if s.node in z.domain:
z.domain ^= s.domain
self._commute_with_following(target)
def _commute_ms(self, target: int) -> None:
"""Perform the commutation of M and S.
Parameters
----------
target : int
target command index. this must point to
a S command followed by M command
"""
s = self.__seq[target]
m = self.__seq[target + 1]
assert s.kind == CommandKind.S
assert m.kind == CommandKind.M
if s.node in m.s_domain:
m.s_domain ^= s.domain
if s.node in m.t_domain:
m.t_domain ^= s.domain
self._commute_with_following(target)
def _commute_ss(self, target: int) -> None:
"""Perform the commutation of two S commands.
Parameters
----------
target : int
target command index. this must point to
a S command followed by S command
"""
s1 = self.__seq[target]
s2 = self.__seq[target + 1]
assert s1.kind == CommandKind.S
assert s2.kind == CommandKind.S
if s1.node in s2.domain:
s2.domain ^= s1.domain
self._commute_with_following(target)
def _commute_with_following(self, target: int) -> None:
"""Perform the commutation of two consecutive commands that commutes.
commutes the target command with the following command.
Parameters
----------
target : int
target command index
"""
a = self.__seq[target + 1]
self.__seq.pop(target + 1)
self.__seq.insert(target, a)
def _commute_with_preceding(self, target: int) -> None:
"""Perform the commutation of two consecutive commands that commutes.
commutes the target command with the preceding command.
Parameters
----------
target : int
target command index
"""
a = self.__seq[target - 1]
self.__seq.pop(target - 1)
self.__seq.insert(target, a)
def _move_n_to_left(self) -> None:
"""Move all 'N' commands to the start of the sequence.
N can be moved to the start of sequence without the need of considering
commutation relations.
"""
new_seq = []
n_list = []
for cmd in self.__seq:
if cmd.kind == CommandKind.N:
n_list.append(cmd)
else:
new_seq.append(cmd)
n_list.sort(key=lambda n_cmd: n_cmd.node)
self.__seq = n_list + new_seq
def _move_byproduct_to_right(self) -> None:
"""Move the byproduct commands to the end of sequence, using the commutation relations implemented in graphix.Pattern class."""
# First, we move all X commands to the end of sequence
index = len(self.__seq) - 1
x_limit = len(self.__seq) - 1
while index > 0:
if self.__seq[index].kind == CommandKind.X:
index_x = index
while index_x < x_limit:
cmd = self.__seq[index_x + 1]
kind = cmd.kind
match kind:
case CommandKind.E:
move = self._commute_ex(index_x)
if move:
x_limit += 1 # addition of extra Z means target must be increased
index_x += 1
case CommandKind.M:
search = self._commute_mx(index_x)
if search:
x_limit -= 1 # XM commutation rule removes X command
break
case _:
self._commute_with_following(index_x)
index_x += 1
else:
x_limit -= 1
index -= 1
# then, move Z to the end of sequence in front of X
index = x_limit
z_limit = x_limit
while index > 0:
if self.__seq[index].kind == CommandKind.Z:
index_z = index
while index_z < z_limit:
cmd = self.__seq[index_z + 1]
if cmd.kind == CommandKind.M:
search = self._commute_mz(index_z)
if search:
z_limit -= 1 # ZM commutation rule removes Z command
break
else:
self._commute_with_following(index_z)
index_z += 1
index -= 1
def _move_e_after_n(self) -> None:
"""Move all E commands to the start of sequence, before all N commands. assumes that _move_n_to_left() method was called."""
moved_e = 0
target = self._find_op_to_be_moved(CommandKind.E, skipnum=moved_e)
while target is not None:
if (target == 0) or (
self.__seq[target - 1].kind == CommandKind.N or self.__seq[target - 1].kind == CommandKind.E
):
moved_e += 1
target = self._find_op_to_be_moved(CommandKind.E, skipnum=moved_e)
continue
self._commute_with_preceding(target)
target -= 1
[docs]
def signals(self) -> dict[int, set[int]]:
"""Extract 't' domain of measurement commands, turn them into signal 'S' commands and add to the command sequence.
This is used for shift_signals() method.
"""
signal_dict = {}
pos = 0
while pos < len(self.__seq):
cmd = self.__seq[pos]
if cmd.kind == CommandKind.M:
extracted_signal = extract_signal(cmd.measurement.to_bloch().plane, cmd.s_domain, cmd.t_domain)
if extracted_signal.signal:
self.__seq.insert(pos + 1, command.S(node=cmd.node, domain=extracted_signal.signal))
cmd.s_domain = extracted_signal.s_domain
cmd.t_domain = extracted_signal.t_domain
pos += 1
signal_dict[cmd.node] = extracted_signal.signal
pos += 1
return signal_dict
[docs]
def partial_order_layers(self) -> tuple[frozenset[int], ...]:
"""Extract the measurement order of the pattern in the form of layers.
This method standardizes the pattern, builds a directed acyclical graph (DAG) from measurement and correction domains, and then performs a topological sort.
Returns
-------
tuple[frozenset[int], ...]
Measurement partial order between the pattern's nodes in a layer form.
Raises
------
RunnabilityError
If the pattern is not runnable.
Notes
-----
- This function wraps :func:`optimization.StandardizedPattern.partial_order_layers`, and the returned object is described in the notes of this method.
"""
return self.to_standardizedpattern().partial_order_layers()
[docs]
def to_causalflow(self) -> CausalFlow[BlochMeasurement]:
r"""Extract the causal flow structure from the current measurement pattern.
This method does not call the flow-extraction routine on the underlying open graph, but constructs the flow from the pattern corrections instead.
Returns
-------
CausalFlow[Measurement]
The causal flow associated with the current pattern.
Raises
------
FlowError
If the pattern:
- Contains measurements in forbidden planes (XZ or YZ),
- Is empty, or
- Induces a correction function and a partial order which fail the well-formedness checks for a valid causal flow.
ValueError
If `N` commands in the pattern do not represent a :math:`|+\rangle` state or if the pattern corrections form closed loops.
Notes
-----
- Applying the chain ``Pattern.to_causalflow().to_xzcorrections().to_pattern()`` to a strongly deterministic pattern returns a new pattern implementing the same unitary transformation. This equivalence holds as long as the original pattern contains no Clifford commands, since those are discarded during open-graph extraction.
- This method requires that all the measurements in the pattern are represented as Bloch measurements (i.e., there are no :class:`PauliMeasurement`s). Use :meth:`to_bloch()` to convert all Pauli measurements.
"""
return self.to_xzcorrections().downcast_bloch().to_causalflow()
[docs]
def to_gflow(self) -> GFlow[BlochMeasurement]:
r"""Extract the generalized flow (gflow) structure from the current measurement pattern.
This method does not call the flow-extraction routine on the underlying open graph, but constructs the gflow from the pattern corrections instead.
Returns
-------
GFlow[Measurement]
The gflow associated with the current pattern.
Raises
------
FlowError
If the pattern is empty or if the extracted structure does not satisfy
the well-formedness conditions required for a valid gflow.
ValueError
If `N` commands in the pattern do not represent a :math:`|+\rangle` state or if the pattern corrections form closed loops.
Notes
-----
The notes provided in :func:`self.to_causalflow` apply here as well.
"""
return self.to_xzcorrections().downcast_bloch().to_gflow()
[docs]
def to_pauliflow(self) -> PauliFlow[Measurement]:
r"""Extract the Pauli flow structure from the current measurement pattern.
This method does not call the flow-extraction routine on the underlying open graph, but
reconstructs the Pauli flow from the pattern corrections instead (see
:meth:`graphix.flow.core.XZCorrections.to_pauliflow`). Contrary to
:meth:`to_causalflow` and :meth:`to_gflow`, Pauli measurements are kept as
axes rather than downcast to planar Bloch measurements, so that the Pauli-basis
structure is preserved.
Returns
-------
PauliFlow[Measurement]
The Pauli flow associated with the current pattern.
Raises
------
FlowError
If the pattern is empty or if no Pauli flow is compatible with the pattern corrections.
ValueError
If `N` commands in the pattern do not represent a :math:`|+\rangle` state or if the
pattern corrections form closed loops.
Notes
-----
The notes provided in :func:`self.to_causalflow` apply here as well.
"""
return self.to_xzcorrections().to_pauliflow()
[docs]
def to_xzcorrections(self) -> XZCorrections[Measurement]:
r"""Extract the XZ-corrections from the current measurement pattern.
Returns
-------
XZCorrections[Measurement]
The XZ-corrections associated with the current pattern.
Raises
------
XZCorrectionsError
If the extracted correction dictionaries are not well formed.
ValueError
If ``N`` commands in the pattern do not represent a :math:`\ket{+}` state or if the pattern corrections form closed loops.
Notes
-----
To ensure that applying the chain ``Pattern.to_xzcorrections().to_pattern()`` to a strongly deterministic pattern returns a new pattern implementing the same unitary transformation, XZ-corrections must be extracted from a standardized pattern. This requirement arises for the same reason that flow extraction also operates correctly on standardized patterns only.
This equivalence holds as long as the original pattern contains no Clifford commands, since those are discarded during open-graph extraction.
See docstring in :func:`optimization.StandardizedPattern.to_gflow` for additional information.
"""
return self.to_standardizedpattern().to_xzcorrections()
def _measurement_order_depth(self) -> list[int]:
"""Obtain a measurement order which reduces the depth of a pattern.
Returns
-------
list[int]
optimal measurement order for parallel computing
"""
partial_order_layers = self.partial_order_layers()
return list(itertools.chain(*reversed(partial_order_layers[1:])))
[docs]
def sort_measurement_commands(self, meas_order: list[int]) -> list[command.M]:
"""Convert measurement order to sequence of measurement commands.
Parameters
----------
meas_order: list of int
optimal measurement order.
Returns
-------
meas_cmds: list of command
sorted measurement commands
"""
meas_dict = self.measurement_commands()
return [meas_dict[i] for i in meas_order]
[docs]
def measurement_commands(self) -> dict[int, command.M]:
"""Return a dictionary mapping nodes to measurement commands.
Returns
-------
meas_dict : dict[int, command.M]
measurement commands indexed by nodes
"""
return {cmd.node: cmd for cmd in self if cmd.kind == CommandKind.M}
[docs]
def max_degree(self) -> int:
"""Get max degree of a pattern.
Returns
-------
max_degree : int
max degree of a pattern
"""
graph = self.to_opengraph().graph
degree = graph.degree()
assert isinstance(degree, nx.classes.reportviews.DiDegreeView)
degrees = dict(degree).values()
if len(degrees) == 0:
return 0
return int(max(degrees))
[docs]
def nodes(self) -> set[int]:
"""Return the set of nodes of the pattern."""
nodes = set(self.input_nodes)
for cmd in self.__seq:
if cmd.kind == CommandKind.N:
nodes.add(cmd.node)
return nodes
[docs]
def isolated_nodes(self) -> set[int]:
"""Return the set of isolated nodes in the pattern.
Returns
-------
set[int]
Set of the isolated nodes
"""
graph = self.to_opengraph().graph
return {node for node, d in graph.degree if d == 0}
[docs]
def to_opengraph(self) -> OpenGraph[Measurement]:
r"""Extract the underlying resource-state open graph from the pattern.
This method standardizes the pattern first to guarantee that
Clifford commands are properly encoded in the resulting open graph.
Specifically, Cliffords acting on measured nodes are absorbed into measurements,
while Cliffords acting on output nodes are stored in ``OpenGraph.output_cliffords``.
Returns
-------
OpenGraph[Measurement]
"""
return self.to_standardizedpattern().to_opengraph()
[docs]
def clifford_commands(self) -> dict[int, Clifford]:
"""Extract Clifford commands.
Returns
-------
vops : dict
"""
return {cmd.node: cmd.clifford for cmd in self.__seq if cmd.kind == CommandKind.C}
[docs]
def connected_nodes(self, node: int, prepared: set[int] | None = None) -> list[int]:
"""Find nodes that are connected to a specified node.
These nodes must be in the statevector when the specified
node is measured, to ensure correct computation.
If connected nodes already exist in the statevector (prepared),
then they will be ignored as they do not need to be prepared again.
Parameters
----------
node : int
node index
prepared : list
list of node indices, which are to be ignored
Returns
-------
node_list : list
list of nodes that are entangled with specified node
"""
if not self.is_standard():
self.standardize()
if prepared is None:
prepared = set()
node_list = []
ind = self._find_op_to_be_moved(CommandKind.E)
if ind is not None: # end -> 'node' is isolated
cmd = self.__seq[ind]
while cmd.kind == CommandKind.E:
if cmd.nodes[0] == node:
if cmd.nodes[1] not in prepared:
node_list.append(cmd.nodes[1])
elif cmd.nodes[1] == node and cmd.nodes[0] not in prepared:
node_list.append(cmd.nodes[0])
ind += 1
cmd = self.__seq[ind]
return node_list
[docs]
def correction_commands(self) -> list[command.X | command.Z]:
"""Return the list of byproduct correction commands."""
assert self.is_standard()
cmds = []
for cmd in self:
match cmd.kind:
case CommandKind.X | CommandKind.Z:
cmds.append(cmd)
return cmds
[docs]
def parallelize_pattern(self) -> None:
"""Optimize the pattern to reduce the depth of the computation by gathering measurement commands that can be performed simultaneously.
This optimized pattern runs efficiently on GPUs and quantum hardwares with
depth (e.g. coherence time) limitations.
"""
if not self.is_standard():
self.standardize()
meas_order = self._measurement_order_depth()
self._reorder_pattern(self.sort_measurement_commands(meas_order))
[docs]
def minimize_space(
self, heuristics: Iterable[SpaceMinimizationHeuristic] | None = None, *, copy: bool = False
) -> Pattern:
"""Return a pattern that reduces the maximal space, i.e. the number of qubits simultaneously required to execute the pattern.
See :func:`graphix.space_minimization.minimize_space` for more
information about the default heuristics.
To obtain a space-optimal pattern while preserving the
measurement order, use ``minimize_space([])``. This commutes
the ``N`` and ``E`` commands so that the pattern is
space-optimal among patterns with the same measurement order
(see :meth:`StandardizedPattern.to_space_optimal_pattern`).
Parameters
----------
heuristics : Iterable[~graphix.space_minimization.SpaceMinimizationHeuristic] | None = None
The heuristics to try in order.
By default, :const:`~graphix.space_minimization.DEFAULT_HEURISTICS` is used.
copy : bool, optional
If ``True``, the current pattern remains unchanged and a
new pattern is returned. The default is ``False``, meaning
that changes are performed in place.
Returns
-------
Pattern
The optimized pattern. Equal to ``self`` if ``copy`` is ``False``.
"""
new = self.to_standardizedpattern().minimize_space(heuristics).to_space_optimal_pattern()
if copy:
return new
self.__seq = new.__seq
return self
def _reorder_pattern(self, meas_commands: list[command.M]) -> None:
"""Reorder the command sequence.
Parameters
----------
meas_commands : list of command
list of measurement ('M') commands
"""
new = dataclasses.replace(self.to_standardizedpattern(), m_list=tuple(meas_commands)).to_space_optimal_pattern()
self.__seq = new.__seq
[docs]
def max_space(self) -> int:
"""Compute the maximum number of nodes that must be present in the graph (graph space) during the execution of the pattern.
For statevector simulation, this is equivalent to the maximum memory
needed for classical simulation.
Returns
-------
n_nodes : int
max number of nodes present in the graph during pattern execution.
"""
return pattern_max_space(self)
[docs]
def space_list(self) -> list[int]:
"""Return the list of the number of nodes present in the graph (space) during each step of execution of the pattern (for N and M commands).
Returns
-------
N_list : list
time evolution of 'space' at each 'N' and 'M' commands of pattern.
"""
nodes = 0
n_list = []
for cmd in self.__seq:
match cmd.kind:
case CommandKind.N:
nodes += 1
n_list.append(nodes)
case CommandKind.M:
nodes -= 1
n_list.append(nodes)
return n_list
[docs]
@override
def to_pattern(self) -> Pattern:
"Return self."
return self
def _warn_non_inferred_pauli_measurements(self, stacklevel: int) -> None:
for cmd in self:
if (
cmd.kind == CommandKind.M
and isinstance(cmd.measurement, BlochMeasurement)
and cmd.measurement.to_pauli_or_none() is not None
):
warnings.warn("Pattern with non-inferred Pauli measurements.", stacklevel=stacklevel + 1)
return
[docs]
def draw(
self,
*,
annotations: DrawPatternAnnotations | None = DrawPatternAnnotations.Flow,
flow_from_pattern: bool = True,
stacklevel: int = 1,
**options: Unpack[DrawKwargs],
) -> None:
"""Visualize the underlying graph of the pattern with annotations.
Parameters
----------
annotations : DrawPatternAnnotations | None, default=DrawPatternAnnotations.Flow
Annotations to be shown.
- ``DrawPatternAnnotations.Flow`` (default): show the pattern's flow if it exists.
- ``DrawPatternAnnotations.XZCorrections``: show the pattern's XZ-corrections.
- ``None``: show the underlying open graph only.
flow_from_pattern : bool, default=True
If ``True``, the command sequence of the pattern is used to derive a flow structure. If ``False``, only the underlying open graph is used.
options : Unpack[DrawKwargs]
Options controlling graph visualization. See :class:`VisualizationOptions`.
stacklevel : int, optional
Stack level to use for warnings. Defaults to 1, meaning that warnings
are reported at this function's call site.
Raises
------
PatternError
If the underlying open graph does not have flow.
Notes
-----
If ``flow_from_pattern==True`` but the pattern is not compatible with a causal flow, gflow, or Pauli flow, an attempt to extract the flow from the underlying open graph will be made while warning the user.
"""
if annotations is None:
og = self.to_opengraph()
gv = GraphVisualizer.from_opengraph(og, **options)
else:
xzcorrections = self.to_xzcorrections()
match annotations:
case DrawPatternAnnotations.Flow:
flow: PauliFlow[Measurement] | None = None
if flow_from_pattern:
try:
# We first try to extract a causal flow because the plotting
# algorithm for causal flows yields nicer plots.
flow = xzcorrections.downcast_bloch().to_causalflow()
except (TypeError, FlowError):
try:
# If the pattern is not consistent with a Pauli flow,
# it won't be consistent with a gflow.
flow = xzcorrections.to_pauliflow()
except FlowError:
warn(
"The pattern is not consistent with a flow. An attempt to be extract the flow from the underlying open graph will be made.",
stacklevel=stacklevel,
)
if flow is None:
og = self.to_opengraph()
try:
bloch_case = og.downcast_bloch()
except TypeError:
pass
else:
flow = bloch_case.to_causalflow_or_none()
if flow is None:
flow = og.to_pauliflow_or_none(stacklevel=stacklevel + 1)
if flow is None:
raise PatternError(
"The pattern's open graph does not have Pauli flow. Consider setting the `annotations` parameter to `None` or `DrawPatternAnnotations.XZCorrections`."
)
gv = GraphVisualizer.from_flow(flow, **options)
case DrawPatternAnnotations.XZCorrections:
gv = GraphVisualizer.from_xzcorrections(xzcorrections, **options)
gv.visualize()
[docs]
def to_qasm3(self, filename: Path | str, input_state: dict[int, State] | State = BasicStates.PLUS) -> None:
"""Export measurement pattern to OpenQASM 3.0 file.
See :func:`graphix.qasm3_exporter.pattern_to_qasm3`.
Parameters
----------
filename : Path | str
File name to export to. Example: ``"filename.qasm"``.
input_state : dict[int, State] | State, default BasicStates.PLUS
The initial state for each input node. Only ``|0⟩`` or ``|+⟩`` states are supported.
"""
with Path(filename).with_suffix(".qasm").open("w", encoding="utf-8") as file:
file.writelines(pattern_to_qasm3_lines(self, input_state=input_state))
[docs]
def is_parameterized(self) -> bool:
"""
Return ``True`` if there is at least one measurement angle that is not just an instance of :class:`SupportsFloat`.
A parameterized pattern is a pattern where at least one
measurement angle is an expression that is not a number,
typically an instance of ``sympy.Expr`` (but we don't force to
choose ``sympy`` here).
"""
return any(
not isinstance(cmd.measurement.angle, SupportsFloat)
for cmd in self
if cmd.kind == command.CommandKind.M and isinstance(cmd.measurement, BlochMeasurement)
)
[docs]
@override
def replace_parameter(
self, variable: Parameter, substitute: ExpressionOrSupportsFloat, *, copy: bool = False
) -> Pattern:
return self.apply(lambda m: m.with_parameter(variable, substitute), copy=copy)
[docs]
@override
def replace_parameters(
self, assignment: Mapping[Parameter, ExpressionOrSupportsFloat], *, copy: bool = False
) -> Pattern:
return self.apply(lambda m: m.with_parameters(assignment), copy=copy)
[docs]
def copy(self) -> Pattern:
"""Return a copy of the pattern."""
result = self.__new__(self.__class__)
result.__seq = [copy.copy(cmd) for cmd in self.__seq]
result.__input_nodes = self.__input_nodes.copy()
result.__output_nodes = self.__output_nodes.copy()
result.__n_node = self.__n_node
return result
[docs]
def check_runnability(self) -> None:
"""Check whether the pattern is runnable.
Raises :class:`RunnabilityError` exception if it is not.
Notes
-----
The runnability check can only guarantee the runnability of
MBQC+LC patterns. Patterns that make use of custom :class:``BaseN``
and :class:``BaseM`` commands can have additional runnability
constraints that are not checked by this method. For instance,
in the Veriphix implementation of VBQC, blind measurements
have hidden domains that cannot be checked.
"""
active = set(self.input_nodes)
measured = set()
def check_active(cmd: CommandType, node: int) -> None:
if node in measured:
raise RunnabilityError(cmd, node, RunnabilityErrorReason.AlreadyMeasured)
if node not in active:
raise RunnabilityError(cmd, node, RunnabilityErrorReason.NotYetActive)
def check_measured(cmd: CommandType, node: int) -> None:
if node not in measured:
raise RunnabilityError(cmd, node, RunnabilityErrorReason.NotYetMeasured)
for cmd in self:
match cmd.kind:
case CommandKind.N:
if cmd.node in active:
raise RunnabilityError(cmd, cmd.node, RunnabilityErrorReason.AlreadyActive)
if cmd.node in measured:
raise RunnabilityError(cmd, cmd.node, RunnabilityErrorReason.AlreadyMeasured)
active.add(cmd.node)
case CommandKind.E:
n0, n1 = cmd.nodes
check_active(cmd, n0)
check_active(cmd, n1)
case CommandKind.M:
check_active(cmd, cmd.node)
if isinstance(cmd, command.M):
# `cmd.s_domain` and `cmd.t_domain` are only
# defined if the command is an actual `M` command,
# which may not be the case if the method is
# called with a pattern constructed with another
# implementation of `BaseM` (for instance, a blind
# pattern from Veriphix).
for domain in cmd.s_domain, cmd.t_domain:
if cmd.node in domain:
raise RunnabilityError(cmd, cmd.node, RunnabilityErrorReason.DomainSelfLoop)
for node in domain:
check_measured(cmd, node)
active.remove(cmd.node)
measured.add(cmd.node)
case CommandKind.X | CommandKind.Z:
check_active(cmd, cmd.node)
for node in cmd.domain:
check_measured(cmd, node)
case CommandKind.C:
check_active(cmd, cmd.node)
[docs]
def apply(self, f: Callable[[Measurement], Measurement], *, copy: bool = False) -> Pattern:
"""Apply the function ``f`` to each measurement.
Parameters
----------
f: Callable[[Measurement], Measurement]
Function applied to each measurement.
copy : bool, optional
If ``True``, the current pattern remains unchanged and a
new pattern is returned. The default is ``False``, meaning
that changes are performed in place.
Returns
-------
Pattern
The resulting pattern.
If ``copy`` is ``False``, the result is ``self``.
Example
-------
>>> from graphix import Pattern, command
>>> from graphix.measurements import BlochMeasurement, Measurement
>>> pattern = Pattern(input_nodes=[0], cmds=[command.M(0, Measurement.XZ(0.25))])
>>> pattern.apply(lambda m: BlochMeasurement(m.angle + 1, m.plane))
Pattern(input_nodes=[0], cmds=[M(0, Measurement.XZ(1.25))])
>>> pattern
Pattern(input_nodes=[0], cmds=[M(0, Measurement.XZ(1.25))])
"""
new_seq = [cmd.map(f) if cmd.kind == CommandKind.M else cmd for cmd in self]
if copy:
new_pattern = Pattern(input_nodes=self.input_nodes, cmds=new_seq)
new_pattern.reorder_output_nodes(self.output_nodes)
return new_pattern
self.__seq = new_seq
return self
[docs]
def map(self, f: Callable[[Measurement], Measurement]) -> Pattern:
"""Return a pattern where the function ``f`` has been applied to each measurement.
Equivalent to ``self.apply(f, copy=True)``.
Parameters
----------
f: Callable[[Measurement], Measurement]
Function applied to each measurement.
Returns
-------
Pattern
The resulting pattern.
Example
-------
>>> from graphix import Pattern, command
>>> from graphix.measurements import BlochMeasurement, Measurement
>>> pattern = Pattern(input_nodes=[0], cmds=[command.M(0, Measurement.XZ(0.25))])
>>> pattern.map(lambda m: BlochMeasurement(m.angle + 1, m.plane))
Pattern(input_nodes=[0], cmds=[M(0, Measurement.XZ(1.25))])
>>> pattern
Pattern(input_nodes=[0], cmds=[M(0, Measurement.XZ(0.25))])
"""
return self.apply(f, copy=True)
[docs]
def infer_pauli_measurements(self, *, rel_tol: float = 1e-09, abs_tol: float = 0.0, copy: bool = False) -> Pattern:
"""Replace Bloch measurements close to a Pauli measurement by Pauli measurements.
Parameters
----------
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``.
copy : bool, optional
If ``True``, the current pattern remains unchanged and a
new pattern is returned. The default is ``False``, meaning
that changes are performed in place.
Returns
-------
Pattern
A pattern in which Bloch measurements close to a Pauli
measurement are replaced by Pauli measurements. If
``copy`` is ``False``, the result is ``self``.
Example
-------
>>> from graphix import Pattern, command
>>> from graphix.measurements import BlochMeasurement, Measurement
>>> pattern = Pattern(input_nodes=[0], cmds=[command.M(0, Measurement.XY(0.5))])
>>> pattern.infer_pauli_measurements()
Pattern(input_nodes=[0], cmds=[M(0, Measurement.Y)])
"""
return self.apply(lambda m: m.to_pauli_or_bloch(rel_tol, abs_tol), copy=copy)
[docs]
def blochify(self, *, copy: bool = False) -> Pattern:
"""Represent all measurements as Bloch measurements.
Parameters
----------
copy : bool, optional
If ``True``, the current pattern remains unchanged and a
new pattern is returned. The default is ``False``, meaning
that changes are performed in place.
Returns
-------
Pattern
A pattern in which all measurements are Bloch measurements.
If ``copy`` is ``False``, the result is ``self``.
Example
-------
>>> from graphix import Pattern, command
>>> from graphix.measurements import BlochMeasurement, Measurement
>>> pattern = Pattern(input_nodes=[0], cmds=[command.M(0, Measurement.Y)])
>>> pattern.blochify()
Pattern(input_nodes=[0], cmds=[M(0, Measurement.XY(0.5))])
>>> pattern
Pattern(input_nodes=[0], cmds=[M(0, Measurement.XY(0.5))])
"""
return self.apply(lambda m: m.to_bloch(), copy=copy)
[docs]
def to_bloch(self) -> Pattern:
"""Return an equivalent pattern in which all measurements are represented as Bloch measurements.
Equivalent to ``self.blochify(copy=True)``.
Example
-------
>>> from graphix import Pattern, command
>>> from graphix.measurements import BlochMeasurement, Measurement
>>> pattern = Pattern(input_nodes=[0], cmds=[command.M(0, Measurement.Y)])
>>> pattern.to_bloch()
Pattern(input_nodes=[0], cmds=[M(0, Measurement.XY(0.5))])
>>> pattern
Pattern(input_nodes=[0], cmds=[M(0, Measurement.Y)])
"""
return self.blochify(copy=True)
[docs]
def remove_pauli_measurements(
self, *, copy: bool = False, standardize: bool = False, stacklevel: int = 1
) -> Pattern:
"""Remove non-input Pauli measurements from the given pattern.
See :func:`~remove_pauli_measurements.remove_pauli_measurements` for more information.
Parameters
----------
copy : bool, optional
If ``True``, the current pattern remains unchanged and a
new pattern is returned. The default is ``False``, meaning
that changes are performed in place.
standardize: bool, optional
If ``True``, the pattern is returned in standardized form.
The default is ``False``: the nodes are prepared on a
need-by-need basis, minimizing space usage.
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
The pattern in which Pauli measurements have been moved
before the other measurements. If ``copy`` is ``False``,
the result is ``self``.
"""
standardized_pattern = self.to_standardizedpattern().remove_pauli_measurements(stacklevel=stacklevel + 1)
pattern = standardized_pattern.to_pattern() if standardize else standardized_pattern.to_space_optimal_pattern()
if copy:
return pattern
self.__seq = pattern.__seq
self.__output_nodes = pattern.__output_nodes
self.__n_node = pattern.n_node
return self
[docs]
def remove_local_clifford_commands(self, *, copy: bool = False) -> Pattern:
"""Return an equivalent pattern where local Clifford commands have been replaced by MBQC commands.
See :func:`~optimization.remove_local_clifford_commands`
for more information.
Parameters
----------
copy : bool, optional
If ``True``, the current pattern remains unchanged and a
new pattern is returned. The default is ``False``, meaning
that changes are performed in place.
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
The pattern in which local Clifford commands have been replaced
by MBQC commands. If ``copy`` is ``False``, the result is
``self``.
"""
new_pattern = optimization.remove_local_clifford_commands(self)
if copy:
return new_pattern
self.__seq = new_pattern.__seq
self.__output_nodes = new_pattern.__output_nodes
return self
[docs]
class PatternError(Exception):
"""Exception subclass to handle pattern errors."""
[docs]
class RunnabilityErrorReason(Enum):
"""Describe the reason for a pattern not being runnable."""
AlreadyActive = enum.auto()
"""A node is prepared whereas it has already been prepared or it is an input node."""
AlreadyMeasured = enum.auto()
"""A node is measured for a second time."""
NotYetActive = enum.auto()
"""A node is entangled, measured or corrected whereas it has not been prepared yet and it is not an input node."""
NotYetMeasured = enum.auto()
"""A node appears in the domain of a measurement of a correction whereas it has not been measured yet."""
DomainSelfLoop = enum.auto()
"""A node appears in the domain of its own measurement. This is a particular case of `NotYetMeasured`, introduced to make the error message clearer."""
[docs]
@dataclass
class RunnabilityError(PatternError):
"""Error raised by :method:`Pattern.check_runnability`."""
cmd: CommandType
node: int
reason: RunnabilityErrorReason
def __str__(self) -> str:
"""Explain the error."""
match self.reason:
case RunnabilityErrorReason.AlreadyActive:
return f"{self.cmd}: node {self.node} is already active."
case RunnabilityErrorReason.AlreadyMeasured:
return f"{self.cmd}: node {self.node} is already measured."
case RunnabilityErrorReason.NotYetActive:
return f"{self.cmd}: node {self.node} is not yet active."
case RunnabilityErrorReason.NotYetMeasured:
return f"{self.cmd}: node {self.node} is not yet measured."
case RunnabilityErrorReason.DomainSelfLoop:
return f"{self.cmd}: node {self.node} appears in the domain of its own measurement command."
case _:
assert_never(self.reason)
[docs]
def assert_permutation(original: list[int], user: list[int]) -> None:
"""Check that the provided `user` node list is a permutation from `original`."""
node_set = set(user)
if node_set != set(original):
raise PatternError(f"{node_set} != {set(original)}")
for node in user:
if node in node_set:
node_set.remove(node)
else:
raise PatternError(f"{node} appears twice")
[docs]
def shift_outcomes(outcomes: Mapping[int, Outcome], signal_dict: Mapping[int, AbstractSet[int]]) -> dict[int, Outcome]:
"""Update outcomes with shifted signals.
Shifted signals (as returned by the method
:func:`Pattern.shift_signals`) affect classical outputs
(measurements) while leaving the quantum state invariant.
This method updates the given ``outcomes`` by swapping the
measurements affected by signals. This can be used either to
transform the results into measurements observed in the unshifted
pattern, or vice versa.
Parameters
----------
outcomes : Mapping[int, Outcome]
Classical outputs.
signal_dict : Mapping[int, AbstractSet[int]]
For each node, the signal that has been shifted
(as returned by :func:`Pattern.shift_signals`).
Returns
-------
shifted_outcomes : dict[int, int]
Classical outputs updated with shifted signals.
"""
return {
node: toggle_outcome(outcome) if sum(outcomes[i] for i in signal_dict.get(node, [])) % 2 == 1 else outcome
for node, outcome in outcomes.items()
}
def _reverse_mapping(m: Mapping[K, V]) -> dict[V, list[K]]:
"""
Build a reverse mapping of ``m``.
For each key-value pair ``k → v`` in the input mapping ``m``, the
returned dictionary contains an entry ``v → [k1, k2, …]`` where the
list holds all keys that were associated with ``v`` in ``m``.
The order of keys inside each list corresponds to their first
appearance while iterating over ``m``.
Parameters
----------
m: Mapping[K, V]
The mapping to be inspected.
Returns
-------
dict[V, list[K]]
A new dictionary where each distinct value from ``m`` maps to a list
of all keys that originally pointed to that value.
"""
result: dict[V, list[K]] = {}
for k, v in m.items():
if (antecedents := result.get(v)) is not None:
antecedents.append(k)
else:
result[v] = [k]
return result
def _duplicates_in_mapping(m: Mapping[K, V]) -> dict[V, list[K]]:
"""
Return a dictionary of values that appear more than once in ``m``.
For each value that is mapped to by multiple keys, the resulting
dictionary contains an entry ``value → [key1, key2, …]`` where the list
holds all keys that map to that value (the order follows the
iteration order of ``m.items()``).
Parameters
----------
m : Mapping[K, V]
The mapping to be inspected.
Returns
-------
dict[V, list[K]]
A dictionary whose keys are the duplicated values and whose values are
lists of the keys that map to them. If the input mapping is injective,
an empty dictionary is returned.
"""
reversed_mapping = _reverse_mapping(m)
return {v: antecedents for v, antecedents in reversed_mapping.items() if len(antecedents) > 1}