diff --git a/samplomatic/builders/box_builder.py b/samplomatic/builders/box_builder.py index 6559257e..2c8835bc 100644 --- a/samplomatic/builders/box_builder.py +++ b/samplomatic/builders/box_builder.py @@ -15,13 +15,14 @@ from __future__ import annotations import numpy as np -from qiskit.circuit import Barrier +from qiskit.circuit import Barrier, IfElseOp from ..aliases import CircuitInstruction, ParamIndices from ..exceptions import BuildError from ..partition import QubitPartition from ..pre_samplex import PreSamplex from .builder import Builder +from .dynamic_builder import BoxLeftIfElseBuilder, BoxRightIfElseBuilder from .specs import CollectionSpec, EmissionSpec, InstructionMode, VirtualType from .template_state import TemplateState @@ -34,12 +35,14 @@ def __init__(self, collection: CollectionSpec, emission: EmissionSpec): self.collection = collection self.emission = emission - self.measured_qubits = QubitPartition(1, []) self.entangled_qubits = set() + self.measured_qubits = QubitPartition(1, []) def _append_dressed_layer(self) -> ParamIndices: """Add a dressed layer.""" - qubits = self.collection.qubits + qubits = self.collection.collect_qubits + if len(qubits) == 0: + return try: remapped_qubits = [ list(map(lambda k: self.template_state.qubit_map[k], subsys)) for subsys in qubits @@ -68,12 +71,21 @@ def _append_dressed_layer(self) -> ParamIndices: return param_idxs.reshape(len(qubits), -1) - def _append_barrier(self, label: str): + def _append_barrier(self, label: str, qubits=None): label = f"{label}{'_'.join(map(str, self.template_state.scope_idx))}" - all_qubits = self.template_state.qubit_map.values() + all_qubits = ( + self.template_state.qubit_map.values() + if qubits is None + else [v for k, v in self.template_state.qubit_map.items() if (k,) in qubits] + ) barrier = CircuitInstruction(Barrier(len(all_qubits), label), all_qubits) self.template_state.template.append(barrier) + def _validate_if_else(self, if_else: IfElseOp): + self.samplex_state.verify_no_twirled_clbits( + self.template_state.get_condition_clbits(if_else.condition) + ) + class LeftBoxBuilder(BoxBuilder): """Box builder for left dressings.""" @@ -89,6 +101,22 @@ def parse(self, instr: CircuitInstruction): self.template_state.append_remapped_gate(instr) return + if name.startswith("if_else"): + self._validate_if_else(instr.operation) + builder = BoxLeftIfElseBuilder( + instr, self.samplex_state, self.collection.synth, self.template_state.param_iter + ) + if_else = builder.build() + new_qubits = [self.template_state.qubit_map.get(qubit, qubit) for qubit in instr.qubits] + self.template_state.template.append(if_else, new_qubits, instr.clbits) + return + + elif not self.collection.dynamic_qubits.all_elements.isdisjoint(instr.qubits): + raise RuntimeError( + "Cannot handle a dynamic instruction and another instruction on " + f"qubits {instr.qubits} in the same dressed box." + ) + if name.startswith("meas"): for qubit in instr.qubits: if (qubit,) not in self.measured_qubits: @@ -127,8 +155,9 @@ def parse(self, instr: CircuitInstruction): "left-dressed box." ) self.entangled_qubits.update(instr.qubits) - params = self.template_state.append_remapped_gate(instr) mode = InstructionMode.PROPAGATE + params = self.template_state.append_remapped_gate(instr) + else: raise BuildError(f"Instruction {instr} could not be parsed.") @@ -137,8 +166,9 @@ def parse(self, instr: CircuitInstruction): def lhs(self): self._append_barrier("L") param_idxs = self._append_dressed_layer() - self.samplex_state.add_collect(self.collection.qubits, self.collection.synth, param_idxs) - self._append_barrier("M") + collect_qubits = self.collection.collect_qubits + self.samplex_state.add_collect(collect_qubits, self.collection.synth, param_idxs) + self._append_barrier("M", collect_qubits) def rhs(self): self._append_barrier("R") @@ -172,19 +202,37 @@ def __init__(self, collection: CollectionSpec, emission: EmissionSpec): def parse(self, instr: CircuitInstruction): if (name := instr.operation.name).startswith("barrier"): - params = self.template_state.append_remapped_gate(instr) + self.template_state.append_remapped_gate(instr) return if name.startswith("meas"): raise BuildError("Measurements are not currently supported in right-dressed boxes.") - elif (num_qubits := instr.operation.num_qubits) == 1: + if not self.collection.dynamic_qubits.all_elements.isdisjoint(instr.qubits): + raise BuildError( + "Cannot handle a dynamic instruction and another instruction on " + f"qubits {instr.qubits} in the same dressed box." + ) + + if name.startswith("if_else"): + for q in instr.qubits: + self.collection.dynamic_qubits.add((q,)) + self._validate_if_else(instr.operation) + builder = BoxRightIfElseBuilder( + instr, self.samplex_state, self.collection.synth, self.template_state.param_iter + ) + if_else = builder.build() + new_qubits = [self.template_state.qubit_map.get(qubit, qubit) for qubit in instr.qubits] + self.template_state.template.append(if_else, new_qubits, instr.clbits) + return + + if (num_qubits := instr.operation.num_qubits) == 1: self.entangled_qubits.update(instr.qubits) # the action of this single-qubit gate will be absorbed into the dressing + mode = InstructionMode.MULTIPLY params = [] if instr.operation.is_parameterized(): params.extend((None, param) for param in instr.operation.params) - mode = InstructionMode.MULTIPLY elif num_qubits > 1: if not self.entangled_qubits.isdisjoint(instr.qubits): @@ -216,7 +264,8 @@ def lhs(self): ) def rhs(self): - self._append_barrier("M") + collect_qubits = self.collection.collect_qubits + self._append_barrier("M", collect_qubits) param_idxs = self._append_dressed_layer() - self.samplex_state.add_collect(self.collection.qubits, self.collection.synth, param_idxs) + self.samplex_state.add_collect(collect_qubits, self.collection.synth, param_idxs) self._append_barrier("R") diff --git a/samplomatic/builders/dynamic_builder.py b/samplomatic/builders/dynamic_builder.py new file mode 100644 index 00000000..ae39918b --- /dev/null +++ b/samplomatic/builders/dynamic_builder.py @@ -0,0 +1,213 @@ +# This code is a Qiskit project. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""DynamicBuilder""" + +import abc +from copy import deepcopy +from typing import Generic, TypeVar + +import numpy as np +from qiskit.circuit import IfElseOp, QuantumCircuit + +from ..aliases import CircuitInstruction, ParamIndices, Qubit, QubitIndex +from ..exceptions import BuildError +from ..partition import QubitPartition +from ..pre_samplex import DanglerMatch, PreSamplex +from ..pre_samplex.graph_data import Direction, PreCollect, PreCopy, PreEdge, PreEmit, PrePropagate +from ..synths import Synth +from .param_iter import ParamIter +from .specs import InstructionMode + +T = TypeVar("T") + + +class DynamicBuilder(abc.ABC, Generic[T]): + """Base class for building dressed conditional operations. + + This class does not inherit from :class:`~.Builder` as it does not add the operations to the + template. Instead, it constructs an operation of the same type while adding nodes to the + corresponding samplex. + + Args: + op: The control flow operation to build. + pre_samplex: The pre-samplex to use. + synth: The synthesizer to use for the dressing. + param_iter: An iterator over parameters to use in the circuit being built. + """ + + def __init__( + self, + op: T, + pre_samplex: PreSamplex, + synth: Synth, + param_iter: ParamIter, + ): + self.op = op + self.pre_samplex = pre_samplex + self.synth = synth + self.param_iter = param_iter + + def _block_qubit_map(self, block) -> dict[Qubit, QubitIndex]: + block_map = {i_q: b_q for i_q, b_q in zip(self.op.qubits, block.qubits)} + qubit_map = self.pre_samplex.qubit_map + return {block_map[i_q]: qubit_map[i_q] for i_q in qubit_map if i_q in block_map} + + def _parse_mq_gate(self, instr: CircuitInstruction, new_block: QuantumCircuit): + new_params = [] + param_mapping = [] + for param in instr.operation.params: + param_mapping.append([self.param_iter.idx, param]) + new_params.append(next(self.param_iter)) + + new_op = type(instr.operation)(*new_params) if new_params else instr.operation + new_block.append(new_op, instr.qubits, instr.clbits) + self.pre_samplex.add_propagate(instr, mode=InstructionMode.PROPAGATE, params=param_mapping) + + def _parse_sq_gate(self, instr: CircuitInstruction): + new_params = [] + if instr.operation.is_parameterized(): + new_params.extend((None, param) for param in instr.operation.params) + self.pre_samplex.add_propagate(instr, mode=InstructionMode.MULTIPLY, params=new_params) + + def _append_dressed_layer(self, new_block: QuantumCircuit) -> ParamIndices: + start = self.param_iter.idx + num_params = len(new_block.qubits) * self.synth.num_params + params = np.arange(start, start + num_params, dtype=np.intp) + for qubit in new_block.qubits: + for instr in self.synth.make_template([qubit], self.param_iter): + new_block.append(instr) + return params.reshape(len(new_block.qubits), -1) + + @abc.abstractmethod + def build_block(self) -> QuantumCircuit: + """Build a block of a control flow operation.""" + + @abc.abstractmethod + def build(self) -> T: + """Build the operation.""" + + +class BoxLeftIfElseBuilder(DynamicBuilder[IfElseOp]): + def build_block(self, block) -> QuantumCircuit: + block = ( + block + if block is not None + else QuantumCircuit( + self.op.operation.params[0].qubits, self.op.operation.params[0].clbits + ) + ) + + new_block = QuantumCircuit(block.qubits, block.clbits) + params = self._append_dressed_layer(new_block) + + pre_samplex = self.pre_samplex.remap(qubit_map=self._block_qubit_map(block)) + qubits = QubitPartition.from_elements(new_block.qubits) + subsystems = pre_samplex.qubits_to_indices(qubits) + dangler_match = DanglerMatch(Direction.LEFT, node_types=(PreCollect, PrePropagate)) + + list(pre_samplex.find_then_remove_danglers(dangler_match, subsystems)) + pre_samplex.add_collect(qubits, self.synth, params) + + entangled_qubits = set() + for instr in block: + if len(instr.qubits) == 1: + if not entangled_qubits.isdisjoint(instr.qubits): + raise BuildError( + "Cannot have entanglers before single-qubit gates in a left-dressed box." + ) + self._parse_sq_gate(instr) + else: + entangled_qubits.update(instr.qubits) + self._parse_mq_gate(instr, new_block) + + copy_idxs = [] + for node_idx, partition in pre_samplex.find_then_remove_danglers(dangler_match, subsystems): + copy_idx = pre_samplex.graph.add_node(PreCopy(partition, Direction.LEFT)) + copy_idxs.append((copy_idx, partition)) + edge = PreEdge(partition, Direction.LEFT) + pre_samplex.graph.add_edge(copy_idx, node_idx, edge) + + return new_block, copy_idxs + + def build(self): + original_danglers = deepcopy(self.pre_samplex.get_all_danglers()) + if_block, if_danglers = self.build_block(self.op.params[0]) + self.pre_samplex.set_all_danglers(*original_danglers) + else_block, else_danglers = self.build_block(self.op.params[1]) + + for node_idx, subsystems in [*if_danglers, *else_danglers]: + self.pre_samplex.add_dangler( + subsystems.all_elements, + node_idx, + ) + + return IfElseOp(self.op.operation.condition, if_block, else_block, self.op.label) + + +class BoxRightIfElseBuilder(DynamicBuilder[IfElseOp]): + def build_block(self, block) -> QuantumCircuit: + block = ( + block + if block is not None + else QuantumCircuit( + self.op.operation.params[0].qubits, self.op.operation.params[0].clbits + ) + ) + + new_block = QuantumCircuit(block.qubits, block.clbits) + pre_samplex = self.pre_samplex.remap(qubit_map=self._block_qubit_map(block)) + + qubits = QubitPartition.from_elements(new_block.qubits) + subsystems = pre_samplex.qubits_to_indices(qubits) + dangler_match = DanglerMatch(Direction.RIGHT, node_types=(PreEmit, PrePropagate)) + + new_danglers = [] + for node_idx, partition in pre_samplex.find_then_remove_danglers(dangler_match, subsystems): + copy_idx = pre_samplex.graph.add_node(PreCopy(partition, Direction.RIGHT)) + edge = PreEdge(partition, Direction.RIGHT) + pre_samplex.graph.add_edge(node_idx, copy_idx, edge) + new_danglers.append((copy_idx, partition)) + + for node_idx, partition in new_danglers: + pre_samplex.add_dangler(partition.all_elements, node_idx) + + unentangled_qubits = set() + for instr in block: + if len(instr.qubits) == 1: + unentangled_qubits.update(instr.qubits) + self._parse_sq_gate(instr) + else: + if not unentangled_qubits.isdisjoint(instr.qubits): + raise BuildError( + "Cannot have entanglers after single-qubit gates in a right-dressed box." + ) + self._parse_mq_gate(instr, new_block) + + params = self._append_dressed_layer(new_block) + collect_idx = pre_samplex.add_collect(qubits, self.synth, params) + + return new_block, (collect_idx, subsystems) + + def build(self): + original_danglers = deepcopy(self.pre_samplex.get_all_danglers()) + if_block, if_dangler = self.build_block(self.op.params[0]) + self.pre_samplex.set_all_danglers(*original_danglers) + else_block, else_dangler = self.build_block(self.op.params[1]) + + for node_idx, subsystems in [if_dangler, else_dangler]: + self.pre_samplex.add_dangler( + subsystems.all_elements, + node_idx, + ) + + return IfElseOp(self.op.operation.condition, if_block, else_block, self.op.label) diff --git a/samplomatic/builders/get_builder.py b/samplomatic/builders/get_builder.py index 04746b55..20bae936 100644 --- a/samplomatic/builders/get_builder.py +++ b/samplomatic/builders/get_builder.py @@ -72,7 +72,13 @@ def get_builder(instr: CircuitInstruction | None, qubits: Sequence[Qubit]) -> Bu if emission.noise_ref and not emission.twirl_register_type: raise BuildError(f"Cannot get a builder for {annotations}. Inject noise requires twirling.") + collection.dynamic_qubits = QubitPartition(1) if collection.dressing is DressingMode.LEFT: + # TODO: if BoxOp contained a DAG we could look at the first topological generation + for box_instr in instr.operation.body: + if box_instr.operation.name.startswith("if_else"): + for q in box_instr.qubits: + collection.dynamic_qubits.add((q,)) return LeftBoxBuilder(collection, emission) return RightBoxBuilder(collection, emission) diff --git a/samplomatic/builders/specs.py b/samplomatic/builders/specs.py index 87b33e7d..b7184c82 100644 --- a/samplomatic/builders/specs.py +++ b/samplomatic/builders/specs.py @@ -93,3 +93,13 @@ class CollectionSpec: synth: Synth[Qubit, Parameter, CircuitInstruction] | None = None """How to synthesize collection gates.""" + + dynamic_qubits: QubitPartition | None = None + """The subset of 'qubits' collected in conditional operations.""" + + @property + def collect_qubits(self): + """The subset of 'qubits' collected in the box boundary.""" + if self.dynamic_qubits is None: + return self.qubits + return self.qubits.difference(self.dynamic_qubits.all_elements) diff --git a/samplomatic/pre_samplex/graph_data.py b/samplomatic/pre_samplex/graph_data.py index bdbe2e92..862f4e12 100644 --- a/samplomatic/pre_samplex/graph_data.py +++ b/samplomatic/pre_samplex/graph_data.py @@ -44,6 +44,11 @@ def get_style(self) -> NodeStyle: return NodeStyle(title=type(self).__name__).append_data("Subsystems", list(self.subsystems)) +@dataclass +class PreCopy(PreNode): + """The copy node type used during building.""" + + @dataclass class PreEdge: """Edge data on a samplex builder's graph.""" @@ -54,9 +59,6 @@ class PreEdge: direction: Direction """Whether the edge is moving forwards or backwards in circuit-time.""" - force_register_copy: bool = False - """Whether the edge should force the receiving node to get a copy of the register.""" - def get_style(self) -> EdgeStyle: """Summarizes the style of this node when plotted via :func:`~.plot_graph`.""" return ( @@ -65,7 +67,6 @@ def get_style(self) -> EdgeStyle: ) .append_data("Subsystems", list(self.subsystems)) .append_data("Direction", self.direction.name) - .append_data("Force copy", self.force_register_copy) ) def add_subsystems(self, new_subsystems: QubitIndicesPartition): diff --git a/samplomatic/pre_samplex/pre_samplex.py b/samplomatic/pre_samplex/pre_samplex.py index e33ed66f..8d9089a1 100644 --- a/samplomatic/pre_samplex/pre_samplex.py +++ b/samplomatic/pre_samplex/pre_samplex.py @@ -62,6 +62,7 @@ CollectTemplateValues, CollectZ2ToOutputNode, CombineRegistersNode, + CopyNode, InjectNoiseNode, LeftMultiplicationNode, LeftU2ParametricMultiplicationNode, @@ -84,6 +85,7 @@ from .graph_data import ( PreChangeBasis, PreCollect, + PreCopy, PreEdge, PreEmit, PreInjectNoise, @@ -165,10 +167,6 @@ class PreSamplex: circuit. passthrough_params: List of :class:`~.ParamSpec` for parameters which exist in the template but are not generated in the collectors. - forced_copy_node_idxs: List of node indices for which copying of registers will be forced. - The nodes behave differently for edges with left/right direction. For left direction, - the incoming node is checked against `forced_copy_node_idxs`, while for right direction, - the outgoing node is. """ def __init__( @@ -184,7 +182,6 @@ def __init__( basis_changes: dict[str, int] | None = None, twirled_clbits: set[ClbitIndex] | None = None, passthrough_params: ParamSpec | None = None, - forced_copy_node_idxs: set[NodeIndex] | None = None, ): self.graph = PyDiGraph[PreNode, PreEdge](multigraph=True) if graph is None else graph self.qubit_map: dict[Qubit, QubitIndex] = {} if qubit_map is None else qubit_map @@ -205,9 +202,6 @@ def __init__( self.passthrough_params: ParamSpec = ( [] if passthrough_params is None else passthrough_params ) - self._forced_copy_node_idxs: set[NodeIndex] = ( - set() if forced_copy_node_idxs is None else forced_copy_node_idxs - ) def remap(self, qubit_map: dict[Qubit, QubitIndex]) -> PreSamplex: """Remap the object to a new :class:`~.PreSamplex` object. @@ -230,7 +224,6 @@ def remap(self, qubit_map: dict[Qubit, QubitIndex]) -> PreSamplex: self._basis_changes, self._twirled_clbits, self.passthrough_params, - self._forced_copy_node_idxs, ) def find_danglers( @@ -356,7 +349,7 @@ def enforce_no_propagation(self, instr: CircuitInstruction): # in the future when we have multi-qubit virtual groups, this can't be hard-coded to 1 subsystems = QubitIndicesPartition(1, [(self.qubit_map[qubit],) for qubit in instr.qubits]) - match = DanglerMatch(node_types=(PreEmit, PrePropagate), direction=Direction.RIGHT) + match = DanglerMatch(node_types=(PreCopy, PreEmit, PrePropagate), direction=Direction.RIGHT) if any(True for _ in self.find_danglers(match, subsystems)): raise SamplexBuildError(f"Cannot propagate through {instr.operation.name} instruction.") match = DanglerMatch(direction=Direction.LEFT) @@ -393,14 +386,6 @@ def qubits_to_indices(self, qubits: QubitPartition) -> QubitIndicesPartition: qubits.num_elements_per_part, [tuple(self.qubit_map[q] for q in sys) for sys in qubits] ) - def add_force_copy_nodes(self, node_idxs: Iterable[NodeIndex]): - """Add node indices for which a register will be forced to copy.""" - self._forced_copy_node_idxs.update(node_idxs) - - def remove_force_copy_nodes(self, node_idxs: Iterable[NodeIndex]): - """Remove node indices for which a register will be forced to copy.""" - self._forced_copy_node_idxs.difference_update(node_idxs) - def add_collect( self, qubits: QubitPartition, @@ -434,19 +419,12 @@ def add_collect( ) # collect any nodes that need collecting and unmark them as dangling - match = DanglerMatch(node_types=(PreEmit, PrePropagate), direction=Direction.RIGHT) + match = DanglerMatch(node_types=(PreCopy, PreEmit, PrePropagate), direction=Direction.RIGHT) for found_idx, found_subsystems in self.find_then_remove_danglers(match, subsystems): if self.graph.has_edge(found_idx, node_idx): - # The force_register_copy stays the same and doesn't need update. self.graph.get_edge_data(found_idx, node_idx).add_subsystems(found_subsystems) else: - self.graph.add_edge( - found_idx, - node_idx, - PreEdge( - found_subsystems, Direction.RIGHT, found_idx in self._forced_copy_node_idxs - ), - ) + self.graph.add_edge(found_idx, node_idx, PreEdge(found_subsystems, Direction.RIGHT)) # prevent dangling pre-propagate left nodes from catching any further action because # this collection is in the way @@ -500,7 +478,7 @@ def add_z2_collect(self, qubits: QubitPartition, clbit_idxs: Sequence[ClbitIndex # Collect relevant nodes which are an optional dangler, and leave them optionally dangling # This needs to happen first, so the new optional danglers created later won't be counted. match = DanglerMatch( - node_types=(PreEmit, PrePropagate), + node_types=(PreCopy, PreEmit, PrePropagate), direction=Direction.RIGHT, dangler_type=DanglerType.OPTIONAL, ) @@ -512,7 +490,7 @@ def add_z2_collect(self, qubits: QubitPartition, clbit_idxs: Sequence[ClbitIndex # Collect every relevant node which is a required dangler, # then convert it to an optional dangler. match = DanglerMatch( - node_types=(PreEmit, PrePropagate), + node_types=(PreCopy, PreEmit, PrePropagate), direction=Direction.RIGHT, dangler_type=DanglerType.REQUIRED, ) @@ -558,7 +536,9 @@ def add_emit_twirl(self, qubits: QubitPartition, register_type: VirtualType) -> # connect this emmision there. we do NOT want to remove them as dangling because they # might need to be used again for future emissions, for example, a Pauli twirl followed by # a noise injection. - match = DanglerMatch(node_types=(PreCollect, PrePropagate), direction=Direction.LEFT) + match = DanglerMatch( + node_types=(PreCollect, PreCopy, PrePropagate), direction=Direction.LEFT + ) aggregate_found_subsystems = set() for found_idx, found_subsystems in self.find_danglers(match, subsystems): @@ -566,7 +546,7 @@ def add_emit_twirl(self, qubits: QubitPartition, register_type: VirtualType) -> self.graph.add_edge( node_idx, found_idx, - PreEdge(found_subsystems, Direction.LEFT, found_idx in self._forced_copy_node_idxs), + PreEdge(found_subsystems, Direction.LEFT), ) if aggregate_found_subsystems != set(subsystems): @@ -580,22 +560,24 @@ def add_emit_twirl(self, qubits: QubitPartition, register_type: VirtualType) -> return node_idx - def _add_emit_left(self, node: PreEmit): + def _add_left(self, node: PreNode): """Add a pre-emit with `Direction.LEFT`. This method adds edges to any node that is dangling with overlapping subsystems. """ node_idx = self.graph.add_node(node) - match = DanglerMatch(node_types=(PreCollect, PrePropagate), direction=Direction.LEFT) + match = DanglerMatch( + node_types=(PreCopy, PreCollect, PrePropagate), direction=Direction.LEFT + ) for found_idx, found_subsystems in self.find_danglers(match, node.subsystems): self.graph.add_edge(node_idx, found_idx, PreEdge(found_subsystems, Direction.LEFT)) return node_idx - def _add_emit_right(self, node: PreEmit): + def _add_right(self, node: PreNode): """Add a pre-emit with `Direction.RIGHT`. - This method sets the pre-emit as dangling. + This method sets the node as dangling. """ node_idx = self.graph.add_node(node) self.add_dangler(node.subsystems.all_elements, node_idx) @@ -639,7 +621,7 @@ def add_emit_noise_left( modifier_ref, next(self._pauli_lindblad_map_count), ) - return self._add_emit_left(node) + return self._add_left(node) def add_emit_noise_right( self, qubits: QubitPartition, noise_ref: StrRef, modifier_ref: StrRef = "" @@ -678,7 +660,7 @@ def add_emit_noise_right( modifier_ref, next(self._pauli_lindblad_map_count), ) - return self._add_emit_right(node) + return self._add_right(node) def add_emit_meas_basis_change(self, qubits: QubitPartition, basis_ref: StrRef) -> NodeIndex: """Add a node that emits virtual gates left to measure in a basis. @@ -704,7 +686,7 @@ def add_emit_meas_basis_change(self, qubits: QubitPartition, basis_ref: StrRef) subsystems = self.qubits_to_indices(qubits) node = PreChangeBasis(subsystems, Direction.LEFT, VirtualType.U2, basis_ref) - return self._add_emit_left(node) + return self._add_left(node) def add_emit_prep_basis_change(self, qubits: QubitPartition, basis_ref: StrRef) -> NodeIndex: """Add a node that emits virtual gates right to prepare a basis. @@ -730,7 +712,7 @@ def add_emit_prep_basis_change(self, qubits: QubitPartition, basis_ref: StrRef) subsystems = self.qubits_to_indices(qubits) node = PreChangeBasis(subsystems, Direction.RIGHT, VirtualType.U2, basis_ref) - return self._add_emit_right(node) + return self._add_right(node) def add_propagate(self, instr: CircuitInstruction, mode: InstructionMode, params: ParamSpec): """Add a node that propagates virtual gates through an operation. @@ -776,13 +758,11 @@ def add_propagate(self, instr: CircuitInstruction, mode: InstructionMode, params rightward_node_candidate = NodeCandidate( self.graph, PrePropagate(subsystems, Direction.RIGHT, op, partition, mode, params) ) - match = DanglerMatch(node_types=(PreEmit, PrePropagate), direction=Direction.RIGHT) + match = DanglerMatch(node_types=(PreCopy, PreEmit, PrePropagate), direction=Direction.RIGHT) all_found_qubit_idxs = set() for found_idx, found_subsystems in self.find_then_remove_danglers(match, subsystems): all_found_qubit_idxs.update(found_subsystems.all_elements) - edge = PreEdge( - found_subsystems, Direction.RIGHT, found_idx in self._forced_copy_node_idxs - ) + edge = PreEdge(found_subsystems, Direction.RIGHT) # if the node candidate hasn't been added to the graph yet, it will be here: self.graph.add_edge(found_idx, rightward_node_candidate.node_idx, edge) @@ -798,12 +778,12 @@ def add_propagate(self, instr: CircuitInstruction, mode: InstructionMode, params self.graph, PrePropagate(subsystems, Direction.LEFT, op, partition, mode, params) ) all_found_qubit_idxs = set() - match = DanglerMatch(node_types=(PreCollect, PrePropagate), direction=Direction.LEFT) + match = DanglerMatch( + node_types=(PreCollect, PreCopy, PrePropagate), direction=Direction.LEFT + ) for found_idx, found_subsystems in self.find_then_remove_danglers(match, subsystems): all_found_qubit_idxs.update(found_subsystems.all_elements) - edge = PreEdge( - found_subsystems, Direction.LEFT, found_idx in self._forced_copy_node_idxs - ) + edge = PreEdge(found_subsystems, Direction.LEFT) # if the node candidate hasn't been added to the graph yet, it will be here: self.graph.add_edge(leftward_node_candidate.node_idx, found_idx, edge) @@ -985,7 +965,7 @@ def validate_no_rightward_danglers(self): Raises: SamplexBuildError: If any nodes are expecting collections. """ - match = DanglerMatch(node_types=(PreEmit, PrePropagate), direction=Direction.RIGHT) + match = DanglerMatch(node_types=(PreCopy, PreEmit, PrePropagate), direction=Direction.RIGHT) uncollected_qubit_idxs = set() for qubit_idx, node_idxs in self._dangling.items(): for node_idx in node_idxs: @@ -1085,6 +1065,14 @@ def finalize(self): order, register_names, ) + elif isinstance(pre_node, PreCopy): + self.add_copy_node( + samplex, + pre_node_idx, + pre_nodes_to_nodes, + order, + register_names, + ) else: raise SamplexBuildError(f"No lowering method found for {pre_node}.") @@ -1221,6 +1209,59 @@ def add_change_basis_node( for pre_successor_idx in self.graph.successor_indices(pre_basis_idx): register_names[pre_successor_idx][pre_basis_idx] = reg_name + def add_copy_node( + self, + samplex: Samplex, + pre_copy_idx: NodeIndex, + pre_nodes_to_nodes: dict[NodeIndex, NodeIndex], + order: dict[NodeIndex, int], + register_names: dict[NodeIndex, dict[NodeIndex, RegisterName]], + ) -> None: + """Add copy node to a samplex, mutating it in place. + + Args: + samplex: The samplex to add nodes to. + pre_copy_idx: The index of the pre-copy node to turn into a copy node. + pre_nodes_to_nodes: A map from pre-node indices to node indices. + order: A map from pre-node indices to integers representing their position in a + topological sort of the samplex state graph. + register_names: A map such that ``register_names[a][b]`` is the name of the register + implied by the edge (a, b) in the samplex state graph. + """ + reg_idx = order[pre_copy_idx] + pre_pred_idx = next(iter(self.sorted_predecessor_idxs(pre_copy_idx, order))) + register_name = register_names[pre_copy_idx][pre_pred_idx] + + node_predecessor = samplex.graph[pre_nodes_to_nodes[pre_pred_idx]] + if register_name in node_predecessor.writes_to(): + virtual_type = node_predecessor.writes_to()[register_name][1] + else: + virtual_type = node_predecessor.instantiates()[register_name][1] + + combine_node_idx = self.add_combine_node( + samplex, + pre_copy_idx, + pre_nodes_to_nodes, + order, + register_names, + combine_name := f"combine_{reg_idx}", + virtual_type, + ) + + copy_node = CopyNode( + combine_name, + copy_name := f"copy_{reg_idx}", + virtual_type, + len(self.graph[pre_copy_idx].subsystems), + ) + + node_idx = samplex.add_node(copy_node) + samplex.add_edge(combine_node_idx, node_idx) + pre_nodes_to_nodes[pre_copy_idx] = node_idx + + for pre_successor_idx in self.graph.successor_indices(pre_copy_idx): + register_names[pre_successor_idx][pre_copy_idx] = copy_name + def add_inject_noise_node( self, samplex: Samplex, @@ -1359,7 +1400,6 @@ def add_combine_node( input_register_name=input_register_name, output_register_name=combined_register_name, slice_idxs=slice_idxs, - force_copy=pre_edge.force_register_copy, ) else: combine_node = CombineRegistersNode( diff --git a/samplomatic/pre_samplex/utils.py b/samplomatic/pre_samplex/utils.py index 56ab7d3b..b341f8f8 100644 --- a/samplomatic/pre_samplex/utils.py +++ b/samplomatic/pre_samplex/utils.py @@ -81,5 +81,4 @@ def merge_pre_edges( return PreEdge( subsystems=QubitIndicesPartition.union(*(edge.subsystems for edge in edges)), direction=next(iter(directions)), - force_register_copy=any(edge.force_register_copy for edge in edges), ) diff --git a/samplomatic/samplex/nodes/__init__.py b/samplomatic/samplex/nodes/__init__.py index 8944ddff..d6ac65db 100644 --- a/samplomatic/samplex/nodes/__init__.py +++ b/samplomatic/samplex/nodes/__init__.py @@ -16,6 +16,7 @@ from .collection_node import CollectionNode from .combine_registers_node import CombineRegistersNode from .conversion_node import ConversionNode +from .copy_node import CopyNode from .evaluation_node import EvaluationNode from .inject_noise_node import InjectNoiseNode from .multiplication_node import LeftMultiplicationNode, RightMultiplicationNode diff --git a/samplomatic/samplex/nodes/copy_node.py b/samplomatic/samplex/nodes/copy_node.py new file mode 100644 index 00000000..4593da52 --- /dev/null +++ b/samplomatic/samplex/nodes/copy_node.py @@ -0,0 +1,72 @@ +# This code is a Qiskit project. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""CopyNode""" + +from copy import deepcopy + +from ...annotations import VirtualType +from .evaluation_node import EvaluationNode + + +class CopyNode(EvaluationNode): + """Copies a register. + + Args: + regoster_name: The name of the register to copy. + output_name: The name of the copy register. + register_type: The type of register to copy. + num_subsystems: The number of subsystems the register contains. + """ + + def __init__( + self, + register_name: str, + output_name: str, + output_type: VirtualType, + num_subsystems: int, + ): + self._register_name = register_name + self._output_name = output_name + self._output_type = output_type + self._num_subsystems = num_subsystems + + @property + def outgoing_register_type(self): + return self._output_type + + def instantiates(self): + return {self._output_name: (self._num_subsystems, self._output_type)} + + def reads_from(self): + return {self._register_name: (set(range(self._num_subsystems)), self._output_type)} + + def evaluate(self, registers, *_): + registers[self._output_name] = deepcopy(registers[self._register_name]) + + def __eq__(self, other): + return ( + isinstance(other, CopyNode) + and self._register_name == other._register_name + and self._output_name == other._output_name + and self._output_type == other._output_type + and self._num_subsystems == other._num_subsystems + ) + + def get_style(self): + return ( + super() + .get_style() + .append_data("Copies", self._register_name) + .append_data("Output Type", self._output_type.name) + .append_data("Output Num Subsystems", self._num_subsystems) + ) diff --git a/samplomatic/samplex/nodes/slice_register_node.py b/samplomatic/samplex/nodes/slice_register_node.py index 929aba71..b0e3eb09 100644 --- a/samplomatic/samplex/nodes/slice_register_node.py +++ b/samplomatic/samplex/nodes/slice_register_node.py @@ -17,7 +17,6 @@ from collections.abc import Sequence import numpy as np -from numpy.typing import ArrayLike from ...aliases import RegisterName, SubsystemIndex from ...annotations import VirtualType @@ -41,8 +40,6 @@ class SliceRegisterNode(EvaluationNode): input_register_name: The name of the input register. output_register_name: The name of the output register. slice_idxs: The indices used to slice the register. - force_copy: Whether or not to force output register to be a copy instead of a view of - the input register. Raises: SamplexConstructionError: If ``slice_idxs`` has the wrong shape. @@ -55,7 +52,6 @@ def __init__( input_register_name: RegisterName, output_register_name: RegisterName, slice_idxs: slice | Sequence[SubsystemIndex], - force_copy: bool = False, ): self._input_type = input_type self._output_type = output_type @@ -71,9 +67,6 @@ def __init__( f"'slice_idxs' for '{input_register_name}' has a shape {slice_idxs.shape}, " "but a shape with a single axes is required." ) - # Check if indices could be converted to a slice - elif not force_copy: - self._slice_idxs = get_slice_from_idxs(slice_idxs) else: self._slice_idxs = slice_idxs @@ -145,27 +138,3 @@ def get_style(self): .append_data("Output Register Name", self._output_register_name) .append_data("Input Register Name", self._input_register_name) ) - - -def get_slice_from_idxs(slice_idxs: ArrayLike) -> ArrayLike | slice: - """Return a :class:`slice` object if the given indices are stridable. - - Args: - slice_idxs: The indices to check. - - Returns: - The provided indices, or an equivalent :class:`slice` object. - """ - if len(slice_idxs) == 1: - return slice(slice_idxs[0], slice_idxs[0] + 1, 1) - else: - step = slice_idxs[1] - slice_idxs[0] - expected = slice_idxs[0] + step * np.arange(len(slice_idxs)) - - if np.array_equal(slice_idxs, expected): - return slice( - int(slice_idxs[0]), - int(slice_idxs[-1] + step) if slice_idxs[-1] + step >= 0 else None, - int(step), - ) - return slice_idxs diff --git a/samplomatic/serialization/node_serializers.py b/samplomatic/serialization/node_serializers.py index 7c6c46a9..a7c714db 100644 --- a/samplomatic/serialization/node_serializers.py +++ b/samplomatic/serialization/node_serializers.py @@ -23,6 +23,7 @@ CollectZ2ToOutputNode, CombineRegistersNode, ConversionNode, + CopyNode, InjectNoiseNode, LeftMultiplicationNode, LeftU2ParametricMultiplicationNode, @@ -468,3 +469,31 @@ def deserialize(cls, data): data["register_name"], orjson.loads(data["param_indices"]), ) + + +class CopyNodeSerializer(TypeSerializer[CopyNode]): + """Serializer for :class:`~.CopyNode`.""" + + TYPE_ID = "N13" + TYPE = CopyNode + + class SSV1(DataSerializer[CopyNode]): + MIN_SSV = 2 + + @classmethod + def serialize(cls, obj): + return { + "register_name": obj._register_name, # noqa: SLF001 + "ouput_name": obj._output_name, # noqa: SLF001 + "output_type": obj._output_type.value, # noqa: SLF001 + "num_subsystems": obj._num_subsystems, # noqa: SLF001 + } + + @classmethod + def deserialize(cls, data): + return CopyNode( + data["register_name"], + data["output_name"], + VirtualType(data["output_type"]), + int(data["num_subsystems"]), + ) diff --git a/test/integration/test_dynamic_circuits.py b/test/integration/test_dynamic_circuits.py index b9ab26ea..e413a6dd 100644 --- a/test/integration/test_dynamic_circuits.py +++ b/test/integration/test_dynamic_circuits.py @@ -63,141 +63,130 @@ def test_non_twirled_parametric_conditional(self, save_plot): def test_right_dressed_twirled_conditional(self, save_plot): """Test a circuit with a conditional in a right-dressed twirl box.""" - # TODO: uncomment these lines when dynamic circuits are supported again - # circuit = QuantumCircuit(3, 2) - # circuit.h(0) - # circuit.measure(0, 0) - - # with circuit.box([Twirl(dressing="left")]): - # circuit.noop(0, 1, 2) - # with circuit.box([Twirl(dressing="right")]): - # circuit.cx(0, 1) - # with circuit.if_test((circuit.clbits[0], 1)) as _else: - # circuit.cx(0, 1) - # circuit.x(0) - # with _else: - # circuit.cx(1, 0) - # circuit.sx(0) - # circuit.x(2) - # circuit.h(1) - # circuit.measure_all() - - # sample_simulate_and_compare_counts(circuit, save_plot) + circuit = QuantumCircuit(3, 2) + circuit.h(0) + circuit.measure(0, 0) + + with circuit.box([Twirl(dressing="left")]): + circuit.cx(0, 1) + with circuit.box([Twirl(dressing="right")]): + with circuit.if_test((circuit.clbits[0], 1)) as _else: + circuit.cx(0, 1) + circuit.x(0) + with _else: + circuit.cx(1, 0) + circuit.sx(0) + circuit.h(1) + circuit.measure_all() + + sample_simulate_and_compare_counts(circuit, save_plot) def test_right_dressed_twirled_conditional_no_else(self, save_plot): """Test a conditional without else clause in a right-dressed twirl box.""" - # TODO: uncomment these lines when dynamic circuits are supported again - # circuit = QuantumCircuit(3, 2) - # circuit.h(0) - # circuit.measure(0, 0) - - # with circuit.box([Twirl(dressing="left")]): - # circuit.noop(0, 1, 2) - # with circuit.box([Twirl(dressing="right")]): - # circuit.cx(0, 1) - # with circuit.if_test((circuit.clbits[0], 1)): - # circuit.cx(0, 1) - # circuit.x(0) - # circuit.x(2) - # circuit.h(1) - # circuit.measure_all() - - # sample_simulate_and_compare_counts(circuit, save_plot) + circuit = QuantumCircuit(2, 2) + circuit.h(0) + circuit.measure(0, 0) + + with circuit.box([Twirl(dressing="left")]): + circuit.cx(0, 1) + with circuit.box([Twirl(dressing="right")]): + with circuit.if_test((circuit.clbits[0], 1)): + circuit.cx(0, 1) + circuit.x(0) + circuit.h(1) + circuit.measure_all() + + sample_simulate_and_compare_counts(circuit, save_plot) def test_right_dressed_parametric_twirled_conditional(self, save_plot): """Test a circuit with a parametric conditional in a right-dressed twirl box.""" - # TODO: uncomment these lines when dynamic circuits are supported again - # circuit = QuantumCircuit(3, 1) - # p = Parameter("p") - # circuit.h(0) - # circuit.measure(0, 0) - - # with circuit.box([Twirl(dressing="left")]): - # circuit.noop(0, 1, 2) - # with circuit.box([Twirl(dressing="right")]): - # circuit.cx(0, 1) - # with circuit.if_test((circuit.clbits[0], 1)) as _else: - # circuit.cx(0, 1) - # circuit.x(0) - # circuit.rx(p, 0) - # with _else: - # circuit.cx(1, 0) - # circuit.sx(0) - # circuit.rx(2 * p, 1) - # circuit.x(2) - # circuit.h(1) - # circuit.measure_all() - - # sample_simulate_and_compare_counts(circuit, save_plot) + circuit = QuantumCircuit(3, 1) + p = Parameter("p") + circuit.h(0) + circuit.measure(0, 0) + + with circuit.box([Twirl(dressing="left")]): + circuit.noop(2) + circuit.cx(0, 1) + with circuit.box([Twirl(dressing="right")]): + circuit.noop(2) + with circuit.if_test((circuit.clbits[0], 1)) as _else: + circuit.cx(0, 1) + circuit.x(0) + circuit.rx(p, 0) + with _else: + circuit.cx(1, 0) + circuit.sx(0) + circuit.rx(2 * p, 1) + circuit.h(1) + circuit.measure_all() + + sample_simulate_and_compare_counts(circuit, save_plot) def test_left_dressed_twirled_conditional(self, save_plot): """Test a circuit with a conditional in a left-dressed twirl box.""" - # TODO: uncomment these lines when dynamic circuits are supported again - # circuit = QuantumCircuit(3, 2) - # circuit.h(0) - # circuit.measure(0, 0) - - # with circuit.box([Twirl(dressing="left")]): - # with circuit.if_test((circuit.clbits[0], 1)) as _else: - # circuit.x(0) - # circuit.cx(0, 1) - # with _else: - # circuit.sx(0) - # circuit.cx(1, 0) - # circuit.x(2) - # circuit.cx(0, 1) - # with circuit.box([Twirl(dressing="right")]): - # circuit.h(1) - # circuit.noop(0, 2) - - # circuit.measure_all() - - # sample_simulate_and_compare_counts(circuit, save_plot) + circuit = QuantumCircuit(3, 2) + circuit.h(0) + circuit.measure(0, 0) + + with circuit.box([Twirl(dressing="left")]): + with circuit.if_test((circuit.clbits[0], 1)) as _else: + circuit.x(0) + circuit.cx(0, 1) + with _else: + circuit.sx(0) + circuit.cx(1, 0) + circuit.x(2) + with circuit.box([Twirl(dressing="right")]): + circuit.cx(0, 1) + circuit.h(1) + circuit.noop(0, 2) + + circuit.measure_all() + + sample_simulate_and_compare_counts(circuit, save_plot) def test_left_dressed_twirled_conditional_no_else(self, save_plot): """Test a conditional without else clause in a left-dressed twirl box.""" - # TODO: uncomment these lines when dynamic circuits are supported again - # circuit = QuantumCircuit(3, 2) - # circuit.h(0) - # circuit.measure(0, 0) + circuit = QuantumCircuit(2, 2) + circuit.h(0) + circuit.measure(0, 0) - # with circuit.box([Twirl(dressing="left")]): - # with circuit.if_test((circuit.clbits[0], 1)): - # circuit.x(0) - # circuit.cx(0, 1) - # circuit.x(2) - # circuit.cx(0, 1) - # with circuit.box([Twirl(dressing="right")]): - # circuit.h(1) - # circuit.noop(0, 2) + with circuit.box([Twirl(dressing="left")]): + with circuit.if_test((circuit.clbits[0], 1)): + circuit.x(0) + circuit.cx(0, 1) + with circuit.box([Twirl(dressing="right")]): + circuit.cx(0, 1) + circuit.h(1) + circuit.noop(0) - # circuit.measure_all() + circuit.measure_all() - # sample_simulate_and_compare_counts(circuit, save_plot) + sample_simulate_and_compare_counts(circuit, save_plot) def test_left_dressed_parametric_twirled_conditional(self, save_plot): """Test a circuit with a parametric conditional in a left-dressed twirl box.""" - # TODO: uncomment these lines when dynamic circuits are supported again - # circuit = QuantumCircuit(3, 1) - # p = Parameter("p") - # circuit.h(0) - # circuit.measure(0, 0) - - # with circuit.box([Twirl(dressing="left")]): - # with circuit.if_test((circuit.clbits[0], 1)) as _else: - # circuit.x(0) - # circuit.rx(p, 1) - # circuit.cx(0, 1) - # with _else: - # circuit.sx(0) - # circuit.rx(2 * p, 0) - # circuit.cx(1, 0) - # circuit.x(2) - # circuit.cx(0, 1) - # with circuit.box([Twirl(dressing="right")]): - # circuit.h(1) - # circuit.noop(0, 2) - - # circuit.measure_all() - - # sample_simulate_and_compare_counts(circuit, save_plot) + circuit = QuantumCircuit(3, 1) + p = Parameter("p") + circuit.h(0) + circuit.measure(0, 0) + + with circuit.box([Twirl(dressing="left")]): + with circuit.if_test((circuit.clbits[0], 1)) as _else: + circuit.x(0) + circuit.rx(p, 1) + circuit.cx(0, 1) + with _else: + circuit.sx(0) + circuit.rx(2 * p, 0) + circuit.cx(1, 0) + circuit.x(2) + with circuit.box([Twirl(dressing="right")]): + circuit.cx(0, 1) + circuit.h(1) + circuit.noop(0, 2) + + circuit.measure_all() + + sample_simulate_and_compare_counts(circuit, save_plot) diff --git a/test/unit/test_builders/test_builders_pre_samplex.py b/test/unit/test_builders/test_builders_pre_samplex.py index a758b9ca..c90bd9be 100644 --- a/test/unit/test_builders/test_builders_pre_samplex.py +++ b/test/unit/test_builders/test_builders_pre_samplex.py @@ -40,7 +40,7 @@ def get_builder(self, qreg, creg=None): pre_samplex = PreSamplex(qubit_map=qubit_map, cregs=[creg]) qubits = QubitPartition.from_elements(qreg) builder = LeftBoxBuilder( - CollectionSpec(qubits, "Left", RzSxSynth()), + CollectionSpec(qubits, "Left", RzSxSynth(), QubitPartition.from_elements([])), EmissionSpec(qubits, "Right", VirtualType.PAULI), ) builder.set_samplex_state(pre_samplex).set_template_state(template_state) @@ -101,7 +101,7 @@ def test_wrong_twirl_type_for_measurement(self): pre_samplex = PreSamplex(qubit_map=qubit_map) qubits = QubitPartition.from_elements(qreg) builder = LeftBoxBuilder( - CollectionSpec(qubits, "Left", RzSxSynth()), + CollectionSpec(qubits, "Left", RzSxSynth(), QubitPartition.from_elements([])), EmissionSpec(qubits, "Right", VirtualType.U2), ) builder.set_samplex_state(pre_samplex) diff --git a/test/unit/test_builders/test_general_build_errors.py b/test/unit/test_builders/test_general_build_errors.py index 15b4987b..ec4efc18 100644 --- a/test/unit/test_builders/test_general_build_errors.py +++ b/test/unit/test_builders/test_general_build_errors.py @@ -39,7 +39,6 @@ def test_no_propagation_through_conditional_error(self): def test_bad_order_right_box(self): """Verify an error is raised if a gate follows a conditional in a right dressed box.""" - # TODO: uncomment these lines when dynamic circuits are supported again # circuit = QuantumCircuit(2, 3) # with circuit.box([Twirl(dressing="left")]): # circuit.noop(1) @@ -55,7 +54,6 @@ def test_bad_order_right_box(self): def test_bad_order_left_box(self): """Verify an error is raised if a conditional follows a gate in a left dressed box.""" - # TODO: uncomment these lines when dynamic circuits are supported again # circuit = QuantumCircuit(2, 3) # with circuit.box([Twirl(dressing="left")]): # circuit.x(1) @@ -69,7 +67,6 @@ def test_bad_order_left_box(self): def test_entangler_bad_order_left_box(self): """Verify that an error is raised if a entanglers appear in bad order.""" - # TODO: uncomment these lines when dynamic circuits are supported again # circuit = QuantumCircuit(2, 3) # with circuit.box([Twirl(dressing="left")]): # with circuit.if_test((circuit.clbits[0], 1)): diff --git a/test/unit/test_pre_samplex/test_pre_samplex.py b/test/unit/test_pre_samplex/test_pre_samplex.py index 32f08bdd..f5fe8e9c 100644 --- a/test/unit/test_pre_samplex/test_pre_samplex.py +++ b/test/unit/test_pre_samplex/test_pre_samplex.py @@ -689,33 +689,6 @@ def test_unsupported_propagate_error(self): ): pre_samplex.finalize() - def test_if_else_with_mergable_preedges(self): - """Test that merging of PreEdges maintains their force_register_copy property. - - Because it is a bit difficult to do by hand, we use the full pre_build. - """ - # circuit = QuantumCircuit(2, 1) - # circuit.measure(0, 0) - # with circuit.box([Twirl(dressing="left")]): - # with circuit.if_test((circuit.clbits[0], 1)) as _else: - # circuit.x(0) - # circuit.x(1) - # with _else: - # circuit.sx(0) - # circuit.sx(1) - # with circuit.box([Twirl(dressing="right")]): - # circuit.noop(0, 1) - - # _, pre_samplex = pre_build(circuit) - # pre_samplex.merge_parallel_pre_propagate_nodes() - # graph = pre_samplex.graph - # for emit_node in [6, 7]: - # assert not graph.get_edge_data(emit_node, 4).force_register_copy - # assert graph.get_edge_data(emit_node, 9).force_register_copy - # assert graph[4].operation.name == "x" - # assert graph[9].operation.name == "sx" - # assert graph[9].subsystems.all_elements == {0, 1} - class TestDraw: """Test the ``draw`` method.""" diff --git a/test/unit/test_samplex/test_nodes/test_copy_node.py b/test/unit/test_samplex/test_nodes/test_copy_node.py new file mode 100644 index 00000000..344741aa --- /dev/null +++ b/test/unit/test_samplex/test_nodes/test_copy_node.py @@ -0,0 +1,35 @@ +# This code is a Qiskit project. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + + +from samplomatic.annotations import VirtualType +from samplomatic.samplex.nodes import CopyNode +from samplomatic.virtual_registers import PauliRegister + + +def test_construction(): + """Test construction and basic attributes.""" + node = CopyNode("existing", "new", VirtualType.PAULI, 5) + assert node.instantiates() == {"new": (5, VirtualType.PAULI)} + assert node.reads_from() == {"existing": (set(range(5)), VirtualType.PAULI)} + assert not node.removes() + assert not node.writes_to() + assert node.outgoing_register_type is VirtualType.PAULI + + +def test_evaluate(): + """Test the evaluation method.""" + node = CopyNode("existing", "new", VirtualType.PAULI, 5) + registers = {"existing": PauliRegister.identity(5, 3)} + node.evaluate(registers) + assert registers["new"] == registers["existing"] + assert registers["new"] is not registers["existing"] diff --git a/test/unit/test_samplex/test_nodes/test_slice_register_node.py b/test/unit/test_samplex/test_nodes/test_slice_register_node.py index 3b6ef7f5..72bcbbe4 100644 --- a/test/unit/test_samplex/test_nodes/test_slice_register_node.py +++ b/test/unit/test_samplex/test_nodes/test_slice_register_node.py @@ -17,7 +17,6 @@ from samplomatic.annotations import VirtualType from samplomatic.exceptions import SamplexConstructionError from samplomatic.samplex.nodes import SliceRegisterNode -from samplomatic.samplex.nodes.slice_register_node import get_slice_from_idxs from samplomatic.virtual_registers import PauliRegister @@ -26,12 +25,7 @@ def test_equality(dummy_evaluation_node): node = SliceRegisterNode(VirtualType.PAULI, VirtualType.U2, "reg_in", "reg_out", [0, 1]) assert node == node assert node == SliceRegisterNode(VirtualType.PAULI, VirtualType.U2, "reg_in", "reg_out", [0, 1]) - assert node == SliceRegisterNode( - VirtualType.PAULI, VirtualType.U2, "reg_in", "reg_out", slice(0, 2, 1) - ) - assert node != SliceRegisterNode( - VirtualType.PAULI, VirtualType.U2, "reg_in", "reg_out", [0, 1], force_copy=True - ) + assert node == SliceRegisterNode(VirtualType.PAULI, VirtualType.U2, "reg_in", "reg_out", [0, 1]) assert node != dummy_evaluation_node assert node != SliceRegisterNode(VirtualType.U2, VirtualType.U2, "reg_in", "reg_out", [0, 1]) assert node != SliceRegisterNode( @@ -88,36 +82,6 @@ def test_slicing(slice_idxs): assert node.outgoing_register_type is VirtualType.PAULI -@pytest.mark.parametrize( - ("slice_idxs", "view"), - [ - ([1, 0], True), - ([2, 1], True), - ([2, 0], True), - ([0, 2, 1], False), - ([0], True), - ([0, 2], True), - ], -) -@pytest.mark.parametrize("force_copy", [True, False]) -def test_slicing_and_permuting(slice_idxs, view, force_copy): - """Test slicing and permuting at the same time a register using a ``SliceRegisterNode``.""" - registers = {"reg_in": PauliRegister([[0, 1, 2, 3], [1, 2, 3, 0], [2, 3, 0, 1], [0, 2, 3, 1]])} - node = SliceRegisterNode( - VirtualType.PAULI, VirtualType.PAULI, "reg_in", "reg_out", slice_idxs, force_copy - ) - - node.evaluate(registers, np.empty(())) - expected = registers["reg_in"].virtual_gates[slice_idxs].tolist() - assert registers["reg_out"].virtual_gates.tolist() == expected - if not force_copy and view: - assert np.shares_memory( - registers["reg_out"].virtual_gates, registers["reg_in"].virtual_gates - ) - assert node.reads_from()["reg_in"][0] == set(slice_idxs) - assert node.instantiates()["reg_out"][0] == len(slice_idxs) - - def test_raises(): """Test that the ``SliceRegisterNode`` raises.""" with pytest.raises(SamplexConstructionError, match="single axes"): @@ -138,25 +102,3 @@ def test_raises(): ) with pytest.raises(SamplexConstructionError, match="convertable"): node.validate_and_update({"reg_in": (2, VirtualType.U2)}) - - -@pytest.mark.parametrize( - "sliced_idxs", [[0, 2, 4], [2, 4], [0, 1, 2], [1, 2], [3, 2, 1], [1, 0], [4]] -) -def test_get_slice_helper_function(sliced_idxs): - """Test that the helper function get_slice_from_idxs returns a slice when possible.""" - a = np.arange(max(sliced_idxs) + 1) - res = a[get_slice_from_idxs(sliced_idxs)] - - assert np.array_equal(res, sliced_idxs) - assert np.shares_memory(res, a) - - -def test_get_slice_helper_function_no_slice(): - """Test thatget_slice_from_idxs doesn't return a slice when it's impossible.""" - sliced_idxs = [0, 1, 6, 3] - a = np.arange(max(sliced_idxs) + 1) - res = a[get_slice_from_idxs(sliced_idxs)] - - assert np.array_equal(res, sliced_idxs) - assert not np.shares_memory(res, a)