From e425f5274a35ae62bf582ed52993d9ec37709e07 Mon Sep 17 00:00:00 2001 From: BrunoLiegiBastonLiegi Date: Thu, 7 Aug 2025 10:30:24 +0200 Subject: [PATCH 1/4] feat: allowing to set even not trainable gates --- src/qibo/backends/abstract.py | 130 ++++++++++++++++++++++++++ src/qibo/hamiltonians/hamiltonians.py | 3 +- src/qibo/models/circuit.py | 26 ++++-- src/qibo/symbols.py | 3 +- 4 files changed, 151 insertions(+), 11 deletions(-) diff --git a/src/qibo/backends/abstract.py b/src/qibo/backends/abstract.py index 6e63de00cc..5b4d7f6bf8 100644 --- a/src/qibo/backends/abstract.py +++ b/src/qibo/backends/abstract.py @@ -1,7 +1,10 @@ import abc from typing import Optional, Union +from qibo import gates from qibo.config import raise_error +from qibo.hamiltonians import SymbolicHamiltonian +from qibo.symbols import Z class Backend(abc.ABC): @@ -416,6 +419,133 @@ def calculate_expectation_density_matrix( """Calculate expectation value of a density matrix given the observable matrix.""" raise_error(NotImplementedError) + def calculate_expectation_from_samples_symbolic( + self, hamiltonian, frequencies: dict, qubit_map: Union[tuple[int], list[int]] + ): + """ + Calculate the expectation value from the samples. + The observable has to be diagonal in the computational basis. + + Args: + hamiltonian (SymbolicHamiltonian): observable to calculate the expectation value of. + freq (dict): input frequencies of the samples. + qubit_map (list): qubit map. + + Returns: + (float): the calculated expectation value. + """ + for term in hamiltonian.terms: + # pylint: disable=E1101 + for factor in term.factors: + if not isinstance(factor, Z): + raise_error( + NotImplementedError, "Observable is not a Pauli-Z string." + ) + + if qubit_map is None: + qubit_map = list(range(hamiltonian.nqubits)) + + keys = list(frequencies.keys()) + counts = list(frequencies.values()) + counts = self.cast(counts, dtype=self.np.float64) / sum(counts) + expvals = [] + for term in hamiltonian.terms: + qubits = { + factor.target_qubit for factor in term.factors if factor.name[0] != "I" + } + expvals.extend( + [ + term.coefficient.real + * (-1) ** [state[qubit_map.index(q)] for q in qubits].count("1") + for state in keys + ] + ) + expvals = self.cast(expvals, dtype=counts.dtype).reshape( + len(hamiltonian.terms), len(frequencies) + ) + return self.np.sum(expvals @ counts) + hamiltonian.constant.real + + def calculate_expectation_from_circuit( + self, hamiltonian, circuit, nshots: int = 1000 + ): + """ + Calculate the expectation value from a circuit. + This even works for observables not completely diagonal in the computational + basis, but only diagonal at a term level in a defined basis. Namely, for + an observable of the form :math:``H = \\sum_i H_i``, where each :math:``H_i`` + consists in a `n`-qubits pauli operator :math:`P_0 \\otimes P_1 \\otimes \\cdots \\otimes P_n`, + the expectation value is computed by rotating the input circuit in the suitable + basis for each term :math:``H_i`` thus extracting the `term-wise` expectations + that are then summed to build the global expectation value. + Each term of the observable is treated separately, by measuring in the correct + basis and re-executing the circuit. + + Args: + hamiltonian (SymbolicHamiltonian): observable to calculate the expectation value of. + circuit (Circuit): input circuit. + nshots (int): number of shots, defaults to 1000. + + Returns: + (float): the calculated expectation value. + """ + # from qibo import gates + # from qibo.symbols import Z + # from qibo.hamiltonians import SymbolicHamiltonian + + rotated_circuits = [] + Z_observables = [] + qubit_maps = [] + # loop over the terms that can be diagonalized simultaneously + for terms in hamiltonian.diagonal_terms: + # for each term that can be diagonalized simultaneously + # extract the coefficient, Z observable and qubit map + # the basis rotation, instead, will be the same + tmp_obs = [] + measurements = {} + for term in terms: + # Only care about non-I terms + non_identity_factors = [] + # prepare the measurement basis and append it to the circuit + for factor in term.factors: + if factor.name[0] != "I": + non_identity_factors.append(factor) + q = factor.target_qubit + if q not in measurements: + measurements[q] = gates.M(q, basis=factor.gate.__class__) + + # build diagonal observable + # including the coefficient + symb_obs = prod( + Z(factor.target_qubit) for factor in non_identity_factors + ) + symb_obs = SymbolicHamiltonian( + term.coefficient * symb_obs, + nqubits=circuit.nqubits, + backend=self, + ) + tmp_obs.append(symb_obs) + + # Get the qubits we want to measure for each term + qubit_maps.append(sorted(measurements.keys())) + Z_observables.append(tmp_obs) + + circ_copy = circuit.copy(True) + circ_copy.add(list(measurements.values())) + rotated_circuits.append(circ_copy) + + # execute the circuits + results = self.execute_circuits(rotated_circuits, nshots=nshots) + + # construct the expectation value for each diagonal term + # and sum all together + expval = 0.0 + for res, obs, qmap in zip(results, Z_observables, qubit_maps): + freq = res.frequencies() + expval += sum( + self.calculate_expectation_from_samples(o, freq, qmap) for o in obs + ) + return hamiltonian.constant + expval + @abc.abstractmethod def calculate_matrix_exp( self, diff --git a/src/qibo/hamiltonians/hamiltonians.py b/src/qibo/hamiltonians/hamiltonians.py index 0a3a7dd8ea..66db309843 100644 --- a/src/qibo/hamiltonians/hamiltonians.py +++ b/src/qibo/hamiltonians/hamiltonians.py @@ -9,7 +9,8 @@ import numpy as np import sympy -from qibo.backends import Backend, _check_backend +import qibo.backends.abstract.Backend +from qibo.backends import _check_backend from qibo.config import log, raise_error from qibo.hamiltonians.abstract import AbstractHamiltonian from qibo.hamiltonians.terms import SymbolicTerm diff --git a/src/qibo/models/circuit.py b/src/qibo/models/circuit.py index 7c5d1ea591..dbbff01428 100644 --- a/src/qibo/models/circuit.py +++ b/src/qibo/models/circuit.py @@ -749,18 +749,21 @@ def gates_of_type(self, gate: Union[str, type]) -> List[Tuple[int, gates.Gate]]: return [(i, g) for i, g in enumerate(self.queue) if isinstance(g, gate)] - def _set_parameters_list(self, parameters, n): + def _set_parameters_list(self, parameters, n, include_not_trainable: bool = False): """Helper method for ``set_parameters`` when a list is given. Also works if ``parameters`` is ``np.ndarray`` or ``tf.Tensor``. """ - if n == len(self.trainable_gates): - for i, gate in enumerate(self.trainable_gates): + _gates = ( + self.parametrized_gates if include_not_trainable else self.trainable_gates + ) + if n == len(_gates): + for i, gate in enumerate(_gates): gate.parameters = parameters[i] - elif n == self.trainable_gates.nparams: + elif n == _gates.nparams: parameters = list(parameters) k = 0 - for i, gate in enumerate(self.trainable_gates): + for i, gate in enumerate(_gates): if gate.nparams == 1: gate.parameters = parameters[i + k] else: @@ -770,10 +773,10 @@ def _set_parameters_list(self, parameters, n): raise_error( ValueError, f"Given list of parameters has length {n} while " - + f"the circuit contains {len(self.trainable_gates)} parametrized gates.", + + f"the circuit contains {len(_gates)} parametrized gates.", ) - def set_parameters(self, parameters): + def set_parameters(self, parameters, include_not_trainable: bool = False): """Updates the parameters of the circuit's parametrized gates. For more information on how to use this method we refer to the @@ -819,8 +822,11 @@ def set_parameters(self, parameters): params = [0.123, 0.456, 0.789, 0.321] circuit.set_parameters(params) """ + _gates = ( + self.parametrized_gates if include_not_trainable else self.trainable_gates + ) if isinstance(parameters, dict): - diff = set(parameters.keys()) - self.trainable_gates.set + diff = set(parameters.keys()) - _gates.set if diff: raise_error( KeyError, @@ -836,7 +842,9 @@ def set_parameters(self, parameters): nparams = int(parameters.shape[0]) except AttributeError: nparams = len(parameters) - self._set_parameters_list(parameters, nparams) + self._set_parameters_list( + parameters, nparams, include_not_trainable=include_not_trainable + ) else: raise_error(TypeError, f"Invalid type of parameters {type(parameters)}.") diff --git a/src/qibo/symbols.py b/src/qibo/symbols.py index af40f65287..265e02ae52 100644 --- a/src/qibo/symbols.py +++ b/src/qibo/symbols.py @@ -3,8 +3,9 @@ import numpy as np import sympy +import qibo.backends.abstract.Backend from qibo import gates -from qibo.backends import Backend, _check_backend, get_backend, matrices +from qibo.backends import _check_backend, get_backend, matrices from qibo.config import raise_error From 0c8c44c5cc0ca768268cbe0d7e8b981fb68ea069 Mon Sep 17 00:00:00 2001 From: BrunoLiegiBastonLiegi Date: Thu, 7 Aug 2025 11:43:59 +0200 Subject: [PATCH 2/4] fix: remove commits from other pr --- src/qibo/backends/abstract.py | 130 -------------------------- src/qibo/hamiltonians/hamiltonians.py | 2 - src/qibo/symbols.py | 3 +- 3 files changed, 1 insertion(+), 134 deletions(-) diff --git a/src/qibo/backends/abstract.py b/src/qibo/backends/abstract.py index 5b4d7f6bf8..6e63de00cc 100644 --- a/src/qibo/backends/abstract.py +++ b/src/qibo/backends/abstract.py @@ -1,10 +1,7 @@ import abc from typing import Optional, Union -from qibo import gates from qibo.config import raise_error -from qibo.hamiltonians import SymbolicHamiltonian -from qibo.symbols import Z class Backend(abc.ABC): @@ -419,133 +416,6 @@ def calculate_expectation_density_matrix( """Calculate expectation value of a density matrix given the observable matrix.""" raise_error(NotImplementedError) - def calculate_expectation_from_samples_symbolic( - self, hamiltonian, frequencies: dict, qubit_map: Union[tuple[int], list[int]] - ): - """ - Calculate the expectation value from the samples. - The observable has to be diagonal in the computational basis. - - Args: - hamiltonian (SymbolicHamiltonian): observable to calculate the expectation value of. - freq (dict): input frequencies of the samples. - qubit_map (list): qubit map. - - Returns: - (float): the calculated expectation value. - """ - for term in hamiltonian.terms: - # pylint: disable=E1101 - for factor in term.factors: - if not isinstance(factor, Z): - raise_error( - NotImplementedError, "Observable is not a Pauli-Z string." - ) - - if qubit_map is None: - qubit_map = list(range(hamiltonian.nqubits)) - - keys = list(frequencies.keys()) - counts = list(frequencies.values()) - counts = self.cast(counts, dtype=self.np.float64) / sum(counts) - expvals = [] - for term in hamiltonian.terms: - qubits = { - factor.target_qubit for factor in term.factors if factor.name[0] != "I" - } - expvals.extend( - [ - term.coefficient.real - * (-1) ** [state[qubit_map.index(q)] for q in qubits].count("1") - for state in keys - ] - ) - expvals = self.cast(expvals, dtype=counts.dtype).reshape( - len(hamiltonian.terms), len(frequencies) - ) - return self.np.sum(expvals @ counts) + hamiltonian.constant.real - - def calculate_expectation_from_circuit( - self, hamiltonian, circuit, nshots: int = 1000 - ): - """ - Calculate the expectation value from a circuit. - This even works for observables not completely diagonal in the computational - basis, but only diagonal at a term level in a defined basis. Namely, for - an observable of the form :math:``H = \\sum_i H_i``, where each :math:``H_i`` - consists in a `n`-qubits pauli operator :math:`P_0 \\otimes P_1 \\otimes \\cdots \\otimes P_n`, - the expectation value is computed by rotating the input circuit in the suitable - basis for each term :math:``H_i`` thus extracting the `term-wise` expectations - that are then summed to build the global expectation value. - Each term of the observable is treated separately, by measuring in the correct - basis and re-executing the circuit. - - Args: - hamiltonian (SymbolicHamiltonian): observable to calculate the expectation value of. - circuit (Circuit): input circuit. - nshots (int): number of shots, defaults to 1000. - - Returns: - (float): the calculated expectation value. - """ - # from qibo import gates - # from qibo.symbols import Z - # from qibo.hamiltonians import SymbolicHamiltonian - - rotated_circuits = [] - Z_observables = [] - qubit_maps = [] - # loop over the terms that can be diagonalized simultaneously - for terms in hamiltonian.diagonal_terms: - # for each term that can be diagonalized simultaneously - # extract the coefficient, Z observable and qubit map - # the basis rotation, instead, will be the same - tmp_obs = [] - measurements = {} - for term in terms: - # Only care about non-I terms - non_identity_factors = [] - # prepare the measurement basis and append it to the circuit - for factor in term.factors: - if factor.name[0] != "I": - non_identity_factors.append(factor) - q = factor.target_qubit - if q not in measurements: - measurements[q] = gates.M(q, basis=factor.gate.__class__) - - # build diagonal observable - # including the coefficient - symb_obs = prod( - Z(factor.target_qubit) for factor in non_identity_factors - ) - symb_obs = SymbolicHamiltonian( - term.coefficient * symb_obs, - nqubits=circuit.nqubits, - backend=self, - ) - tmp_obs.append(symb_obs) - - # Get the qubits we want to measure for each term - qubit_maps.append(sorted(measurements.keys())) - Z_observables.append(tmp_obs) - - circ_copy = circuit.copy(True) - circ_copy.add(list(measurements.values())) - rotated_circuits.append(circ_copy) - - # execute the circuits - results = self.execute_circuits(rotated_circuits, nshots=nshots) - - # construct the expectation value for each diagonal term - # and sum all together - expval = 0.0 - for res, obs, qmap in zip(results, Z_observables, qubit_maps): - freq = res.frequencies() - expval += sum( - self.calculate_expectation_from_samples(o, freq, qmap) for o in obs - ) - return hamiltonian.constant + expval - @abc.abstractmethod def calculate_matrix_exp( self, diff --git a/src/qibo/hamiltonians/hamiltonians.py b/src/qibo/hamiltonians/hamiltonians.py index 66db309843..9a39f5325b 100644 --- a/src/qibo/hamiltonians/hamiltonians.py +++ b/src/qibo/hamiltonians/hamiltonians.py @@ -9,8 +9,6 @@ import numpy as np import sympy -import qibo.backends.abstract.Backend -from qibo.backends import _check_backend from qibo.config import log, raise_error from qibo.hamiltonians.abstract import AbstractHamiltonian from qibo.hamiltonians.terms import SymbolicTerm diff --git a/src/qibo/symbols.py b/src/qibo/symbols.py index 265e02ae52..af40f65287 100644 --- a/src/qibo/symbols.py +++ b/src/qibo/symbols.py @@ -3,9 +3,8 @@ import numpy as np import sympy -import qibo.backends.abstract.Backend from qibo import gates -from qibo.backends import _check_backend, get_backend, matrices +from qibo.backends import Backend, _check_backend, get_backend, matrices from qibo.config import raise_error From f91f5dd720a9cfa9fab39216916e68cc132ff95d Mon Sep 17 00:00:00 2001 From: BrunoLiegiBastonLiegi Date: Thu, 7 Aug 2025 11:45:39 +0200 Subject: [PATCH 3/4] fix: restored backend import --- src/qibo/hamiltonians/hamiltonians.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/qibo/hamiltonians/hamiltonians.py b/src/qibo/hamiltonians/hamiltonians.py index 9a39f5325b..0a3a7dd8ea 100644 --- a/src/qibo/hamiltonians/hamiltonians.py +++ b/src/qibo/hamiltonians/hamiltonians.py @@ -9,6 +9,7 @@ import numpy as np import sympy +from qibo.backends import Backend, _check_backend from qibo.config import log, raise_error from qibo.hamiltonians.abstract import AbstractHamiltonian from qibo.hamiltonians.terms import SymbolicTerm From 3cae0d55838cb6d30ad1cc5e5bc18ea88ebea2aa Mon Sep 17 00:00:00 2001 From: BrunoLiegiBastonLiegi Date: Tue, 12 May 2026 15:29:24 +0200 Subject: [PATCH 4/4] fix: clean up unwanted files --- src/qibo/models/#circuit.py# | 1592 ---------------------------------- src/qibo/models/.#circuit.py | 1 - src/qibo/models/circuit.py | 11 +- 3 files changed, 6 insertions(+), 1598 deletions(-) delete mode 100644 src/qibo/models/#circuit.py# delete mode 120000 src/qibo/models/.#circuit.py diff --git a/src/qibo/models/#circuit.py# b/src/qibo/models/#circuit.py# deleted file mode 100644 index ae84a636c3..0000000000 --- a/src/qibo/models/#circuit.py# +++ /dev/null @@ -1,1592 +0,0 @@ -"""Module for the Circuit class.""" - -import copy -import sys -from collections import Counter -from collections.abc import Iterable -from typing import Dict, List, Optional, Tuple, Union - -import numpy as np -from numpy.typing import ArrayLike -from tabulate import tabulate - -from qibo import __version__, gates -from qibo.backends import Backend, _check_backend, _Global -from qibo.config import raise_error -from qibo.gates import ParametrizedGate -from qibo.gates.abstract import Gate -from qibo.models._openqasm import QASMParser -from qibo.result import CircuitResult, QuantumState - -NoiseMapType = Union[Tuple[int, int, int], Dict[int, Tuple[int, int, int]]] - - -class _ParametrizedGates(list): - """Simple data structure for keeping track of parametrized gates. - - Useful for the ``circuit.set_parameters()`` method. - Holds parametrized gates in a list and a set and also keeps track of the - total number of parameters. - """ - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self.set = set() - self.nparams = 0 - - def append(self, gate): - super().append(gate) - self.set.add(gate) - self.nparams += gate.nparams - - -class _Queue(list): - """List that holds the queue of gates of a circuit. - - In addition to the queue, it holds a list of gate moments, where - each gate is placed in the earliest possible position depending for - the qubits it acts. - """ - - def __init__(self, nqubits: int): - super().__init__(self) - self.nqubits = nqubits - - def to_fused(self): - """Transform all gates in queue to :class:`qibo.gates.FusedGate`.""" - last_gate = {} - queue = self.__class__(self.nqubits) - for gate in self: - fgate = gates.FusedGate.from_gate(gate) - if isinstance(gate, gates.SpecialGate): - fgate.qubit_set = set(range(self.nqubits)) - fgate.init_args = sorted(fgate.qubit_set) - fgate.target_qubits = tuple(fgate.init_args) - - for q in fgate.qubits: - if q in last_gate: - neighbor = last_gate.get(q) - fgate.left_neighbors[q] = neighbor - neighbor.right_neighbors[q] = fgate - last_gate[q] = fgate - queue.append(fgate) - return queue - - def from_fused(self): - """Create queue from fused circuit. - - Create the fused circuit queue by removing gates that have been - fused to others. - """ - queue = self.__class__(self.nqubits) - for gate in self: - if not gate.marked: - if len(gate.gates) == 1: - # replace ``FusedGate``s that contain only one gate - # by this gate for efficiency - queue.append(gate.gates[0]) - else: - queue.append(gate) - elif isinstance(gate.gates[0], (gates.SpecialGate, gates.M)): - # special gates are marked by default so we need - # to add them manually - queue.append(gate.gates[0]) - return queue - - @property - def nmeasurements(self) -> int: - return len(list(filter(lambda gate: isinstance(gate, gates.M), self))) - - @property - def moments(self) -> list: - moments = [self.nqubits * [None]] - moment_index = self.nqubits * [0] - for gate in self: - qubits = ( - gate.qubits - if not isinstance(gate, gates.CallbackGate) - else tuple(range(self.nqubits)) # special gate acting on all qubits - ) - - # calculate moment index for this gate - idx = max(moment_index[q] for q in qubits) - for q in qubits: - if idx >= len(moments): - # Add a moment - moments.append(len(moments[-1]) * [None]) - moments[idx][q] = gate - moment_index[q] = idx + 1 - return moments - - -class Circuit: - """Circuit object which holds a list of gates. - - This circuit is symbolic and cannot perform calculations. - A specific backend has to be used for performing calculations. - - Circuits can be created with a specific number of qubits and wire names. - - - Either ``nqubits`` or ``wire_names`` must be provided. - - If only ``nqubits`` is provided, wire names will default to ``[0, 1, ..., nqubits - 1]``. - - If only ``wire_names`` is provided, ``nqubits`` will be set to the length of ``wire_names``. - - ``nqubits`` and ``wire_names`` must be consistent with each other. - - - Example:: - - from qibo import Circuit - - # Every circuit initialization below is valid - circuit = Circuit(5) # Default wire names are [0, 1, 2, 3, 4] - circuit = Circuit(["A", "B", "C", "D", "E"]) - circuit = Circuit(5, wire_names=["A", "B", "C", "D", "E"]) - circuit = Circuit(wire_names=["A", "B", "C", "D", "E"]) - - Args: - nqubits (int | list, optional): Number of qubits in the circuit or a list of wire names. - wire_names (list, optional): List of wire names - init_kwargs (dict): a dictionary with the following keys - - - *nqubits* - - *accelerators* - - *density_matrix* - - *wire_names*. - - queue (_Queue): List that holds the queue of gates of a circuit. - parametrized_gates (_ParametrizedGates): List of parametric gates. - trainable_gates (_ParametrizedGates): List of trainable gates. - measurements (list): List of non-collapsible measurements. - _final_state : Final result after full simulation of the circuit. - compiled (CompiledExecutor): Circuit executor. Defaults to ``None``. - repeated_execution (bool): If `True`, the circuit would be re-executed when sampling. - Defaults to ``False``. - density_matrix (bool, optional): If `True`, the circuit would evolve density matrices. - If ``False``, defaults to statevector simulation. - Defaults to ``False``. - accelerators (dict, optional): Dictionary that maps device names to the number of times each - device will be used. Defaults to ``None``. - ndevices (int): Total number of devices. Defaults to ``None``. - nglobal (int): Base two logarithm of the number of devices. Defaults to ``None``. - nlocal (int): Total number of available qubits in each device. Defaults to ``None``. - queues (DistributedQueues): Gate queues for each accelerator device. - Defaults to ``None``. - """ - - def __init__( - self, - nqubits: Optional[Union[int, list]] = None, - accelerators: Optional[dict] = None, - density_matrix: bool = False, - wire_names: Optional[list] = None, - ): - nqubits, wire_names = _resolve_qubits(nqubits, wire_names) - self.nqubits = nqubits - self.init_kwargs = { - "nqubits": nqubits, - "accelerators": accelerators, - "density_matrix": density_matrix, - "wire_names": wire_names, - } - self.wire_names = wire_names - self.queue = _Queue(nqubits) - # Keep track of parametrized gates for the ``set_parameters`` method - self.parametrized_gates = _ParametrizedGates() - self.trainable_gates = _ParametrizedGates() - self.measurements = [] # list of non-collapsible measurements - - self._final_state = None - self.compiled = None - - self.has_collapse = False - self.has_unitary_channel = False - self.density_matrix = density_matrix - - # for distributed circuits - self.accelerators = accelerators - self.ndevices = None - self.nglobal = None - self.nlocal = None - self.queues = None - if accelerators: # pragma: no cover - if density_matrix: - raise_error( - NotImplementedError, - "Distributed circuit is not implemented for density matrices.", - ) - self._distributed_init(nqubits, accelerators) - - def _distributed_init(self, nqubits: int, accelerators) -> None: # pragma: no cover - """Distributed implementation of :class:`qibo.models.circuit.Circuit`. - - Uses multiple `accelerator` devices (GPUs) for applying gates to the state vector. - The full state vector is saved in the given `memory device` (usually the CPU) - during the simulation. A gate is applied by splitting the state to pieces - and copying each piece to an accelerator device that is used to perform the - matrix multiplication. An `accelerator` device can be used more than once - resulting to logical devices that are more than the physical accelerators in - the system. - - Distributed circuits currently do not support native tensorflow gates, - compilation and callbacks. - - Example: - .. code-block:: python - - from qibo import Circuit - # The system has two GPUs and we would like to use each GPU twice - # resulting to four total logical accelerators - accelerators = {'/GPU:0': 2, '/GPU:1': 2} - # Define a circuit on 32 qubits to be run in the above GPUs keeping - # the full state vector in the CPU memory. - circuit = Circuit(32, accelerators) - - Args: - nqubits (int): Total number of qubits in the circuit. - accelerators (dict): Dictionary that maps device names to the number of - times each device will be used. - The total number of logical devices must be a power of 2. - """ - from qibo.models.distcircuit import DistributedQueues # pylint: disable=C0415 - - self.ndevices = sum(accelerators.values()) - self.nglobal = float(np.log2(self.ndevices)) - if not (self.nglobal.is_integer() and self.nglobal > 0): - raise_error( - ValueError, - "Number of calculation devices should be a power " - + f"of 2 but is {self.ndevices}.", - ) - self.nglobal = int(self.nglobal) - self.nlocal = self.nqubits - self.nglobal - - self.queues = DistributedQueues(self) - - def __add__(self, circuit): - """Add circuits. - - Args: - circuit: Circuit to be added to the current one. - - Returns: - The resulting circuit from the addition. - """ - for k, kwarg1 in self.init_kwargs.items(): - kwarg2 = circuit.init_kwargs[k] - if kwarg1 != kwarg2: - raise_error( - ValueError, - "Cannot add circuits with different kwargs. " - + f"{k} is {kwarg1} for first circuit and {kwarg2} " - + "for the second.", - ) - - newcircuit = self.__class__(**self.init_kwargs) - # Add gates from `self` to `newcircuit` (including measurements) - for gate in self.queue: - newcircuit.add(gate) - # Add gates from `circuit` to `newcircuit` (including measurements) - for gate in circuit.queue: - newcircuit.add(gate) - - return newcircuit - - @property - def wire_names(self) -> Union[List[int], List[str]]: - if self._wire_names is None: - return list(range(self.nqubits)) - return self._wire_names - - @wire_names.setter - def wire_names(self, wire_names: Optional[list]) -> None: - if not isinstance(wire_names, (list, type(None))): - raise_error( - TypeError, - f"``wire_names`` must be type ``list``, but is {type(wire_names)}.", - ) - - if wire_names is not None: - if len(wire_names) != self.nqubits: - raise_error( - ValueError, - f"Number of wire names must be equal to the number of qubits ({self.nqubits}), " - f"but is {len(wire_names)}.", - ) - self._wire_names = wire_names.copy() - else: - self._wire_names = None - self.init_kwargs["wire_names"] = self._wire_names - - @property - def repeated_execution(self) -> bool: - return self.has_collapse or ( - self.has_unitary_channel and not self.density_matrix - ) - - def on_qubits(self, *qubits): - """Generator of gates contained in the circuit acting on specified - qubits. - - Useful for adding a circuit as a subroutine in a larger circuit. - - Args: - qubits (int): Qubit ids that the gates should act. - - Example: - - .. testcode:: - - from qibo import Circuit, gates - - # create small circuit on 4 qubits - small_circuit = Circuit(4) - small_circuit.add(gates.RX(qubit, theta=0.1) for qubit in range(4)) - small_circuit.add((gates.CNOT(0, 1), gates.CNOT(2, 3))) - # create large circuit on 8 qubits - large_circuit = Circuit(8) - large_circuit.add(gates.RY(qubit, theta=0.1) for qubit in range(8)) - # add the small circuit to the even qubits of the large one - large_circuit.add(small_circuit.on_qubits(*range(0, 8, 2))) - """ - if len(qubits) != self.nqubits: - raise_error( - ValueError, - f"Cannot return gates on {len(qubits)} qubits because " - + f"the circuit contains {self.nqubits} qubits.", - ) - if self.accelerators and self.queues.queues: # pragma: no cover - raise_error( - RuntimeError, - "Cannot use distributed circuit as a subroutine after it was executed.", - ) - - qubit_map = dict(enumerate(qubits)) - for gate in self.queue: - yield gate.on_qubits(qubit_map) - - def light_cone(self, *qubits): - """Reduces circuit to the qubits relevant for an observable. - - Useful for calculating expectation values of local observables without - requiring simulation of large circuits. - Uses the light cone construction described in - `issue #571 `_. - - Args: - qubits (int): Qubit ids that the observable has support on. - - Returns: - circuit (:class:`qibo.models.Circuit`): Circuit that contains only - the qubits that are required for calculating expectation - involving the given observable qubits. - qubit_map (dict): Dictionary mapping the qubit ids of the original - circuit to the ids in the new one. - """ - # original qubits that are in the light cone - qubits = set(qubits) - # original gates that are in the light cone - list_of_gates = [] - for gate in reversed(self.queue): - gate_qubits = set(gate.qubits) - if gate_qubits & qubits: - # if the gate involves any qubit included in the - # light cone, add all its qubits in the light cone - qubits |= gate_qubits - list_of_gates.append(gate) - - # Create a new circuit ignoring gates that are not in the light cone - qubit_map = {q: i for i, q in enumerate(sorted(qubits))} - kwargs = dict(self.init_kwargs) - kwargs["nqubits"] = len(qubits) - kwargs["wire_names"] = [self.wire_names[q] for q in sorted(qubits)] - circuit = self.__class__(**kwargs) - circuit.add(gate.on_qubits(qubit_map) for gate in reversed(list_of_gates)) - return circuit, qubit_map - - def _shallow_copy(self): - """Helper method for :meth:`qibo.models.circuit.Circuit.copy` and - :meth:`qibo.core.circuit.Circuit.fuse`.""" - new_circuit = self.__class__(**self.init_kwargs) - new_circuit.parametrized_gates = _ParametrizedGates(self.parametrized_gates) - new_circuit.trainable_gates = _ParametrizedGates(self.trainable_gates) - new_circuit.measurements = self.measurements - return new_circuit - - def copy(self, deep: bool = False): - """Creates a copy of the current ``circuit`` as a new ``Circuit`` - model. - - Args: - deep (bool): If ``True`` copies of the gate objects will be created - for the new circuit. If ``False``, the same gate objects of - ``circuit`` will be used. - - Returns: - The copied circuit object. - """ - if deep: - new_circuit = self.__class__(**self.init_kwargs) - for gate in self.queue: - if isinstance(gate, gates.FusedGate): # pragma: no cover - # impractical case - raise_error( - NotImplementedError, - "Cannot create deep copy of fused circuit.", - ) - if isinstance(gate, gates.M): - new_circuit.add(gate.__class__(*gate.init_args, **gate.init_kwargs)) - else: - new_circuit.add(copy.copy(gate)) - else: - if self.accelerators: # pragma: no cover - raise_error( - ValueError, - "Non-deep copy is not allowed for distributed " - "circuits because they modify gate objects.", - ) - new_circuit = self.__class__(**self.init_kwargs) - for gate in self.queue: - new_circuit.add(gate) - return new_circuit - - def invert(self): - """Creates a new ``Circuit`` that is the inverse of the original. - - Inversion is obtained by taking the dagger of all gates in reverse order. - If the original circuit contains parametrized gates, dagger will change - their parameters. This action is not persistent, so if the parameters - are updated afterwards, for example using - :meth:`qibo.models.circuit.Circuit.set_parameters`, the action of dagger - will be overwritten. If the original circuit contains measurement gates, - these are included in the inverted circuit. - - Returns: - :class:`qibo.models.Circuit`: Circuit corresponding to the inverse of - the original ``circuit``. - """ - skip_measurements = True - measurements = [] - new_circuit = self.__class__(**self.init_kwargs) - for gate in self.queue[::-1]: - if isinstance(gate, gates.Channel): - raise_error( - NotImplementedError, - "`invert` method not implemented for circuits that contain noise channels.", - ) - elif isinstance(gate, gates.M) and skip_measurements: - measurements.append(gate) - else: - new_gate = gate.dagger() - if isinstance(gate, ParametrizedGate): - new_gate.trainable = gate.trainable - new_circuit.add(new_gate) - skip_measurements = False - new_circuit.add(measurements[::-1]) - return new_circuit - - def _check_noise_map(self, noise_map: NoiseMapType) -> NoiseMapType: - if isinstance(noise_map, list) and not all( - isinstance(n, (tuple, list)) for n in noise_map - ): - raise_error( - TypeError, - f"Type {type(noise_map)} of noise map is not recognized.", - ) - elif isinstance(noise_map, dict): - if len(noise_map) != self.nqubits: - raise_error( - ValueError, - f"Noise map has {len(noise_map)} qubits while the circuit has {self.nqubits}.", - ) - - return noise_map - - return {q: noise_map for q in range(self.nqubits)} - - def decompose(self, *free: int, method: str = "standard", **kwargs): - """Decomposes circuit's gates to gates supported by OpenQASM. - - Args: - free (int): Ids of free (work) qubits to use for gate decomposition. - method (str, optional): Choice of gate set for the decomposition. - If ``"standard"``, decomposes circuit into :class:`qibo.gates.gates.CNOT`, - :class:`qibo.gates.gates.RX`, :class:`qibo.gates.gates.RY`, - :class:`qibo.gates.gates.RZ`, :class:`qibo.gates.gates.U1`, - :class:`qibo.gates.gates.U2`, :class:`qibo.gates.gates.U3`, - and Clifford gates. If ``"clifford_plus_t"``, decomposes the circuit - into :class:`qibo.gates.gates.CNOT`, :class:`qibo.gates.gates.H`, - :class:`qibo.gates.gates.S`, :class:`qibo.gates.gates.X`, - :class:`qibo.gates.gates.Y`, :class:`qibo.gates.gates.Z`, - and :class:`qibo.gates.gates.T`. Defaults to ``"standard"``. - - Returns: - :class:`qibo.models.circuit.Circuit`: Circuit that contains only - gates that are supported by OpenQASM and has the same effect as - the original circuit. - """ - # FIXME: This method is not completed until the ``decompose`` is - # implemented for all gates not supported by OpenQASM. - decomp_circuit = self.__class__(self.nqubits) - for gate in self.queue: - decomp_circuit.add(gate.decompose(*free, method=method, **kwargs)) - return decomp_circuit - - def with_pauli_noise(self, noise_map: NoiseMapType): - """Creates a copy of the circuit with Pauli noise gates after each - gate. - - If the original circuit uses state vectors then noise simulation will - be done using sampling and repeated circuit execution. - In order to use density matrices the original circuit should be created - setting the flag ``density_matrix=True``. - For more information we refer to the - :ref:`How to perform noisy simulation? ` example. - - Args: - noise_map (dict): list of tuples :math:`(P_{k}, p_{k})`, where - :math:`P_{k}` is a ``str`` representing the :math:`k`-th - :math:`n`-qubit Pauli operator, and :math:`p_{k}` is the - associated probability. - - Returns: - Circuit object that contains all the gates of the original circuit - and additional noise channels on all qubits after every gate. - - Example: - .. testcode:: - - from qibo import Circuit, gates - # use density matrices for noise simulation - circuit = Circuit(2, density_matrix=True) - circuit.add([gates.H(0), gates.H(1), gates.CNOT(0, 1)]) - noise_map = { - 0: list(zip(["X", "Z"], [0.1, 0.2])), - 1: list(zip(["Y", "Z"], [0.2, 0.1])) - } - noisy_circuit = circuit.with_pauli_noise(noise_map) - # ``noisy_circuit`` will be equivalent to the following circuit - circuit_2 = Circuit(2, density_matrix=True) - circuit_2.add(gates.H(0)) - circuit_2.add(gates.PauliNoiseChannel(0, [("X", 0.1), ("Z", 0.2)])) - circuit_2.add(gates.H(1)) - circuit_2.add(gates.PauliNoiseChannel(1, [("Y", 0.2), ("Z", 0.1)])) - circuit_2.add(gates.CNOT(0, 1)) - circuit_2.add(gates.PauliNoiseChannel(0, [("X", 0.1), ("Z", 0.2)])) - circuit_2.add(gates.PauliNoiseChannel(1, [("Y", 0.2), ("Z", 0.1)])) - """ - if self.accelerators: # pragma: no cover - raise_error( - NotImplementedError, - "Distributed circuit does not support density matrices yet.", - ) - - noise_map = self._check_noise_map(noise_map) - # Generate noise gates - noise_gates = [] - for gate in self.queue: - if isinstance(gate, gates.KrausChannel): - raise_error( - ValueError, - "`.with_pauli_noise` method is not available " - + "for circuits that already contain " - + "channels.", - ) - noise_gates.append([]) - if not isinstance(gate, gates.M): - for q in gate.qubits: - if q in noise_map and sum(row[1] for row in noise_map[q]) > 0: - noise_gates[-1].append(gates.PauliNoiseChannel(q, noise_map[q])) - - # Create new circuit with noise gates inside - noisy_circuit = self.__class__(**self.init_kwargs) - for i, gate in enumerate(self.queue): - noisy_circuit.add(gate) - for noise_gate in noise_gates[i]: - noisy_circuit.add(noise_gate) - return noisy_circuit - - def add(self, gate: Gate): - """Add a gate to a given queue. - - Args: - gate (:class:`qibo.gates.Gate`): the gate object to add. - See :ref:`Gates` for a list of available gates. - `gate` can also be an iterable or generator of gates. - In this case all gates in the iterable will be added in the - circuit. - - Returns: - If the circuit contains measurement gates with ``collapse=True`` - a ``sympy.Symbol`` that parametrizes the corresponding outcome. - """ - if isinstance(gate, Iterable): - for g in gate: - self.add(g) - - else: - if self.accelerators: # pragma: no cover - if isinstance(gate, gates.KrausChannel): - raise_error( - NotImplementedError, - "Distributed circuits do not support channels.", - ) - elif self.nqubits - len( - gate.target_qubits - ) < self.nglobal and not isinstance(gate, gates.M): - # Check if there is sufficient number of local qubits - raise_error( - ValueError, - "Insufficient qubits to use for global in distributed circuit.", - ) - - if not isinstance(gate, gates.Gate): - raise_error(TypeError, f"Unknown gate type {type(gate)}.") - - if self._final_state is not None: - raise_error( - RuntimeError, - "Cannot add gates to a circuit after it is executed.", - ) - - for q in gate.target_qubits: - if q >= self.nqubits: - raise_error( - ValueError, - f"Attempting to add gate with target qubits {gate.target_qubits} " - + f"on a circuit of {self.nqubits} qubits.", - ) - - if isinstance(gate, gates.M): - # The following loop is useful when two circuits are added together: - # all the gates in the basis of the measure gates should not - # be added to the new circuit, otherwise once the measure gate - # is added in the circuit there will be two of the same. - - for base in gate.basis: - if base not in self.queue: - self.add(base) - - self.queue.append(gate) - if gate.register_name is None: - # add default register name - nreg = self.queue.nmeasurements - 1 - gate.register_name = f"register{nreg}" - else: - name = gate.register_name - for mgate in self.measurements: - if name == mgate.register_name: - raise_error( - KeyError, f"Register {name} already exists in circuit." - ) - - gate.result.circuit = self - if gate.collapse: - self.has_collapse = True - else: - self.measurements.append(gate) - - return gate.result - - self.queue.append(gate) - for measurement in list(self.measurements): - if set(measurement.qubits) & set(gate.qubits): - measurement.collapse = True - self.has_collapse = True - self.measurements.remove(measurement) - - if isinstance(gate, gates.UnitaryChannel): - self.has_unitary_channel = True - if isinstance(gate, gates.ParametrizedGate): - self.parametrized_gates.append(gate) - if gate.trainable: - self.trainable_gates.append(gate) - - @property - def measurement_tuples(self) -> dict: - """used for testing only""" - return {m.register_name: m.target_qubits for m in self.measurements} - - @property - def ngates(self) -> int: - """Total number of gates/operations in the circuit.""" - return len(self.queue) - - @property - def depth(self) -> int: - """Circuit depth if each gate is placed at the earliest possible - position.""" - return len(self.queue.moments) - - @property - def gate_types(self) -> Counter: - """``collections.Counter`` with the number of appearances of each gate - type.""" - gatecounter = Counter() - for gate in self.queue: - gatecounter[gate.__class__] += 1 - return gatecounter - - @property - def gate_names(self) -> Counter: - """``collections.Counter`` with the number of appearances of each gate - name.""" - gatecounter = Counter() - for gate in self.queue: - gatecounter[gate.name] += 1 - return gatecounter - - def gates_of_type(self, gate: Union[str, type]) -> List[Tuple[int, gates.Gate]]: - """Finds all gate objects of specific type or name. - - This method can be affected by how :meth:`qibo.gates.Gate.controlled_by` - behaves with certain gates. To see how :meth:`qibo.gates.Gate.controlled_by` - affects gates, we refer to the documentation of :meth:`qibo.gates.Gate.controlled_by`. - - Args: - gate (str or type): The name of a gate or the corresponding gate class. - - Returns: - list: gates that are in the circuit and have the same type as ``gate``. - The list contains tuples ``(k, g)`` where ``k`` is the index of the gate - ``g`` in the circuit's gate queue. - """ - gate_str = bool(isinstance(gate, str)) - gate_subclass = bool(isinstance(gate, type) and issubclass(gate, gates.Gate)) - - if not gate_str and not gate_subclass: - raise_error(TypeError, f"Gate identifier {gate} not recognized.") - - if gate_str: - return [(i, g) for i, g in enumerate(self.queue) if g.name == gate] - - return [(i, g) for i, g in enumerate(self.queue) if isinstance(g, gate)] - - def _set_parameters_list(self, parameters: ArrayLike, n: int, include_not_trainable: bool = False) -> None: - """Helper method for ``set_parameters`` when a list is given. - - Also works if ``parameters`` is ``np.ndarray`` or ``tf.Tensor``. - """ - _gates = ( - self.parametrized_gates if include_not_trainable else self.trainable_gates - ) - if n == len(_gates): - for i, gate in enumerate(_gates): - gate.parameters = parameters[i] - elif n == _gates.nparams: - parameters = list(parameters) - k = 0 - for i, gate in enumerate(_gates): - if gate.nparams == 1: - gate.parameters = parameters[i + k] - else: - gate.parameters = parameters[i + k : i + k + gate.nparams] - k += gate.nparams - 1 - else: - raise_error( - ValueError, - f"Given list of parameters has length {n} while " - + f"the circuit contains {len(_gates)} parametrized gates.", - ) - - def set_parameters( - self, parameters: Union[List[float], Dict[str, float], ArrayLike], include_not_trainable: bool = False - ) -> None: - """Updates the parameters of the circuit's parametrized gates. - - For more information on how to use this method we refer to the - :ref:`How to use parametrized gates?` example. - - Args: - parameters: Container holding the new parameter values. - It can have one of the following types: - List with length equal to the number of parametrized gates and - each of its elements compatible with the corresponding gate. - Dictionary with keys that are references to the parametrized - gates and values that correspond to the new parameters for - each gate. - Flat list with length equal to the total number of free - parameters in the circuit. - A backend supported tensor (for example ``np.ndarray`` or - ``tf.Tensor``) may also be given instead of a flat list. - - - Example: - .. testcode:: - - from qibo import Circuit, gates - # create a circuit with all parameters set to 0. - circuit = Circuit(3) - circuit.add(gates.RX(0, theta=0)) - circuit.add(gates.RY(1, theta=0)) - circuit.add(gates.CZ(1, 2)) - circuit.add(gates.fSim(0, 2, theta=0, phi=0)) - circuit.add(gates.H(2)) - - # set new values to the circuit's parameters using list - params = [0.123, 0.456, (0.789, 0.321)] - circuit.set_parameters(params) - # or using dictionary - params = { - circuit.queue[0]: 0.123, - circuit.queue[1]: 0.456, - circuit.queue[3]: (0.789, 0.321) - } - circuit.set_parameters(params) - # or using flat list (or an equivalent `np.array`/`tf.Tensor`/`torch.Tensor`) - params = [0.123, 0.456, 0.789, 0.321] - circuit.set_parameters(params) - """ - _gates = ( - self.parametrized_gates if include_not_trainable else self.trainable_gates - ) - # reset the final state - self._final_state = None - if isinstance(parameters, dict): - diff = set(parameters.keys()) - _gates.set - if diff: - raise_error( - KeyError, - f"Dictionary contains gates {diff} which are " - + "not on the list of parametrized gates of the circuit.", - ) - for gate, params in parameters.items(): - gate.parameters = params - elif isinstance(parameters, Iterable) and not isinstance( - parameters, (set, str) - ): - try: - nparams = int(parameters.shape[0]) - except AttributeError: - nparams = len(parameters) - self._set_parameters_list( - parameters, nparams, include_not_trainable=include_not_trainable - ) - else: - raise_error(TypeError, f"Invalid type of parameters {type(parameters)}.") - - def get_parameters( - self, output_format: str = "list", include_not_trainable: bool = False - ) -> Union[List, Dict]: # pylint: disable=W0622 - """Returns the parameters of all parametrized gates in the circuit. - - Inverse method of :meth:`qibo.models.circuit.Circuit.set_parameters`. - - Args: - output_format (str): Format to return the variational parameters. - Available formats are ``"list"``, ``"dict"`` and ``"flatlist"``. - See :meth:`qibo.models.circuit.Circuit.set_parameters` - for more details on each format. Default is ``"list"``. - include_not_trainable (bool): If ``True`` it includes the parameters - of non-trainable parametrized gates in the returned list or - dictionary. Default is ``False``. - """ - parametrized_gates = ( - self.parametrized_gates if include_not_trainable else self.trainable_gates - ) - - if output_format not in ["list", "dict", "flatlist"]: - raise_error( - ValueError, - f"Unknown format {output_format} given in ``get_parameters``.", - ) - - if output_format == "list": - params = [gate.parameters for gate in parametrized_gates] - elif output_format == "dict": - params = {gate: gate.parameters for gate in parametrized_gates} - else: - params = _get_parameters_flatlist(parametrized_gates) - - return params - - def associate_gates_with_parameters(self) -> List[Gate]: - """Associates to each parameter its gate. - - Returns: - A nparams-long flatlist whose i-th element is the gate parameterized - by the i-th parameter. - """ - - parameter_to_gate = [] - for gate in self.parametrized_gates: - npar = len(gate.parameters) - parameter_to_gate.extend([gate] * npar) - - return parameter_to_gate - - def summary_string(self) -> str: - """Generate a summary of the circuit.""" - logs = [ - f"Circuit depth = {self.depth}", - f"Total number of gates = {self.ngates}", - f"Number of qubits = {self.nqubits}", - "Most common gates:", - ] - common_gates = self.gate_names.most_common() - logs.extend(f"{g}: {n}" for g, n in common_gates) - return "\n".join(logs) - - def summary(self) -> None: - """Display a summary of the circuit. - - The summary contains the circuit depths, total number of qubits and - the all gates sorted in decreasing number of appearance. - - Example: - .. code-block:: python - from qibo import Circuit, gates - - circuit = Circuit(3) - circuit.add(gates.H(0)) - circuit.add(gates.H(1)) - circuit.add(gates.CNOT(0, 2)) - circuit.add(gates.CNOT(1, 2)) - circuit.add(gates.H(2)) - circuit.add(gates.TOFFOLI(0, 1, 2)) - - circuit.summary() - - # Prints - ''' - Circuit depth = 5 - Total number of gates = 6 - Number of qubits = 3 - Most common gates: - h: 3 - cx: 2 - ccx: 1 - ''' - """ - sys.stdout.write(self.summary_string() + "\n") - - def fuse(self, max_qubits: int = 2): - """Creates an equivalent circuit by fusing gates for increased - simulation performance. - - Args: - max_qubits (int): Maximum number of qubits in the fused gates. - - Returns: - A :class:`qibo.core.circuit.Circuit` object containing - :class:`qibo.gates.FusedGate` gates, each of which - corresponds to a group of some original gates. - For more details on the fusion algorithm we refer to the - :ref:`Circuit fusion ` section. - - Example: - .. testcode:: - - from qibo import Circuit, gates - - circuit = Circuit(2) - circuit.add([gates.H(0), gates.H(1)]) - circuit.add(gates.CNOT(0, 1)) - circuit.add([gates.Y(0), gates.Y(1)]) - # create circuit with fused gates - fused_circuit = circuit.fuse() - # now ``fused_circuit`` contains a single ``FusedGate`` that is - # equivalent to applying the five original gates - """ - if self.accelerators: # pragma: no cover - raise_error( - NotImplementedError, - "Fusion is not implemented for distributed circuits.", - ) - - queue = self.queue.to_fused() - for gate in queue: - if not gate.marked: - for q in gate.qubits: - # fuse nearest neighbors forth in time - neighbor = gate.right_neighbors.get(q) - if gate.can_fuse(neighbor, max_qubits): - gate.fuse(neighbor) - # fuse nearest neighbors back in time - neighbor = gate.left_neighbors.get(q) - if gate.can_fuse(neighbor, max_qubits): - neighbor.fuse(gate) - # create a circuit and assign the new queue - circuit = self._shallow_copy() - circuit.queue = queue.from_fused() - return circuit - - def unitary(self, backend: Optional[Backend] = None) -> ArrayLike: - """Creates the unitary matrix corresponding to all circuit gates. - - This is a :math:`2^{n} \\times 2^{n}`` matrix obtained by - multiplying all circuit gates, where :math:`n` is ``nqubits``. - """ - backend = _check_backend(backend) - - fgate = gates.FusedGate(*range(self.nqubits)) - for gate in self.queue: - if isinstance(gate, gates.Channel): - raise_error( - NotImplementedError, - "`unitary` method not implemented for circuits that contain noise channels.", - ) - elif not isinstance(gate, (gates.SpecialGate, gates.M)): - fgate.append(gate) - return fgate.matrix(backend) - - @property - def final_state(self) -> ArrayLike: - """Returns the final state after full simulation of the circuit. - - If the circuit is executed more than once, only the last final - state is returned. - """ - if self._final_state is None: - raise_error( - RuntimeError, - "Cannot access final state before the circuit is executed.", - ) - return self._final_state - - def compile(self, backend: Optional[Backend] = None) -> None: - if self.accelerators: # pragma: no cover - raise_error( - RuntimeError, "Cannot compile circuit that uses custom operators." - ) - - if self.compiled: - raise_error(RuntimeError, "Circuit is already compiled.") - - if not self.queue: - raise_error(RuntimeError, "Cannot compile circuit without gates.") - - for gate in self.queue: - if isinstance(gate, gates.CallbackGate): # pragma: no cover - raise_error( - NotImplementedError, - "Circuit compilation is not available with callbacks.", - ) - - backend = _check_backend(backend) - - executor = lambda state, nshots: backend.execute_circuit( - self, state, nshots - ).state() - self.compiled = type("CompiledExecutor", (), {})() - self.compiled.executor = backend.compile(executor) - if self.measurements: - self.compiled.result = lambda state, nshots: CircuitResult( - state, self.measurements, backend=backend, nshots=nshots - ) - else: - self.compiled.result = lambda state, nshots: QuantumState(state, backend) - - def execute(self, initial_state: ArrayLike = None, nshots: int = 1000, **kwargs): - """Executes the circuit. Exact implementation depends on the backend. - - Args: - initial_state (ndarray or :class:`qibo.models.circuit.Circuit`): - Initial configuration. It can be specified by the setting the state - vector using an array or a circuit. If ``None``, the initial state - is ``|000..00>``. - nshots (int, optional): Number of shots. Defaults ot :math:`1000`. - - Returns: - Either a ``qibo.result.QuantumState``, ``qibo.result.MeasurementOutcomes`` - or ``qibo.result.CircuitResult`` depending on the circuit's configuration. - """ - if self.compiled: - # pylint: disable=E1101 - state = self.compiled.executor(initial_state, nshots) - self._final_state = self.compiled.result(state, nshots) - return self._final_state - - backend = _Global.backend() - transpiler = _Global.transpiler() - transpiled_circuit, _ = transpiler(self) # pylint: disable=E1102 - - if self.accelerators: # pragma: no cover - return backend.execute_distributed_circuit( - transpiled_circuit, initial_state, nshots - ) - - args = [transpiled_circuit, initial_state, nshots] - - if backend.name == "hamming_weight": - weight = kwargs["weight"] - args = args[:1] + [weight] + args[1:] - - return backend.execute_circuit(*args) - - def __call__(self, initial_state: ArrayLike = None, nshots: int = 1000, **kwargs): - """Equivalent to ``circuit.execute``.""" - return self.execute(initial_state=initial_state, nshots=nshots, **kwargs) - - @property - def raw(self) -> dict: - """Serialize to dictionary. - - This is a thin wrapper over :meth:`Gate.raw`. - """ - return { - "queue": [gate.raw for gate in self.queue], - "nqubits": self.nqubits, - "density_matrix": self.density_matrix, - "wire_names": self.wire_names, - "qibo_version": __version__, - } - - @classmethod - def from_dict(cls, raw: dict): - """Load from serialization. - - Essentially the counter-part of :meth:`raw`. - """ - circuit = cls( - raw["nqubits"], - density_matrix=raw["density_matrix"], - wire_names=raw.get("wire_names"), - ) - - for gate in raw["queue"]: - circuit.add(Gate.from_dict(gate)) - - return circuit - - def to_qasm(self, extended_compatibility: bool = False) -> str: - """Convert circuit to a QASM string. - - Args: - extended_compatibility (bool): if ``True``, unrolls more exotic gates - in their decomposition by defining them as custom gates, - this increases the compatibility with other frameworks, - such as ``qiskit``. Defaults to ``False``. - - Returns: - str: The generated QASM. - - .. note:: - This method does not support multi-controlled gates - and gates with ``torch.Tensor`` as parameters. - """ - code = [f"// Generated by QIBO {__version__}"] - code += ["OPENQASM 2.0;"] - code += ['include "qelib1.inc";'] - code += [f"qreg q[{self.nqubits}];"] - - gate_definitions = [] - - # Set measurements - for register, qubits in self.measurement_tuples.items(): - if not register.islower(): - raise_error( - NameError, - "OpenQASM does not support capital letters in " - + f"register names but {register} was used", - ) - code.append(f"creg {register}[{len(qubits)}];") - - # Add gates - for gate in self.queue: - if isinstance(gate, gates.M): - continue - - if gate.is_controlled_by: - raise_error( - ValueError, "OpenQASM does not support multi-controlled gates." - ) - - qubits = ",".join(f"q[{i}]" for i in gate.qubits) - if isinstance(gate.qasm_label, tuple): - qasm_label, qasm_definition = gate.qasm_label - if qasm_definition not in gate_definitions: - gate_definitions.append(qasm_definition) - else: - qasm_label = gate.qasm_label - if isinstance(gate, gates.ParametrizedGate): - params = (str(float(x)) for x in gate.parameters) - name = f"{qasm_label}({', '.join(params)})" - else: - name = qasm_label - code.append(f"{name} {qubits};") - - if extended_compatibility: - code = code[:4] + gate_definitions + code[4:] - - # Add measurements - for register, qubits in self.measurement_tuples.items(): - for i, q in enumerate(qubits): - code.append(f"measure q[{q}] -> {register}[{i}];") - - return "\n".join(code) - - def to_qasm_file( - self, qasm_file: str, extended_compatibility: bool = False - ) -> None: - """_summary_ - - Args: - qasm_file (str): Path to file containing the QASM script. - extended_compatibility (bool, optional): if ``True``, unrolls more exotic - gates in their decomposition by defining them as custom gates, - this increases the compatibility with other frameworks, - such as ``qiskit``. Defaults to ``False``. - """ - string = self.to_qasm(extended_compatibility=extended_compatibility) - with open(qasm_file, "w") as file: # pylint: disable=W1514 - file.writelines(string) - - @classmethod - def from_qasm(cls, qasm_code: str, **circuit_kwargs): - """Constructs a circuit from QASM code. - - Args: - qasm_code (str): String with the QASM script. - circuit_kwargs (dict): Keyword arguments of the circuit. - - Returns: - :class:`qibo.models.circuit.Circuit`: Circuit containing the gates - specified by the given QASM script. - - Example: - - .. testcode:: - - from qibo import Circuit, gates - qasm_code = '''OPENQASM 2.0; - include "qelib1.inc"; - qreg q[2]; - h q[0]; - h q[1]; - cx q[0],q[1];''' - circuit = Circuit.from_qasm(qasm_code) - # is equivalent to creating the following circuit - circuit_2 = Circuit(2) - circuit_2.add(gates.H(0)) - circuit_2.add(gates.H(1)) - circuit_2.add(gates.CNOT(0, 1)) - """ - parser = QASMParser() - circuit_kwargs.pop("nqubits", None) - return parser.to_circuit(qasm_code, **circuit_kwargs) - - @classmethod - def from_qasm_file(cls, qasm_file: str, **circuit_kwargs): - """Constructs a circuit from QASM code in a ``qasm_file``. - - Args: - qasm_file (str): Path to file containing the string with the QASM script. - circuit_kwargs (dict): Keyword arguments of the circuit. - - Returns: - :class:`qibo.models.circuit.Circuit`: Circuit containing the gates - specified by the given QASM script. - """ - with open(qasm_file) as file: # pylint: disable=W1514 - string = file.read() - - return cls.from_qasm(string, **circuit_kwargs) - - def to_qir(self): - """ - Convert circuit to QIR circuit. - - Uses `qbraid` (https://github.com/qBraid/qBraid) to transpile - the circuit into `pyqir` circuits. - """ - try: - import qbraid_qir # pylint: disable=C0415, W0611 - from qbraid.transpiler.conversions.qasm2 import ( # pylint: disable=C0415 - qasm2_to_qasm3, - ) - from qbraid.transpiler.conversions.qasm3 import ( # pylint: disable=C0415 - qasm3_to_pyqir, - ) - except ModuleNotFoundError as e: # pragma: no cover - raise ModuleNotFoundError( - "The optional dependency qbraid is missing, please install it with `poetry install --extras qir`" - ) from e - return qasm3_to_pyqir(qasm2_to_qasm3(self.to_qasm())) - - def to_cudaq(self): # pragma: no cover - """Convert circuit to CUDA-Q (quake) code. - - Uses `qbraid` (https://github.com/qBraid/qBraid) to transpile - the circuit into `cudaq` circuits. - """ - try: - from qbraid.transpiler.conversions.openqasm3 import ( # pylint: disable=C0415 - openqasm3_to_cudaq, - ) - from qbraid.transpiler.conversions.qasm2 import ( # pylint: disable=C0415 - qasm2_to_qasm3, - ) - except ModuleNotFoundError as e: - raise ModuleNotFoundError( - "The optional dependency qbraid is missing, " - "please install it with `poetry install --extras cudaq`" - ) from e - - try: - import cudaq # pylint: disable=C0415, W0611 - except ModuleNotFoundError as e: - raise ModuleNotFoundError( - "``cudaq`` is not installed, please install it with `pip install cudaq`." - ) from e - - return openqasm3_to_cudaq(qasm2_to_qasm3(self.to_qasm())) - - @classmethod - def from_cudaq(cls, cudaq_circuit_code): # pragma: no cover - """Construct a circuit from CUDA-Q (quake) code. - - Uses `qbraid` (https://github.com/qBraid/qBraid) to transpile - the `cudaq` circuit into a `qibo.models.circuit.Circuit`. - - Args: - cudaq_circuit_code (PyKernel): kernel containing the - cudaq circuit code. - - Returns: - :class:`qibo.models.circuit.Circuit` representing the - given circuit. - """ - try: - from qbraid.transpiler.conversions.cudaq import ( # pylint: disable=C0415 - cudaq_to_qasm2, - ) - except ModuleNotFoundError as e: - raise ModuleNotFoundError( - "The optional dependency qbraid is missing, " - "please install it with `poetry install --extras cudaq`" - ) from e - - try: - import cudaq # pylint: disable=C0415, W0611 - except ModuleNotFoundError as e: - raise ModuleNotFoundError( - "``cudaq`` is not installed, please install it with `pip install cudaq`." - ) from e - - return cls.from_qasm(cudaq_to_qasm2(cudaq_circuit_code)) - - def _update_draw_matrix( - self, - matrix: ArrayLike, - idx: List[int], - gate: Gate, - gate_symbol: Optional[str] = None, - ) -> Tuple[ArrayLike, List[int]]: - """Helper method for :meth:`qibo.models.circuit.Circuit.draw`.""" - if gate_symbol is None: - if gate.draw_label: - gate_symbol = gate.draw_label - elif gate.name: - gate_symbol = gate.name[:4] - else: - raise_error( - NotImplementedError, - f"{gate.__class__.__name__} gate is not supported by `circuit.draw`", - ) - - if isinstance(gate, gates.CallbackGate): - targets = list(range(self.nqubits)) - else: - targets = list(gate.target_qubits) - controls = list(gate.control_qubits) - - # identify boundaries - qubits = targets + controls - qubits.sort() - min_qubits_id = qubits[0] - max_qubits_id = qubits[-1] - - # identify column - col = idx[targets[0]] if not controls and len(targets) == 1 else max(idx) - - # extend matrix - for iq in range(self.nqubits): - matrix[iq].extend((1 + col - len(matrix[iq])) * [""]) - - # fill - for iq in range(min_qubits_id, max_qubits_id + 1): - if iq in targets: - matrix[iq][col] = gate_symbol - elif iq in controls: - matrix[iq][col] = "o" - else: - matrix[iq][col] = "|" - - # update indexes - if not controls and len(targets) == 1: - idx[targets[0]] += 1 - else: - idx = [col + 1] * self.nqubits - - return matrix, idx - - def diagram(self, line_wrap: int = 70, legend: bool = False) -> str: - """Build the string representation of the circuit diagram.""" - # build string representation of gates - matrix = [[] for _ in range(self.nqubits)] - wire_names = [str(name) for name in self.wire_names] - idx = [0] * self.nqubits - - for gate in self.queue: - if isinstance(gate, gates.FusedGate): - # start fused gate - matrix, idx = self._update_draw_matrix(matrix, idx, gate, "[") - # draw gates contained in the fused gate - for subgate in gate.gates: - matrix, idx = self._update_draw_matrix(matrix, idx, subgate) - # end fused gate - matrix, idx = self._update_draw_matrix(matrix, idx, gate, "]") - else: - matrix, idx = self._update_draw_matrix(matrix, idx, gate) - - # Add some spacers - for col in range(len(matrix[0])): - maxlen = max(len(matrix[l][col]) for l in range(self.nqubits)) - for row in range(self.nqubits): - matrix[row][col] += "─" * (1 + maxlen - len(matrix[row][col])) - - # Print to terminal - max_name_len = max(len(name) for name in wire_names) - output = "" - for q in range(self.nqubits): - output += ( - wire_names[q] - + " " * (max_name_len - len(wire_names[q])) - + ": ─" - + "".join(matrix[q]) - + "\n" - ) - - # legend - if legend: - legend_rows = { - (gate.name, gate.draw_label) - for gate in self.queue - if isinstance(gate, (gates.SpecialGate, gates.Channel)) - } - - table = tabulate( - [list(l) for l in sorted(legend_rows)], - headers=["Gate", "Symbol"], - tablefmt="orgtbl", - ) - table = "\n Legend for callbacks and channels: \n" + table - - # line wrap - if line_wrap: - loutput = output.splitlines() - - def chunkstring(string, length): - nchunks = range(0, len(string), length) - return (string[elem : length + elem] for elem in nchunks), len(nchunks) - - for row in range(self.nqubits): - chunks, nchunks = chunkstring( - loutput[row][3 + max_name_len - 1 :], line_wrap - ) - if nchunks == 1: - loutput = None - break - for elem, c in enumerate(chunks): - loutput += ["" for _ in range(self.nqubits)] - suffix = " ...\n" - prefix = ( - wire_names[row] - + " " * (max_name_len - len(wire_names[row])) - + ": " - ) - if elem == 0: - prefix += " " * 4 - elif row == 0: - prefix = "\n" + prefix + "... " - else: - prefix += "... " - if elem == nchunks - 1: - suffix = "\n" - loutput[row + elem * self.nqubits] = prefix + c + suffix - if loutput is not None: - output = "".join(loutput) - - if legend: - output += table - - return output.rstrip("\n") - - def __str__(self): - return self.diagram() - - def draw(self, line_wrap: int = 70, legend: bool = False) -> None: - """Draw text circuit using unicode symbols. - - Args: - line_wrap (int, optional): maximum number of characters per line. This option - split the circuit text diagram in chunks of line_wrap characters. - Defaults to :math:`70`. - legend (bool, optional): If ``True`` prints a legend below the circuit for - callbacks and channels. Defaults to ``False``. - """ - sys.stdout.write(self.diagram(line_wrap, legend) + "\n") - - -def _resolve_qubits( - qubits: Union[int, List[int]], wire_names: Union[list, dict] -) -> Tuple[int, Union[list, dict]]: - """Parse the input arguments for defining a circuit. - - Allows the user to initialize the circuit as follows: - - Example: - .. code-block:: python - from qibo import Circuit - - circuit = Circuit(3) - circuit = Circuit(3, wire_names=["q0", "q1", "q2"]) - circuit = Circuit(["q0", "q1", "q2"]) - circuit = Circuit(wire_names=["q0", "q1", "q2"]) - """ - no_qubits_yes_wires = bool(qubits is None and wire_names is not None) - yes_qubits_no_wires = bool(qubits is not None and wire_names is None) - yes_qubits_yes_wires = bool(qubits is not None and wire_names is not None) - - if not no_qubits_yes_wires and not yes_qubits_no_wires and not yes_qubits_yes_wires: - raise_error( - ValueError, - "Invalid input arguments for defining a circuit.", - ) - - if yes_qubits_no_wires: - if isinstance(qubits, int) and qubits > 0: - return qubits, None - - if isinstance(qubits, list): - return len(qubits), qubits - - if yes_qubits_yes_wires: - if isinstance(qubits, int) and isinstance(wire_names, list): - if qubits == len(wire_names): - return qubits, wire_names - - raise_error( - ValueError, - "Invalid input arguments for defining a circuit.", - ) - - return len(wire_names), wire_names - - -def _get_parameters_flatlist(parametrized_gates: List[Gate]): - params = [] - for gate in parametrized_gates: - gparams = gate.parameters - if len(gparams) == 1: - gparams = gparams[0] - - if isinstance(gparams, Iterable): - if not isinstance(gparams, (list, tuple)): - # necessary for 0-dimensional tensors - gparams = [gparams] if len(gparams.shape) == 0 else gparams.flatten() - params.extend(list(gparams)) - else: - params.append(gparams) - - return params diff --git a/src/qibo/models/.#circuit.py b/src/qibo/models/.#circuit.py deleted file mode 120000 index a97b3dc0d4..0000000000 --- a/src/qibo/models/.#circuit.py +++ /dev/null @@ -1 +0,0 @@ -andrea@ubuntu-desktop.33596:1778570606 \ No newline at end of file diff --git a/src/qibo/models/circuit.py b/src/qibo/models/circuit.py index f2bf9b5a8a..39543dfeb6 100644 --- a/src/qibo/models/circuit.py +++ b/src/qibo/models/circuit.py @@ -761,7 +761,9 @@ def gates_of_type(self, gate: Union[str, type]) -> List[Tuple[int, gates.Gate]]: return [(i, g) for i, g in enumerate(self.queue) if isinstance(g, gate)] - def _set_parameters_list(self, parameters: ArrayLike, n: int, include_not_trainable: bool = False) -> None: + def _set_parameters_list( + self, parameters: ArrayLike, n: int, include_not_trainable: bool = False + ) -> None: """Helper method for ``set_parameters`` when a list is given. Also works if ``parameters`` is ``np.ndarray`` or ``tf.Tensor``. @@ -789,7 +791,9 @@ def _set_parameters_list(self, parameters: ArrayLike, n: int, include_not_traina ) def set_parameters( - self, parameters: Union[List[float], Dict[str, float], ArrayLike], include_not_trainable: bool = False + self, + parameters: Union[List[float], Dict[str, float], ArrayLike], + include_not_trainable: bool = False, ) -> None: """Updates the parameters of the circuit's parametrized gates. @@ -836,14 +840,11 @@ def set_parameters( params = [0.123, 0.456, 0.789, 0.321] circuit.set_parameters(params) """ -<<<<<<< HEAD _gates = ( self.parametrized_gates if include_not_trainable else self.trainable_gates ) -======= # reset the final state self._final_state = None ->>>>>>> master if isinstance(parameters, dict): diff = set(parameters.keys()) - _gates.set if diff: