Codi font per a qilisdk.backends.qutip_backend

# Copyright 2025 Qilimanjaro Quantum Tech
#
# 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.
from __future__ import annotations

from typing import TYPE_CHECKING, Callable, Type, TypeVar, cast

import numpy as np
import qutip_qip.operations as QutipGates
from loguru import logger
from qutip import Qobj, basis, mesolve, qeye, tensor
from qutip_qip.circuit import CircuitSimulator, QubitCircuit
from qutip_qip.operations.gateclass import SingleQubitGate, is_qutip5

from qilisdk.analog.hamiltonian import Hamiltonian, PauliI, PauliOperator
from qilisdk.backends.backend import Backend
from qilisdk.core.qtensor import InitialState, QTensor, tensor_prod
from qilisdk.digital import RX, RY, RZ, SWAP, U1, U2, U3, Circuit, H, I, M, S, T, X, Y, Z
from qilisdk.digital.circuit_transpiler_passes import DecomposeMultiControlledGatesPass
from qilisdk.digital.exceptions import UnsupportedGateError
from qilisdk.digital.gates import Adjoint, BasicGate, Controlled
from qilisdk.functionals.functional_result import FunctionalResult
from qilisdk.noise import SupportsStaticLindblad, SupportsTimeDerivedLindblad
from qilisdk.settings import get_settings

if TYPE_CHECKING:
    from qilisdk.functionals import AnalogEvolution, DigitalPropagation
    from qilisdk.noise import LindbladGenerator, NoiseModel
    from qilisdk.readout import ReadoutMethod


