diff --git a/doc/releases/changelog-dev.md b/doc/releases/changelog-dev.md index ef823cfdae9..7027ae80d5b 100644 --- a/doc/releases/changelog-dev.md +++ b/doc/releases/changelog-dev.md @@ -80,6 +80,9 @@ added `binary_mapping()` function to map `BoseWord` and `BoseSentence` to qubit * Added method `preprocess` to the `QubitMixed` device class to preprocess the quantum circuit before execution. Necessary non-intrusive interfaces changes to class init method were made along the way to the `QubitMixed` device class to support new API feature. [(#6601)](https://github.com/PennyLaneAI/pennylane/pull/6601) +* Added method `preprocess` to the `QubitMixed` device class to preprocess the quantum circuit before execution. Necessary non-intrusive interfaces changes to class init method were made along the way to the `QubitMixed` device class to support new API feature. + [(#6601)](https://github.com/PennyLaneAI/pennylane/pull/6601) + * Added a second class `DefaultMixedNewAPI` to the `qml.devices.qubit_mixed` module, which is to be the replacement of legacy `DefaultMixed` which for now to hold the implementations of `preprocess` and `execute` methods. [(#6607)](https://github.com/PennyLaneAI/pennylane/pull/6507) diff --git a/pennylane/devices/default_mixed.py b/pennylane/devices/default_mixed.py index f123bc21a30..22e97d9202b 100644 --- a/pennylane/devices/default_mixed.py +++ b/pennylane/devices/default_mixed.py @@ -55,11 +55,12 @@ import warnings from collections.abc import Callable, Sequence from dataclasses import replace -from typing import Optional +from typing import Optional, Union from pennylane.ops.channel import __qubit_channels__ as channels from pennylane.transforms.core import TransformProgram from pennylane.tape import QuantumScript +from pennylane.typing import Result, ResultBatch from . import Device from .execution_config import ExecutionConfig @@ -72,6 +73,7 @@ validate_observables, ) from .modifiers import simulator_tracking, single_tape_support +from .qubit_mixed.simulate import simulate logger = logging.getLogger(__name__) logger.addHandler(logging.NullHandler()) @@ -182,8 +184,7 @@ def stopping_condition(op: qml.operation.Operator) -> bool: def warn_readout_error_state( tape: qml.tape.QuantumTape, ) -> tuple[Sequence[qml.tape.QuantumTape], Callable]: - """If a measurement in the QNode is an analytic state or density_matrix, and a readout error - parameter is defined, warn that readout error will not be applied. + """If a measurement in the QNode is an analytic state or density_matrix, warn that readout error will not be applied. Args: tape (QuantumTape, .QNode, Callable): a quantum circuit. @@ -975,8 +976,24 @@ def execute( self, circuits: QuantumScript, execution_config: Optional[ExecutionConfig] = None, - ) -> None: - raise NotImplementedError + ) -> Union[Result, ResultBatch]: + interface = ( + execution_config.interface + if execution_config.gradient_method in {"best", "backprop", None} + else None + ) + + return tuple( + simulate( + c, + rng=self._rng, + prng_key=self._prng_key, + debugger=self._debugger, + interface=interface, + readout_errors=self.readout_err, + ) + for c in circuits + ) def _setup_execution_config(self, execution_config: ExecutionConfig) -> ExecutionConfig: """This is a private helper for ``preprocess`` that sets up the execution config. diff --git a/pennylane/devices/qubit_mixed/__init__.py b/pennylane/devices/qubit_mixed/__init__.py index 608d1c1593d..a0e92dca775 100644 --- a/pennylane/devices/qubit_mixed/__init__.py +++ b/pennylane/devices/qubit_mixed/__init__.py @@ -24,7 +24,9 @@ apply_operation create_initial_state measure + simulate """ from .apply_operation import apply_operation from .initialize_state import create_initial_state from .measure import measure +from .simulate import get_final_state, measure_final_state, simulate diff --git a/pennylane/devices/qubit_mixed/initialize_state.py b/pennylane/devices/qubit_mixed/initialize_state.py index 26055909e61..4e1a68896c0 100644 --- a/pennylane/devices/qubit_mixed/initialize_state.py +++ b/pennylane/devices/qubit_mixed/initialize_state.py @@ -14,6 +14,7 @@ """Functions to prepare a state.""" from collections.abc import Iterable +from typing import Union import pennylane as qml import pennylane.numpy as np @@ -22,8 +23,8 @@ def create_initial_state( # pylint: disable=unsupported-binary-operation - wires: qml.wires.Wires | Iterable, - prep_operation: qml.operation.StatePrepBase | qml.QubitDensityMatrix = None, + wires: Union[qml.wires.Wires, Iterable], + prep_operation: Union[qml.operation.StatePrepBase, qml.QubitDensityMatrix] = None, like: str = None, ): r""" @@ -60,7 +61,7 @@ def _post_process(density_matrix, num_axes, like): r""" This post processor is necessary to ensure that the density matrix is in the correct format, i.e. the original tensor form, instead of the pure matrix form, as requested by all the other more fundamental chore functions in the module (again from some legacy code). """ - density_matrix = np.reshape(density_matrix, (-1,) + (2,) * num_axes) + density_matrix = np.reshape(density_matrix, (2,) * num_axes) dtype = str(density_matrix.dtype) floating_single = "float32" in dtype or "complex64" in dtype dtype = "complex64" if floating_single else "complex128" diff --git a/pennylane/devices/qubit_mixed/simulate.py b/pennylane/devices/qubit_mixed/simulate.py new file mode 100644 index 00000000000..47278071765 --- /dev/null +++ b/pennylane/devices/qubit_mixed/simulate.py @@ -0,0 +1,182 @@ +# Copyright 2018-2024 Xanadu Quantum Technologies Inc. + +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at + +# http://www.apache.org/licenses/LICENSE-2.0 + +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Simulate a quantum script for a qubit mixed state device.""" +# pylint: skip-file +# black: skip-file +import pennylane as qml +from pennylane.typing import Result + +from .apply_operation import apply_operation +from .initialize_state import create_initial_state +from .measure import measure + +INTERFACE_TO_LIKE = { + # map interfaces known by autoray to themselves + None: None, + "numpy": "numpy", + "autograd": "autograd", + "jax": "jax", + "torch": "torch", + "tensorflow": "tensorflow", + # map non-standard interfaces to those known by autoray + "auto": None, + "scipy": "numpy", + "jax-jit": "jax", + "jax-python": "jax", + "JAX": "jax", + "pytorch": "torch", + "tf": "tensorflow", + "tensorflow-autograph": "tensorflow", + "tf-autograph": "tensorflow", +} + + +def get_final_state(circuit, debugger=None, interface=None, **kwargs): + """ + Get the final state that results from executing the given quantum script. + + This is an internal function that will be called by ``default.mixed``. + + Args: + circuit (.QuantumScript): The single circuit to simulate + debugger (._Debugger): The debugger to use + interface (str): The machine learning interface to create the initial state with + + Returns: + Tuple[TensorLike, bool]: A tuple containing the final state of the quantum script and + whether the state has a batch dimension. + + """ + circuit = circuit.map_to_standard_wires() + + prep = None + if len(circuit) > 0 and isinstance(circuit[0], qml.operation.StatePrepBase): + prep = circuit[0] + + state = create_initial_state(sorted(circuit.op_wires), prep, like=INTERFACE_TO_LIKE[interface]) + + # initial state is batched only if the state preparation (if it exists) is batched + is_state_batched = bool(prep and prep.batch_size is not None) + for op in circuit.operations[bool(prep) :]: + state = apply_operation( + op, + state, + is_state_batched=is_state_batched, + debugger=debugger, + tape_shots=circuit.shots, + **kwargs, + ) + + # new state is batched if i) the old state is batched, or ii) the new op adds a batch dim + is_state_batched = is_state_batched or op.batch_size is not None + + num_operated_wires = len(circuit.op_wires) + for i in range(len(circuit.wires) - num_operated_wires): + # If any measured wires are not operated on, we pad the density matrix with zeros. + # We know they belong at the end because the circuit is in standard wire-order + # Since it is a dm, we must pad it with 0s on the last row and last column + current_axis = num_operated_wires + i + is_state_batched + state = qml.math.stack(([state] + [qml.math.zeros_like(state)]), axis=current_axis) + state = qml.math.stack(([state] + [qml.math.zeros_like(state)]), axis=-1) + + return state, is_state_batched + + +def measure_final_state( # pylint: disable=too-many-arguments + circuit, state, is_state_batched, rng=None, prng_key=None, readout_errors=None +) -> Result: + """ + Perform the measurements required by the circuit on the provided state. + + This is an internal function that will be called by ``default.mixed``. + + Args: + circuit (.QuantumScript): The single circuit to simulate + state (TensorLike): The state to perform measurement on + is_state_batched (bool): Whether the state has a batch dimension or not. + rng (Union[None, int, array_like[int], SeedSequence, BitGenerator, Generator]): A + seed-like parameter matching that of ``seed`` for ``numpy.random.default_rng``. + If no value is provided, a default RNG will be used. + prng_key (Optional[jax.random.PRNGKey]): An optional ``jax.random.PRNGKey``. This is + the key to the JAX pseudo random number generator. Only for simulation using JAX. + If None, the default ``sample_state`` function and a ``numpy.random.default_rng`` + will be for sampling. + readout_errors (List[Callable]): List of channels to apply to each wire being measured + to simulate readout errors. + + Returns: + Tuple[TensorLike]: The measurement results + """ + + circuit = circuit.map_to_standard_wires() + + if not circuit.shots: + # analytic case + if len(circuit.measurements) == 1: + return measure(circuit.measurements[0], state, is_state_batched, readout_errors) + + return tuple( + measure(mp, state, is_state_batched, readout_errors) for mp in circuit.measurements + ) + + +def simulate( # pylint: disable=too-many-arguments + circuit: qml.tape.QuantumScript, + rng=None, + prng_key=None, + debugger=None, + interface=None, + readout_errors=None, +) -> Result: + """Simulate a single quantum script. + + This is an internal function that will be called by ``default.mixed``. + + Args: + circuit (QuantumTape): The single circuit to simulate + rng (Union[None, int, array_like[int], SeedSequence, BitGenerator, Generator]): A + seed-like parameter matching that of ``seed`` for ``numpy.random.default_rng``. + If no value is provided, a default RNG will be used. + prng_key (Optional[jax.random.PRNGKey]): An optional ``jax.random.PRNGKey``. This is + the key to the JAX pseudo random number generator. If None, a random key will be + generated. Only for simulation using JAX. + debugger (_Debugger): The debugger to use + interface (str): The machine learning interface to create the initial state with + readout_errors (List[Callable]): List of channels to apply to each wire being measured + to simulate readout errors. + + Returns: + tuple(TensorLike): The results of the simulation + + Note that this function can return measurements for non-commuting observables simultaneously. + + This function assumes that all operations provide matrices. + + >>> qs = qml.tape.QuantumScript([qml.TRX(1.2, wires=0)], [qml.expval(qml.GellMann(0, 3)), qml.probs(wires=(0,1))]) + >>> simulate(qs) + (0.36235775447667357, + tensor([0.68117888, 0. , 0. , 0.31882112, 0. , 0. ], requires_grad=True)) + + """ + state, is_state_batched = get_final_state( + circuit, debugger=debugger, interface=interface, rng=rng, prng_key=prng_key + ) + return measure_final_state( + circuit, + state, + is_state_batched, + rng=rng, + prng_key=prng_key, + readout_errors=readout_errors, + ) diff --git a/tests/devices/qubit_mixed/test_qubit_mixed_simulate.py b/tests/devices/qubit_mixed/test_qubit_mixed_simulate.py new file mode 100644 index 00000000000..a898a1591dd --- /dev/null +++ b/tests/devices/qubit_mixed/test_qubit_mixed_simulate.py @@ -0,0 +1,275 @@ +# Copyright 2018-2024 Xanadu Quantum Technologies Inc. + +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at + +# http://www.apache.org/licenses/LICENSE-2.0 + +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Unit tests for simulate in devices/qubit_mixed.""" +import numpy as np +import pytest + +import pennylane as qml +from pennylane import math +from pennylane.devices.qubit_mixed import get_final_state, measure_final_state, simulate + +ml_interfaces = ["numpy", "autograd", "jax", "torch", "tensorflow"] + + +# pylint: disable=too-few-public-methods +class TestResultInterface: + """Test that the result interface is correct.""" + + @pytest.mark.all_interfaces + @pytest.mark.parametrize( + "op", [qml.RX(np.pi, [0]), qml.BasisState(np.array([1, 1]), wires=range(2))] + ) + @pytest.mark.parametrize("interface", ml_interfaces) + def test_result_has_correct_interface(self, op, interface): + """Test that even if no interface parameters are given, result is correct.""" + qs = qml.tape.QuantumScript([op], [qml.expval(qml.Z(0))]) + res = simulate(qs, interface=interface) + + assert qml.math.get_interface(res) == interface + + +# pylint: disable=too-few-public-methods +class TestStatePrepBase: + """Tests integration with various state prep methods.""" + + def test_basis_state(self): + """Test that the BasisState operator prepares the desired state.""" + qs = qml.tape.QuantumScript( + ops=[qml.BasisState(np.array([1, 1]), wires=[0, 1])], # prod state |1, 1> + measurements=[qml.probs(wires=[0, 1])], # measure only the wires we prepare + ) + probs = simulate(qs) + + # For state |1, 1>, only the |11> probability should be 1, others 0 + expected = np.zeros(4) + expected[3] = 1.0 # |11> is the last basis state + assert np.allclose(probs, expected) + + def test_basis_state_padding(self): + """Test that the BasisState operator prepares the desired state, with actual wires larger than the initial.""" + qs = qml.tape.QuantumScript( + ops=[qml.BasisState(np.array([1, 1]), wires=[0, 1])], # prod state |1, 1> + measurements=[qml.probs(wires=[0, 1, 2])], + ) + probs = simulate(qs) + expected = np.zeros(8) + expected[6] = 1.0 # Should be |110> = |6> + assert qml.math.allclose(probs, expected) + + +@pytest.mark.parametrize("subspace", [(0, 1), (0, 2), (2, 1)]) +class TestBasicCircuit: + """Tests a basic circuit with one RX gate and a few simple expectation values.""" + + @staticmethod + def get_quantum_script(phi, subspace): + """Get the quantum script where RX is applied then observables are measured""" + ops = [qml.RX(phi, wires=subspace[0])] + obs = [ + qml.expval(qml.PauliX(subspace[0])), + qml.expval(qml.PauliY(subspace[0])), + qml.expval(qml.PauliZ(subspace[0])), + ] + return qml.tape.QuantumScript(ops, obs) + + def test_basic_circuit_numpy(self, subspace): + """Test execution with a basic circuit.""" + phi = np.array(0.397) + + qs = self.get_quantum_script(phi, subspace) + result = simulate(qs) + + # For density matrix simulation of RX(phi), the expectations are: + expected_measurements = ( + 0, # appears to be 0 in density matrix formalism + -np.sin(phi), # has negative sign + np.cos(phi), # is correct + ) + + assert isinstance(result, tuple) + assert len(result) == 3 + assert np.allclose(result, expected_measurements) + + # Test state evolution and measurement separately + state, is_state_batched = get_final_state(qs) + result = measure_final_state(qs, state, is_state_batched) + + # For RX rotation in density matrix form - note flipped signs + expected_state = np.array( + [ + [np.cos(phi / 2) ** 2, 0.5j * np.sin(phi)], + [-0.5j * np.sin(phi), np.sin(phi / 2) ** 2], + ] + ) + + assert np.allclose(state, expected_state) + assert not is_state_batched + assert np.allclose(result, expected_measurements) + + @pytest.mark.autograd + def test_autograd_results_and_backprop(self, subspace): + """Tests execution and gradients with autograd""" + phi = qml.numpy.array(-0.52) + + def f(x): + qs = self.get_quantum_script(x, subspace) + return qml.numpy.array(simulate(qs)) + + result = f(phi) + expected = (0, -np.sin(phi), np.cos(phi)) # Note negative sin + assert qml.math.allclose(result, expected) + + g = qml.jacobian(f)(phi) + expected = (0, -np.cos(phi), -np.sin(phi)) # Note negative derivatives + assert qml.math.allclose(g, expected) + + @pytest.mark.jax + @pytest.mark.parametrize("use_jit", (True, False)) + def test_jax_results_and_backprop(self, use_jit, subspace): + """Tests execution and gradients with jax.""" + import jax + + phi = jax.numpy.array(0.678) + + def f(x): + qs = self.get_quantum_script(x, subspace) + return simulate(qs) + + if use_jit: + f = jax.jit(f) + + result = f(phi) + expected = (0, -np.sin(phi), np.cos(phi)) # Adjusted expectations + assert qml.math.allclose(result, expected) + + g = jax.jacobian(f)(phi) + expected = (0, -np.cos(phi), -np.sin(phi)) # Adjusted gradients + assert qml.math.allclose(g, expected) + + @pytest.mark.torch + def test_torch_results_and_backprop(self, subspace): + """Tests execution and gradients with torch.""" + import torch + + phi = torch.tensor(-0.526, requires_grad=True) + + def f(x): + qs = self.get_quantum_script(x, subspace) + return simulate(qs) + + result = f(phi) + expected = (0, -np.sin(phi.detach().numpy()), np.cos(phi.detach().numpy())) + + result_detached = math.asarray(result, like="torch").detach().numpy() + assert math.allclose(result_detached, expected) + + # Convert complex jacobian to real and take only real part for comparison + jacobian = math.asarray(torch.autograd.functional.jacobian(f, phi + 0j), like="torch") + jacobian = jacobian.real if hasattr(jacobian, "real") else jacobian + expected = (0, -np.cos(phi.detach().numpy()), -np.sin(phi.detach().numpy())) + assert math.allclose(jacobian.detach().numpy(), expected) + + @pytest.mark.tf + def test_tf_results_and_backprop(self, subspace): + """Tests execution and gradients with tensorflow.""" + import tensorflow as tf + + phi = tf.Variable(4.873) + + with tf.GradientTape(persistent=True) as grad_tape: + qs = self.get_quantum_script(phi, subspace) # Fixed: using phi instead of x + result = simulate(qs) + + expected = (0, -np.sin(float(phi)), np.cos(float(phi))) + assert qml.math.allclose(result, expected) + + expected = (0, -np.cos(float(phi)), -np.sin(float(phi))) + assert math.all( + [ + math.allclose(grad_tape.jacobian(one_obs_result, [phi])[0], one_obs_expected) + for one_obs_result, one_obs_expected in zip(result, expected) + ] + ) + + +class TestBroadcasting: + """Test that simulate works with broadcasted parameters.""" + + @staticmethod + def get_expected_state(x): + """Gets the expected final state of the circuit described in `get_ops_and_measurements`.""" + states = [] + for x_val in x: + cos = np.cos(x_val / 2) + sin = np.sin(x_val / 2) + state = np.array([[cos**2, 0.5j * np.sin(x_val)], [-0.5j * np.sin(x_val), sin**2]]) + states.append(state) + return np.stack(states) + + @staticmethod + def get_expectation_values(x): + """Gets the expected final expvals of the circuit described in `get_ops_and_measurements`.""" + return [-np.sin(x), np.cos(x)] + + @staticmethod + def get_quantum_script(x, shots=None, extra_wire=False): + """Gets quantum script of a circuit that includes parameter broadcasted operations and measurements.""" + # Use consistent wire ordering for the mapping test + wire_list = [0, 1] + if extra_wire: + wire_list.append(2) + + ops = [qml.RX(x, wires=wire_list[0])] + measurements = [qml.expval(qml.PauliY(wire_list[0])), qml.expval(qml.PauliZ(wire_list[0]))] + if extra_wire: + # Add measurement on the last wire for the extra wire case + measurements.insert(0, qml.expval(qml.PauliY(wire_list[-1]))) + + return qml.tape.QuantumScript(ops, measurements, shots=shots) + + def test_broadcasted_op_state(self): + """Test that simulate works for state measurements + when an operation has broadcasted parameters""" + x = np.array([0.8, 1.0, 1.2, 1.4]) + + qs = self.get_quantum_script(x) + res = simulate(qs) + + expected = self.get_expectation_values(x) + assert isinstance(res, tuple) + assert len(res) == 2 + assert np.allclose(res, expected) + + state, is_state_batched = get_final_state(qs) + res = measure_final_state(qs, state, is_state_batched) + + assert np.allclose(state, self.get_expected_state(x)) + assert is_state_batched + assert isinstance(res, tuple) + assert len(res) == 2 + assert np.allclose(res, expected) + + def test_broadcasting_with_extra_measurement_wires(self, mocker): + """Test that broadcasting works when the operations don't act on all wires.""" + spy = mocker.spy(qml, "map_wires") + x = np.array([0.8, 1.0, 1.2, 1.4]) + qs = self.get_quantum_script(x, extra_wire=True) + res = simulate(qs) + + assert isinstance(res, tuple) + assert len(res) == 3 + assert np.allclose(res[0], np.zeros_like(x)) + assert np.allclose(res[1:], self.get_expectation_values(x)) + # The mapping should be consistent with the wire ordering in get_quantum_script + assert spy.call_args_list[0].args == (qs, {0: 0, 2: 1}) diff --git a/tests/devices/test_default_mixed.py b/tests/devices/test_default_mixed.py index e298554b0c6..2472994a814 100644 --- a/tests/devices/test_default_mixed.py +++ b/tests/devices/test_default_mixed.py @@ -1360,9 +1360,3 @@ def test_too_many_wires(self): """Test error raised when too many wires requested""" with pytest.raises(ValueError, match="This device does not currently support"): DefaultMixedNewAPI(wires=24) - - def test_execute(self): - """Test that the execute method is defined""" - dev = DefaultMixedNewAPI(wires=[0, 1]) - with pytest.raises(NotImplementedError): - dev.execute(qml.tape.QuantumScript())