[documents] TBasicGate = TypeVar("TBasicGate", bound=BasicGate)
[documents] BasicGateHandlersMapping = dict[Type[TBasicGate], Callable[[QubitCircuit, TBasicGate, int], None]]
[documents] TPauliOperator = TypeVar("TPauliOperator", bound=PauliOperator)
[documents] PauliOperatorHandlersMapping = dict[Type[TPauliOperator], Callable[[TPauliOperator], Qobj]]
def _complex_dtype() -> np.dtype: return get_settings().complex_precision.dtype
[documents] class QutipI(SingleQubitGate): """ Single-qubit I gate. Examples -------- >>> from qutip_qip.operations import X >>> I(0).get_compact_qobj() # doctest: +NORMALIZE_WHITESPACE Quantum object: dims=[[2], [2]], shape=(2, 2), type='oper', dtype=Dense, isherm=True Qobj data = [[1. 0.] [0. 1.]] """ def __init__(self, targets, **kwargs) -> None: # ruff: ignore[missing-type-function-argument, missing-type-kwargs] super().__init__(targets=targets, **kwargs)
[documents] self.name = "I"
[documents] self.latex_str = r"I"
[documents] def get_compact_qobj(self): # ruff: ignore[missing-return-type-undocumented-public-function, no-self-use] return qeye(2) if not is_qutip5 else qeye(2, dtype="dense")
[documents] class QutipBackend(Backend): """Backend that runs digital-circuit and analog time-evolution experiments using QuTiP. The backend is CPU-only and has no hardware dependencies, which makes it ideal for local development, CI pipelines, and educational notebooks. """ def __init__(self, nsteps: int = 10_000, noise_model: NoiseModel | None = None) -> None: """Instantiate a new :class:`QutipBackend`. Args: nsteps (int): The maximum number of internal steps for the ODE solver. Defaults to ``10_000``. noise_model (NoiseModel | None): Optional noise model. For analog time evolution, its global and per-qubit Lindblad channels are converted into QuTiP collapse operators (including time-dependent rates ``rate(t)``). Digital-circuit noise is not supported and is ignored with a warning. """
[documents] self.noise_model = noise_model
[documents] self.nsteps = nsteps
super().__init__() self._basic_gate_handlers: BasicGateHandlersMapping = { I: QutipBackend._handle_I, X: QutipBackend._handle_X, Y: QutipBackend._handle_Y, Z: QutipBackend._handle_Z, H: QutipBackend._handle_H, S: QutipBackend._handle_S, T: QutipBackend._handle_T, RX: QutipBackend._handle_RX, RY: QutipBackend._handle_RY, RZ: QutipBackend._handle_RZ, U1: QutipBackend._handle_U1, U2: QutipBackend._handle_U2, U3: QutipBackend._handle_U3, SWAP: QutipBackend._handle_SWAP, } # ty:ignore[invalid-assignment] logger.info("[QutipBackend] QutipBackend initialised") def _execute_digital_propagation( self, functional: DigitalPropagation, readout: list[ReadoutMethod] ) -> FunctionalResult: """Execute a digital-circuit propagation functional using QuTiP. Translates the circuit gates into QuTiP operations, runs the simulation via ``CircuitSimulator``, and returns the requested readout results. Args: functional (DigitalPropagation): The digital propagation functional to execute. readout (list[ReadoutMethod]): Readout specifications for result extraction. Returns: FunctionalResult: The execution result containing the requested readout data. Raises: ValueError: If the circuit contains mid-circuit measurements. """ logger.info("[QutipBackend] Executing Sampling") # Warn if noise model is provided, since we don't support digital noise here for now if self.noise_model is not None: logger.warning( "Noise model provided to QutipBackend, but digital noise is not supported. " "The noise model will be ignored." ) init_state = tensor(*[basis(2, 0) for _ in range(functional.circuit.nqubits)]) measurements_set = {q for g in functional.circuit.gates if isinstance(g, M) for q in g.qubits} measurements = sorted(measurements_set) gates = functional.circuit.gates # Find the index of the first measurement gate try: first_m = next(i for i, g in enumerate(gates) if isinstance(g, M)) except StopIteration: first_m = None # No measurements at all # If there's a measurement, check that nothing after it is a non-measurement if first_m is not None and any(not isinstance(g, M) for g in gates[first_m + 1 :]): raise ValueError( "Mid-circuit measurements are not supported in the QutipBackend. " "All measurements must be at the end of the circuit." ) transpiled_circuit = DecomposeMultiControlledGatesPass().run(functional.circuit) qutip_circuit = self._get_qutip_circuit(transpiled_circuit) sim = CircuitSimulator(qutip_circuit) final_state = QTensor(sim.run(init_state).get_final_states()[0].full()) logger.info("[QutipBackend] Sampling finished") if len(measurements) > 0 and len(measurements) != functional.circuit.nqubits: return FunctionalResult( readout_results=QutipBackend._construct_results_list( final_state=final_state, readout=readout, qubits_to_measure=measurements ) ) return FunctionalResult( readout_results=QutipBackend._construct_results_list(final_state=final_state, readout=readout) ) def _execute_analog_evolution(self, functional: AnalogEvolution, readout: list[ReadoutMethod]) -> FunctionalResult: """Compute the time evolution of an initial state under the given schedule. Uses QuTiP's ``mesolve`` to integrate the master equation over the schedule time steps. Args: functional (AnalogEvolution): The analog evolution functional to execute, containing the schedule and initial state. readout (list[ReadoutMethod]): Readout specifications for result extraction. Returns: FunctionalResult: The execution result, optionally including intermediate-state readouts when ``functional.store_intermediate_results`` is ``True``. Raises: ValueError: If the initial state has an unsupported shape (must be a ket, bra, or density matrix). """ logger.info( "[QutipBackend] Executing TimeEvolution (T={}, dt={})", functional.schedule.T, functional.schedule.dt ) steps = functional.schedule.tlist qutip_hamiltonians = [] for hamiltonian in functional.schedule.hamiltonians.values(): qutip_hamiltonians.append( Qobj( hamiltonian.to_qtensor(total_nqubits=functional.schedule.nqubits).data, dims=[[2 for _ in range(functional.schedule.nqubits)] for _ in range(2)], ) ) h_t = [ [ qutip_hamiltonians[i], np.array( [functional.schedule.coefficients[h][t] for t in steps], dtype=_complex_dtype(), ), ] for i, h in enumerate(functional.schedule.hamiltonians) ] state_dim = [] if isinstance(functional.initial_state, InitialState): initial_state = functional.initial_state.as_qtensor(functional.schedule.nqubits) else: initial_state = functional.initial_state if initial_state.is_density_matrix(): state_dim = [[2 for _ in range(initial_state.nqubits)] for _ in range(2)] elif initial_state.is_bra(): state_dim = [[1], [2 for _ in range(initial_state.nqubits)]] elif initial_state.is_ket(): state_dim = [[2 for _ in range(initial_state.nqubits)], [1]] else: logger.error("[QutipBackend] Invalid initial state provided") raise ValueError("invalid initial state provided.") qutip_init_state = Qobj(initial_state.dense(), dims=state_dim) qutip_obs: list[Qobj] = [] jump_ops: list = [] if self.noise_model is not None: jump_ops, hamiltonian_deltas = self._noise_model_to_qutip_cops( self.noise_model, functional.schedule.nqubits, steps, functional.schedule.dt ) for h in hamiltonian_deltas: h_t.append([h, np.ones(len(steps), dtype=_complex_dtype())]) results = mesolve( H=h_t, e_ops=qutip_obs, c_ops=jump_ops, rho0=qutip_init_state, tlist=steps, options={ "store_states": functional.store_intermediate_results, "store_final_state": True, "nsteps": self.nsteps, }, ) logger.info("[QutipBackend] TimeEvolution finished") return FunctionalResult( readout_results=QutipBackend._construct_results_list( final_state=QTensor(results.final_state.full()), readout=readout ), intermediate_results=( [ QutipBackend._construct_results_list(final_state=QTensor(state.full()), readout=readout) for state in results.states ] if functional.store_intermediate_results else None ), ) def _noise_model_to_qutip_cops( self, noise_model: NoiseModel, nqubits: int, steps: list[float], dt: float ) -> tuple[list, list[Qobj]]: """Convert a noise model into QuTiP collapse operators for ``mesolve``. Args: noise_model (NoiseModel): The qiliSDK noise model to convert. nqubits (int): Total number of qubits in the system. steps (list[float]): The schedule time grid (``mesolve`` ``tlist``), used to sample time-dependent rates. dt (float): Schedule time step, used to derive time-dependent Lindblad generators. Returns: tuple[list, list[Qobj]]: A pair ``(jump_ops, hamiltonian_deltas)`` where ``jump_ops`` is the list of (possibly time-dependent) collapse operators and ``hamiltonian_deltas`` are constant coherent Hamiltonian corrections to add to the Hamiltonian. """ jump_ops: list = [] hamiltonian_deltas: list[Qobj] = [] # Global noise for noise in noise_model.global_noise: generator = self._noise_as_lindblad(noise, dt) if generator is not None: self._add_lindblad_cops(generator, nqubits, steps, jump_ops, hamiltonian_deltas, qubits=None) # Per-qubit noise for qubit, noises in noise_model.per_qubit_noise.items(): for noise in noises: generator = self._noise_as_lindblad(noise, dt) if generator is not None: self._add_lindblad_cops(generator, nqubits, steps, jump_ops, hamiltonian_deltas, qubits=[qubit]) return jump_ops, hamiltonian_deltas @staticmethod def _noise_as_lindblad(noise: object, dt: float) -> LindbladGenerator | None: """Return the Lindblad generator for a noise source, if it exposes one. Args: noise (object): The noise source to convert. dt (float): Schedule time step, used to derive time-derived Lindblad generators. Returns: LindbladGenerator | None: The Lindblad generator, or ``None`` if the noise source does not expose a Lindblad representation. """ if isinstance(noise, SupportsStaticLindblad): return noise.as_lindblad() if isinstance(noise, SupportsTimeDerivedLindblad): return noise.as_lindblad_from_duration(duration=dt) return None def _add_lindblad_cops( self, generator: LindbladGenerator, nqubits: int, steps: list[float], jump_ops: list, hamiltonian_deltas: list[Qobj], qubits: list[int] | None, ) -> None: """Append the collapse operators of a Lindblad generator to ``jump_ops``. Single-qubit jump operators are embedded into the full ``nqubits`` Hilbert space; full-system operators are used as-is. Constant rates scale the operator by ``sqrt(rate)``; callable rates ``rate(t)`` are emitted as QuTiP time-dependent collapse operators. Args: generator (LindbladGenerator): The Lindblad generator to convert. nqubits (int): Total number of qubits in the system. steps (list[float]): The schedule time grid used to sample time-dependent rates. jump_ops (list): Accumulator for QuTiP collapse operators. hamiltonian_deltas (list[Qobj]): Accumulator for coherent Hamiltonian corrections. qubits (list[int] | None): Target qubit indices for single-qubit operators, or ``None`` to broadcast over every qubit (global). Raises: ValueError: If a jump operator is not a square matrix, its dimension is not a power of 2, or a single-qubit operator is requested on a multi-qubit operator scope. """ rates = generator.rates target_qubits = list(range(nqubits)) if qubits is None else qubits for i, jump_op in enumerate(generator.jump_operators): op_np = np.array(jump_op.dense(), dtype=_complex_dtype()) dim = op_np.shape[0] if op_np.shape[1] != dim: raise ValueError("Lindblad jump operators must be square matrices.") operator_nqubits = int(np.round(np.log2(dim))) if dim == 2**nqubits: embedded = [Qobj(op_np, dims=[[2 for _ in range(nqubits)] for _ in range(2)])] elif operator_nqubits == 1: embedded = [self._embed_single_qubit_operator(op_np, q, nqubits) for q in target_qubits] else: raise ValueError("Lindblad jump operators must be either single-qubit or full-system operators.") rate = rates[i] if rates is not None else None for op in embedded: if rate is None: jump_ops.append(op) elif callable(rate): rate_fn = cast("Callable[[float], float]", rate) # needed for type safety coeff = np.array([np.sqrt(rate_fn(t)) for t in steps], dtype=_complex_dtype()) jump_ops.append([op, coeff]) else: jump_ops.append(np.sqrt(float(cast("float", rate))) * op) if generator.hamiltonian is not None: hamiltonian_deltas.append(self._to_qubip_observables(generator.hamiltonian, nqubits)) @staticmethod def _embed_single_qubit_operator(op: np.ndarray, target_qubit: int, nqubits: int) -> Qobj: """Embed a single-qubit operator into the full ``nqubits`` Hilbert space. Args: op (np.ndarray): The 2x2 single-qubit operator matrix. target_qubit (int): The qubit index the operator acts on. nqubits (int): Total number of qubits in the system. Returns: Qobj: The operator acting on ``target_qubit`` and identity elsewhere. """ factors = [qeye(2) for _ in range(nqubits)] factors[target_qubit] = Qobj(op, dims=[[2], [2]]) return tensor(*factors) @staticmethod def _to_qubip_observables(obs: QTensor | Hamiltonian | PauliOperator, nqubits: int) -> Qobj: """Convert a QiliSDK observable to a QuTiP Qobj. Args: obs (QTensor | Hamiltonian | PauliOperator): The observable to convert. nqubits (int): The total number of qubits in the system. Returns: Qobj: The corresponding QuTiP Qobj representation of the observable. Raises: ValueError: If the observable type is unsupported. """ aux_obs = None identity = QTensor(PauliI(0).matrix) if isinstance(obs, PauliOperator): for i in range(nqubits): if aux_obs is None: aux_obs = identity if i != obs.qubit else QTensor(obs.matrix) else: aux_obs = ( tensor_prod([aux_obs, identity]) if i != obs.qubit else tensor_prod([aux_obs, QTensor(obs.matrix)]) ) elif isinstance(obs, Hamiltonian): aux_obs = QTensor(obs.to_matrix()) if obs.nqubits < nqubits: for _ in range(nqubits - obs.nqubits): aux_obs = tensor_prod([aux_obs, identity]) elif isinstance(obs, QTensor): aux_obs = obs if aux_obs is not None: return Qobj(aux_obs.dense(), dims=[[2 for _ in range(nqubits)] for _ in range(2)]) logger.error("[QutipBackend] Unsupported observable type {}", obs.__class__.__name__) raise ValueError(f"unsupported observable type of {obs.__class__}") def _get_qutip_circuit(self, circuit: Circuit) -> QubitCircuit: """Translate a qiliSDK circuit to a QuTiP circuit. Args: circuit (Circuit): the qiliSDK circuit to be translated to qutip. Raises: UnsupportedGateError: If the circuit contains a gate for which no handler is registered. Returns: QubitCircuit: the translated qutip circuit. """ qutip_circuit = QubitCircuit( circuit.nqubits, num_cbits=circuit.nqubits, input_states=[0 for _ in range(circuit.nqubits)] ) for gate in circuit.gates: if isinstance(gate, Controlled): self._handle_controlled(qutip_circuit, gate) elif isinstance(gate, Adjoint): self._handle_adjoint(qutip_circuit, gate) elif isinstance(gate, M): self._handle_M(qutip_circuit, gate) else: handler = self._basic_gate_handlers.get(type(gate), None) if handler is None: logger.error("[QutipBackend] Unsupported gate {}", type(gate).__name__) raise UnsupportedGateError(f"Unsupported gate {type(gate).__name__}") qubits = gate.target_qubits handler(qutip_circuit, gate, *qubits) return qutip_circuit def _handle_controlled(self, circuit: QubitCircuit, gate: Controlled) -> None: # ruff: ignore[no-self-use] """Handle a controlled gate operation by registering a custom QuTiP gate. For non-native controlled gates the block-matrix is constructed explicitly, mirroring the approach recommended in the QuTiP QIP documentation for custom controlled rotations. Args: circuit (QubitCircuit): The QuTiP circuit being constructed. gate (Controlled): The controlled gate to handle. """ if gate.name == "CNOT": circuit.add_gate("CNOT", targets=[*gate.target_qubits], controls=[*gate.control_qubits]) else: base_matrix = gate.basic_gate.matrix dim_target = base_matrix.shape[0] dim_total = 2 * dim_target dims = [[2] + [2] * len(gate.target_qubits), [2] + [2] * len(gate.target_qubits)] def qutip_controlled_gate() -> Qobj: mat = np.zeros((dim_total, dim_total), dtype=np.complex128) mat[:dim_target, :dim_target] = np.eye(dim_target, dtype=np.complex128) mat[dim_target:, dim_target:] = base_matrix return Qobj(mat, dims=dims) matrix_digest = base_matrix.tobytes().hex()[:16] gate_name = f"{gate.name}_{matrix_digest}" if gate_name not in circuit.user_gates: circuit.user_gates[gate_name] = qutip_controlled_gate circuit.add_gate(gate_name, targets=[*gate.control_qubits, *gate.target_qubits]) def _handle_adjoint(self, circuit: QubitCircuit, gate: Adjoint) -> None: # ruff: ignore[no-self-use] """Handle an adjoint (inverse) gate operation. Registers a custom QuTiP gate whose unitary is the adjoint of the wrapped basic gate. Args: circuit (QubitCircuit): The QuTiP circuit being constructed. gate (Adjoint): The adjoint gate to handle. """ def qutip_adjoined_gate() -> Qobj: return Qobj(gate.matrix) gate_name = "Adjoint_" + gate.name if gate_name not in circuit.user_gates: circuit.user_gates[gate_name] = qutip_adjoined_gate circuit.add_gate(gate_name, targets=[*gate.target_qubits]) @staticmethod def _handle_M(qutip_circuit: QubitCircuit, gate: M) -> None: """Handle a measurement gate. Adds a measurement operation to each target qubit, storing the classical result in the corresponding classical bit. Args: qutip_circuit (QubitCircuit): The QuTiP circuit being constructed. gate (M): The measurement gate to handle. """ for i in gate.target_qubits: qutip_circuit.add_measurement(f"M{i}", targets=[i], classical_store=i) @staticmethod def _handle_I(circuit: QubitCircuit, gate: I, qubit: int) -> None: """Handle an I (identity) gate operation.""" circuit.add_gate(QutipI(targets=qubit)) @staticmethod def _handle_X(circuit: QubitCircuit, gate: X, qubit: int) -> None: """Handle an X gate operation.""" circuit.add_gate(QutipGates.X(targets=qubit)) @staticmethod def _handle_Y(circuit: QubitCircuit, gate: Y, qubit: int) -> None: """Handle a Y gate operation.""" circuit.add_gate(QutipGates.Y(targets=qubit)) @staticmethod def _handle_Z(circuit: QubitCircuit, gate: Z, qubit: int) -> None: """Handle a Z gate operation.""" circuit.add_gate(QutipGates.Z(targets=qubit)) @staticmethod def _handle_H(circuit: QubitCircuit, gate: H, qubit: int) -> None: """Handle an H gate operation.""" circuit.add_gate(QutipGates.H(targets=qubit)) @staticmethod def _handle_S(circuit: QubitCircuit, gate: S, qubit: int) -> None: """Handle an S gate operation.""" circuit.add_gate(QutipGates.S(targets=qubit)) @staticmethod def _handle_T(circuit: QubitCircuit, gate: T, qubit: int) -> None: """Handle a T gate operation.""" circuit.add_gate(QutipGates.T(targets=qubit)) @staticmethod def _handle_RX(circuit: QubitCircuit, gate: RX, qubit: int) -> None: """Handle an RX gate operation.""" circuit.add_gate(QutipGates.RX(targets=[qubit], arg_value=gate.get_parameter_values()[0])) @staticmethod def _handle_RY(circuit: QubitCircuit, gate: RY, qubit: int) -> None: """Handle an RY gate operation.""" circuit.add_gate(QutipGates.RY(targets=[qubit], arg_value=gate.get_parameter_values()[0])) @staticmethod def _handle_RZ(circuit: QubitCircuit, gate: RZ, qubit: int) -> None: """Handle an RZ gate operation.""" circuit.add_gate(QutipGates.RZ(targets=[qubit], arg_value=gate.get_parameter_values()[0])) @staticmethod def _qutip_U1(phi: float) -> Qobj: mat = np.array([[1, 0], [0, np.exp(1j * phi)]], dtype=_complex_dtype()) return Qobj(mat, dims=[[2], [2]]) @staticmethod def _handle_U1(circuit: QubitCircuit, gate: U1, qubit: int) -> None: """Handle a U1 gate operation.""" u1_label = "U1" if u1_label not in circuit.user_gates: circuit.user_gates[u1_label] = QutipBackend._qutip_U1 circuit.add_gate(u1_label, targets=qubit, arg_value=gate.phi) @staticmethod def _qutip_U2(angles: list[float]) -> Qobj: phi = angles[0] gamma = angles[1] mat = (1 / np.sqrt(2)) * np.array( [ [1, -np.exp(1j * gamma)], [np.exp(1j * phi), np.exp(1j * (phi + gamma))], ], dtype=_complex_dtype(), ) return Qobj(mat, dims=[[2], [2]]) @staticmethod def _handle_U2(circuit: QubitCircuit, gate: U2, qubit: int) -> None: """Handle a U2 gate operation.""" u2_label = "U2" if u2_label not in circuit.user_gates: circuit.user_gates[u2_label] = QutipBackend._qutip_U2 circuit.add_gate(u2_label, targets=qubit, arg_value=[gate.phi, gate.gamma]) @staticmethod def _qutip_U3(angles: list[float]) -> Qobj: phi = angles[0] gamma = angles[1] theta = angles[2] mat = np.array( [ [np.cos(theta / 2), -np.exp(1j * gamma) * np.sin(theta / 2)], [np.exp(1j * phi) * np.sin(theta / 2), np.exp(1j * (phi + gamma)) * np.cos(theta / 2)], ], dtype=_complex_dtype(), ) return Qobj(mat, dims=[[2], [2]]) @staticmethod def _handle_U3(circuit: QubitCircuit, gate: U3, qubit: int) -> None: """Handle a U3 gate operation.""" u3_label = "U3" if u3_label not in circuit.user_gates: circuit.user_gates[u3_label] = QutipBackend._qutip_U3 circuit.add_gate(u3_label, targets=qubit, arg_value=[gate.phi, gate.gamma, gate.theta]) @staticmethod def _handle_SWAP(circuit: QubitCircuit, gate: SWAP, qubit_0: int, qubit_1: int) -> None: """Handle a SWAP gate operation.""" circuit.add_gate(QutipGates.SWAP(targets=[qubit_0, qubit_1]))