Coverage for qml_essentials/model.py: 92%
439 statements
« prev ^ index » next coverage.py v7.9.2, created at 2025-10-02 13:10 +0000
1from typing import Dict, Optional, Tuple, Callable, Union, List
2import pennylane as qml
3import pennylane.numpy as np
4import hashlib
5import os
6import warnings
7from autograd.numpy import numpy_boxes
8from copy import deepcopy
9import math
11from qml_essentials.ansaetze import Gates, Ansaetze, Circuit
12from qml_essentials.ansaetze import PulseInformation as pinfo
13from qml_essentials.utils import PauliCircuit, QuanTikz, MultiprocessingPool
15import logging
17log = logging.getLogger(__name__)
20class Model:
21 """
22 A quantum circuit model.
23 """
25 lightning_threshold = 12
26 cpu_scaler = 0.9 # default cpu scaler, =1 means full CPU for MP
28 def __init__(
29 self,
30 n_qubits: int,
31 n_layers: int,
32 circuit_type: Union[str, Circuit] = "No_Ansatz",
33 data_reupload: Union[bool, List[bool], List[List[bool]]] = True,
34 state_preparation: Union[str, Callable, List[str], List[Callable]] = None,
35 encoding: Union[str, Callable, List[str], List[Callable]] = Gates.RX,
36 trainable_frequencies: bool = False,
37 initialization: str = "random",
38 initialization_domain: List[float] = [0, 2 * np.pi],
39 output_qubit: Union[List[int], int] = -1,
40 shots: Optional[int] = None,
41 random_seed: int = 1000,
42 as_pauli_circuit: bool = False,
43 remove_zero_encoding: bool = True,
44 mp_threshold: int = -1,
45 ) -> None:
46 """
47 Initialize the quantum circuit model.
48 Parameters will have the shape [impl_n_layers, parameters_per_layer]
49 where impl_n_layers is the number of layers provided and added by one
50 depending if data_reupload is True and parameters_per_layer is given by
51 the chosen ansatz.
53 The model is initialized with the following parameters as defaults:
54 - noise_params: None
55 - execution_type: "expval"
56 - shots: None
58 Args:
59 n_qubits (int): The number of qubits in the circuit.
60 n_layers (int): The number of layers in the circuit.
61 circuit_type (str, Circuit): The type of quantum circuit to use.
62 If None, defaults to "no_ansatz".
63 data_reupload (Union[bool, List[bool], List[List[bool]]], optional):
64 Whether to reupload data to the quantum device on each
65 layer and qubit. Detailed re-uploading instructions can be given
66 as a list/array of 0/False and 1/True with shape (n_qubits,
67 n_layers) to specify where to upload the data. Defaults to True
68 for applying data re-uploading to the full circuit.
69 encoding (Union[str, Callable, List[str], List[Callable]], optional):
70 The unitary to use for encoding the input data. Can be a string
71 (e.g. "RX") or a callable (e.g. qml.RX). Defaults to qml.RX.
72 If input is multidimensional it is assumed to be a list of
73 unitaries or a list of strings.
74 trainable_frequencies (bool, optional):
75 Sets trainable encoding parameters for trainable frequencies.
76 Defaults to False.
77 initialization (str, optional): The strategy to initialize the parameters.
78 Can be "random", "zeros", "zero-controlled", "pi", or "pi-controlled".
79 Defaults to "random".
80 output_qubit (List[int], int, optional): The index of the output
81 qubit (or qubits). When set to -1 all qubits are measured, or a
82 global measurement is conducted, depending on the execution
83 type.
84 shots (Optional[int], optional): The number of shots to use for
85 the quantum device. Defaults to None.
86 random_seed (int, optional): seed for the random number generator
87 in initialization is "random" and for random noise parameters.
88 Defaults to 1000.
89 as_pauli_circuit (bool, optional): whether the circuit is
90 transformed to a Pauli-Clifford circuit as described by Nemkov
91 et al. (https://doi.org/10.1103/PhysRevA.108.032406), which is
92 required for analytical Fourier coefficient computation.
93 Defaults to False.
94 remove_zero_encoding (bool, optional): whether to
95 remove the zero encoding from the circuit. Defaults to True.
96 mp_threshold (int, optional): threshold above which the parameter
97 batch dimension is split across multiple processes.
98 Defaults to -1.
100 Returns:
101 None
102 """
103 # Initialize default parameters needed for circuit evaluation
104 self.noise_params: Optional[Dict[str, Union[float, Dict[str, float]]]] = None
105 self.execution_type: Optional[str] = "expval"
106 self.shots = shots
107 self.remove_zero_encoding = remove_zero_encoding
108 self.mp_threshold = mp_threshold
109 self.n_qubits: int = n_qubits
110 self.n_layers: int = n_layers
111 self.trainable_frequencies: bool = trainable_frequencies
113 if isinstance(output_qubit, list):
114 assert (
115 len(output_qubit) <= n_qubits
116 ), f"Size of output_qubit {len(output_qubit)} cannot be\
117 larger than number of qubits {n_qubits}."
118 self.output_qubit: Union[List[int], int] = output_qubit
120 # Initialize rng in Gates
121 Gates.init_rng(random_seed)
123 # --- State Preparation ---
124 # first check if we have a str, list or callable
125 if isinstance(state_preparation, str):
126 # if str, use the pennylane fct
127 self._sp = [getattr(Gates, f"{state_preparation}")]
128 elif isinstance(state_preparation, list):
129 # if list, check if str or callable
130 if isinstance(state_preparation[0], str):
131 self._sp = [getattr(Gates, f"{sp}") for sp in state_preparation]
132 else:
133 self._sp = state_preparation
134 elif state_preparation is None:
135 self._sp = [lambda *args, **kwargs: None]
136 else:
137 # default to callable
138 self._sp = [state_preparation]
140 # prepare corresponding pulse parameters (always optimized pulses)
141 self.sp_pulse_params = []
142 for sp in self._sp:
143 sp_name = sp.__name__ if hasattr(sp, "__name__") else str(sp)
145 if sp_name in pinfo.OPTIMIZED_PULSES:
146 params = np.array(pinfo.optimized_params(sp_name), requires_grad=False)
147 self.sp_pulse_params.append(params)
148 else:
149 # gate has no pulse parametrization
150 self.sp_pulse_params.append(None)
152 # --- Encoding ---
153 # first check if we have a str, list or callable
154 if isinstance(encoding, str):
155 # if str, use the pennylane fct
156 self._enc = [getattr(Gates, f"{encoding}")]
157 elif isinstance(encoding, list):
158 # if list, check if str or callable
159 if isinstance(encoding[0], str):
160 self._enc = [getattr(Gates, f"{enc}") for enc in encoding]
161 else:
162 self._enc = encoding
163 else:
164 # default to callable
165 self._enc = [encoding]
167 # Number of possible inputs
168 self.n_input_feat = len(self._enc)
169 log.info(f"Number of input features: {self.n_input_feat}")
171 # Trainable frequencies, default initialization as in arXiv:2309.03279v2
172 self.enc_params = np.ones(
173 (self.n_qubits, self.n_input_feat), requires_grad=trainable_frequencies
174 )
176 # --- Data-Reuploading ---
177 # Process data reuploading strategy and set degree
178 if not isinstance(data_reupload, bool):
179 if not isinstance(data_reupload, np.ndarray):
180 data_reupload = np.array(data_reupload)
181 if data_reupload.shape == (
182 n_layers,
183 n_qubits,
184 ):
185 data_reupload = data_reupload.reshape(*data_reupload.shape, 1)
186 data_reupload = np.repeat(data_reupload, self.n_input_feat, axis=2)
188 assert data_reupload.shape == (
189 n_layers,
190 n_qubits,
191 self.n_input_feat,
192 ), f"Data reuploading array has wrong shape. \
193 Expected {(n_layers, n_qubits)} or\
194 {(n_layers, n_qubits, self.n_input_feat)},\
195 got {data_reupload.shape}."
197 log.debug(f"Data reuploading array:\n{data_reupload}")
198 else:
199 if data_reupload:
200 impl_n_layers: int = (
201 n_layers + 1
202 ) # we need L+1 according to Schuld et al.
203 data_reupload = np.ones((n_layers, n_qubits, self.n_input_feat))
204 log.debug("Full data reuploading.")
205 else:
206 impl_n_layers: int = n_layers
207 data_reupload = np.zeros((n_layers, n_qubits, self.n_input_feat))
208 data_reupload[0][0] = 1
209 log.debug("No data reuploading.")
211 # convert to boolean values
212 self.data_reupload = data_reupload.astype(bool)
213 self.frequencies = [
214 np.count_nonzero(self.data_reupload[..., i])
215 for i in range(self.n_input_feat)
216 ]
218 if self.degree > 1:
219 impl_n_layers: int = n_layers + 1 # we need L+1 according to Schuld et al.
220 else:
221 impl_n_layers = n_layers
222 log.info(f"Number of implicit layers: {impl_n_layers}.")
224 # --- Ansatz ---
225 # only weak check for str. We trust the user to provide sth useful
226 if isinstance(circuit_type, str):
227 self.pqc: Callable[[Optional[np.ndarray], int], int] = getattr(
228 Ansaetze, circuit_type or "No_Ansatz"
229 )()
230 else:
231 self.pqc = circuit_type()
232 log.info(f"Using Ansatz {circuit_type}.")
234 # calculate the shape of the parameter vector here, we will re-use this in init.
235 params_per_layer = self.pqc.n_params_per_layer(self.n_qubits)
236 self._params_shape: Tuple[int, int] = (impl_n_layers, params_per_layer)
237 log.info(f"Parameters per layer: {params_per_layer}")
239 pulse_params_per_layer = self.pqc.n_pulse_params_per_layer(self.n_qubits)
240 self._pulse_params_shape: Tuple[int, int] = (
241 impl_n_layers,
242 pulse_params_per_layer,
243 )
245 self.batch_shape = (1, 1)
246 # this will also be re-used in the init method,
247 # however, only if nothing is provided
248 self._inialization_strategy = initialization
249 self._initialization_domain = initialization_domain
251 # ..here! where we only require a rng
252 self.initialize_params(np.random.default_rng(random_seed))
254 # Initialize two circuits, one with the default device and
255 # one with the mixed device
256 # which allows us to later route depending on the state_vector flag
257 self.as_pauli_circuit = as_pauli_circuit
259 self.circuit_mixed: qml.QNode = qml.QNode(
260 self._circuit,
261 qml.device("default.mixed", shots=self.shots, wires=self.n_qubits),
262 interface="autograd" if self.shots is not None else "auto",
263 diff_method="parameter-shift" if self.shots is not None else "best",
264 )
266 @property
267 def degree(self):
268 return max(self.frequencies)
270 @property
271 def as_pauli_circuit(self) -> bool:
272 return self._as_pauli_circuit
274 @as_pauli_circuit.setter
275 def as_pauli_circuit(self, value: bool) -> None:
276 self._as_pauli_circuit = value
278 if self.n_qubits < self.lightning_threshold:
279 device = "default.qubit"
280 else:
281 device = "lightning.qubit"
282 self.mp_threshold = -1
284 self.circuit: qml.QNode = qml.QNode(
285 self._circuit,
286 qml.device(
287 device,
288 shots=self.shots,
289 wires=self.n_qubits,
290 ),
291 interface="autograd" if self.shots is not None else "auto",
292 diff_method="parameter-shift" if self.shots is not None else "best",
293 )
295 if value:
296 pauli_circuit_transform = qml.transform(
297 PauliCircuit.from_parameterised_circuit
298 )
299 self.circuit = pauli_circuit_transform(self.circuit)
301 @property
302 def noise_params(self) -> Optional[Dict[str, Union[float, Dict[str, float]]]]:
303 """
304 Gets the noise parameters of the model.
306 Returns:
307 Optional[Dict[str, float]]: A dictionary of
308 noise parameters or None if not set.
309 """
310 return self._noise_params
312 @noise_params.setter
313 def noise_params(
314 self, kvs: Optional[Dict[str, Union[float, Dict[str, float]]]]
315 ) -> None:
316 """
317 Sets the noise parameters of the model.
319 Typically a "noise parameter" refers to the error probability.
320 ThermalRelaxation is a special case, and supports a dict as value with
321 structure:
322 "ThermalRelaxation":
323 {
324 "t1": 2000, # relative t1 time.
325 "t2": 1000, # relative t2 time
326 "t_factor" 1: # relative gate time factor
327 },
329 Args:
330 kvs (Optional[Dict[str, Union[float, Dict[str, float]]]]): A
331 dictionary of noise parameters. If all values are 0.0, the noise
332 parameters are set to None.
334 Returns:
335 None
336 """
337 # set to None if only zero values provided
338 if kvs is not None and all(np == 0.0 for np in kvs.values()):
339 kvs = None
341 # set default values
342 if kvs is not None:
343 kvs.setdefault("BitFlip", 0.0)
344 kvs.setdefault("PhaseFlip", 0.0)
345 kvs.setdefault("Depolarizing", 0.0)
346 kvs.setdefault("MultiQubitDepolarizing", 0.0)
347 kvs.setdefault("AmplitudeDamping", 0.0)
348 kvs.setdefault("PhaseDamping", 0.0)
349 kvs.setdefault("GateError", 0.0)
350 kvs.setdefault("ThermalRelaxation", None)
351 kvs.setdefault("StatePreparation", 0.0)
352 kvs.setdefault("Measurement", 0.0)
354 # check if there are any keys not supported
355 for key in kvs.keys():
356 if key not in [
357 "BitFlip",
358 "PhaseFlip",
359 "Depolarizing",
360 "MultiQubitDepolarizing",
361 "AmplitudeDamping",
362 "PhaseDamping",
363 "GateError",
364 "ThermalRelaxation",
365 "StatePreparation",
366 "Measurement",
367 ]:
368 warnings.warn(
369 f"Noise type {key} is not supported by this package",
370 UserWarning,
371 )
373 # check valid params for thermal relaxation noise channel
374 tr_params = kvs["ThermalRelaxation"]
375 if isinstance(tr_params, dict):
376 tr_params.setdefault("t1", 0.0)
377 tr_params.setdefault("t2", 0.0)
378 tr_params.setdefault("t_factor", 0.0)
379 for k in tr_params.keys():
380 if k not in [
381 "t1",
382 "t2",
383 "t_factor",
384 ]:
385 warnings.warn(
386 f"Thermal Relaxation parameter {k} is not supported "
387 f"by this package",
388 UserWarning,
389 )
390 if not all(tr_params.values()) or tr_params["t2"] > 2 * tr_params["t1"]:
391 warnings.warn(
392 "Received invalid values for Thermal Relaxation noise "
393 "parameter. Thermal relaxation is not applied!",
394 UserWarning,
395 )
396 kvs["ThermalRelaxation"] = 0.0
398 self._noise_params = kvs
400 @property
401 def execution_type(self) -> str:
402 """
403 Gets the execution type of the model.
405 Returns:
406 str: The execution type, one of 'density', 'expval', or 'probs'.
407 """
408 return self._execution_type
410 @execution_type.setter
411 def execution_type(self, value: str) -> None:
412 if value not in ["density", "state", "expval", "probs"]:
413 raise ValueError(f"Invalid execution type: {value}.")
415 if (value == "density" or value == "state") and self.output_qubit != -1:
416 warnings.warn(
417 f"{value} measurement does ignore output_qubit, which is "
418 f"{self.output_qubit}.",
419 UserWarning,
420 )
422 if value == "probs" and self.shots is None:
423 warnings.warn(
424 "Setting execution_type to probs without specifying shots.",
425 UserWarning,
426 )
428 if value == "density" and self.shots is not None:
429 warnings.warn(
430 "Setting execution_type to density with specified shots.",
431 UserWarning,
432 )
434 self._execution_type = value
436 @property
437 def shots(self) -> Optional[int]:
438 """
439 Gets the number of shots to use for the quantum device.
441 Returns:
442 Optional[int]: The number of shots.
443 """
444 return self._shots
446 @shots.setter
447 def shots(self, value: Optional[int]) -> None:
448 """
449 Sets the number of shots to use for the quantum device.
451 Args:
452 value (Optional[int]): The number of shots.
453 If an integer less than or equal to 0 is provided, it is set to None.
455 Returns:
456 None
457 """
458 if type(value) is int and value <= 0:
459 value = None
460 self._shots = value
462 def initialize_params(
463 self,
464 rng: np.random.Generator,
465 repeat: int = None,
466 initialization: str = None,
467 initialization_domain: List[float] = None,
468 ) -> None:
469 """
470 Initializes the parameters of the model.
472 Args:
473 rng: A random number generator to use for initialization.
474 repeat: The number of times to repeat the parameters.
475 If None, the number of layers is used.
476 initialization: The strategy to use for parameter initialization.
477 If None, the strategy specified in the constructor is used.
478 initialization_domain: The domain to use for parameter initialization.
479 If None, the domain specified in the constructor is used.
481 Returns:
482 None
483 """
484 # Initializing params
485 params_shape = (
486 self._params_shape if repeat is None else [*self._params_shape, repeat]
487 )
488 # use existing strategy if not specified
489 initialization = initialization or self._inialization_strategy
490 initialization_domain = initialization_domain or self._initialization_domain
492 def set_control_params(params: np.ndarray, value: float) -> np.ndarray:
493 indices = self.pqc.get_control_indices(self.n_qubits)
494 if indices is None:
495 warnings.warn(
496 f"Specified {initialization} but circuit\
497 does not contain controlled rotation gates.\
498 Parameters are intialized randomly.",
499 UserWarning,
500 )
501 else:
502 params[:, indices[0] : indices[1] : indices[2]] = (
503 np.ones_like(params[:, indices[0] : indices[1] : indices[2]])
504 * value
505 )
506 return params
508 if initialization == "random":
509 self.params: np.ndarray = rng.uniform(
510 *initialization_domain, params_shape, requires_grad=True
511 )
512 elif initialization == "zeros":
513 self.params: np.ndarray = np.zeros(params_shape, requires_grad=True)
514 elif initialization == "pi":
515 self.params: np.ndarray = np.ones(params_shape, requires_grad=True) * np.pi
516 elif initialization == "zero-controlled":
517 self.params: np.ndarray = rng.uniform(
518 *initialization_domain, params_shape, requires_grad=True
519 )
520 self.params = set_control_params(self.params, 0)
521 elif initialization == "pi-controlled":
522 self.params: np.ndarray = rng.uniform(
523 *initialization_domain, params_shape, requires_grad=True
524 )
525 self.params = set_control_params(self.params, np.pi)
526 else:
527 raise Exception("Invalid initialization method")
529 log.info(
530 f"Initialized parameters with shape {self.params.shape}\
531 using strategy {initialization}."
532 )
534 # Initializing pulse params
535 shape = (
536 self._pulse_params_shape
537 if repeat is None
538 else (*self._pulse_params_shape, repeat)
539 )
540 self.pulse_params: np.ndarray = np.ones(shape, requires_grad=False)
542 log.info(f"Initialized pulse parameters with shape {self.pulse_params.shape}.")
544 def transform_input(self, inputs: np.ndarray, enc_params: Optional[np.ndarray]):
545 """
546 Transforms the input as in arXiv:2309.03279v2
548 Args:
549 inputs (np.ndarray): single input point of shape (1, n_input_feat)
550 idx (int): feature index
551 qubit (int): qubit on which to the encoding is being performed
552 enc_params (np.ndarray): encoding weight vector of
553 shape (n_qubits)
555 Returns:
556 np.ndarray: transformed input of shape (1,), linearly scaled by
557 enc_params, ready for encoding
558 """
559 return inputs * enc_params
561 def _iec(
562 self,
563 inputs: np.ndarray,
564 data_reupload: np.ndarray,
565 enc: Union[Callable, List[Callable]],
566 enc_params: np.ndarray,
567 noise_params: Optional[Dict[str, Union[float, Dict[str, float]]]] = None,
568 ) -> None:
569 """
570 Creates an AngleEncoding using RX gates
572 Args:
573 inputs (np.ndarray): single input point of shape (1, n_input_feat)
574 data_reupload (np.ndarray): Boolean array to indicate positions in
575 the circuit for data re-uploading for the IEC, shape is
576 (n_qubits, n_layers).
577 enc: Callable or List[Callable]: encoding function or list of encoding
578 functions
579 enc_params (np.ndarray): encoding weight vector
580 of shape [n_qubits, n_inputs]
581 noise_params (Optional[Dict[str, Union[float, Dict[str, float]]]]):
582 The noise parameters.
583 Returns:
584 None
585 """
586 # check for zero, because due to input validation, input cannot be none
587 if self.remove_zero_encoding and not inputs.any():
588 return
590 for q in range(self.n_qubits):
591 for idx in range(inputs.shape[1]):
592 if data_reupload[q, idx]:
593 enc[idx](
594 self.transform_input(inputs[:, idx], enc_params[q, idx]),
595 wires=q,
596 noise_params=noise_params,
597 )
599 def _circuit(
600 self,
601 params: np.ndarray,
602 inputs: np.ndarray,
603 pulse_params: np.ndarray = None,
604 enc_params: Optional[np.ndarray] = None,
605 gate_mode: str = "unitary",
606 ) -> Union[float, np.ndarray]:
607 # TODO: Is the shape of params below correct?
608 """
609 Creates a quantum circuit, optionally with noise or pulse simulation.
611 Args:
612 params (np.ndarray): weight vector of shape
613 [n_layers, n_qubits*(n_params_per_layer+trainable_frequencies)]
614 inputs (np.ndarray): input vector of size 1
615 pulse_params Optional[np.ndarray]: pulse parameter scaler weights of shape
616 [n_layers, n_pulse_params_per_layer]
617 enc_params Optional[np.ndarray]: encoding weight vector
618 of shape [n_qubits, n_inputs]
619 gate_mode (str): Backend mode for gate execution. Can be
620 "unitary" (default) or "pulse".
621 Returns:
622 Union[float, np.ndarray]: Expectation value of PauliZ(0)
623 of the circuit if state_vector is False and expval is True,
624 otherwise the density matrix of all qubits.
625 """
627 self._variational(
628 params=params,
629 inputs=inputs,
630 pulse_params=pulse_params,
631 enc_params=enc_params,
632 gate_mode=gate_mode,
633 )
634 return self._observable()
636 def _variational(
637 self,
638 params: np.ndarray,
639 inputs: np.ndarray,
640 pulse_params: Optional[np.ndarray] = None,
641 enc_params: Optional[np.ndarray] = None,
642 gate_mode: str = "unitary",
643 ) -> None:
644 """
645 Builds the variational quantum circuit with state preparation,
646 variational ansatz layers, and intertwined encoding layers.
648 Args:
649 params (np.ndarray): weight vector of shape
650 [n_layers, n_qubits*(n_params_per_layer+trainable_frequencies)]
651 inputs (np.ndarray): input vector of size 1
652 pulse_params Optional[np.ndarray]: pulse parameter scaler weights of shape
653 [n_layers, n_pulse_params_per_layer]
654 enc_params Optional[np.ndarray]: encoding weight vector
655 of shape [n_qubits, n_inputs]
656 gate_mode (str): Backend mode for gate execution. Can be
657 "unitary" (default) or "pulse".
659 Returns:
660 None
661 """
662 if enc_params is None:
663 # TODO: Raise warning if trainable frequencies is True, or similar. I.e., no
664 # warning if user does not care for frequencies or enc_params
665 if self.trainable_frequencies:
666 warnings.warn(
667 "Explicit call to `_circuit` or `_variational` detected: "
668 "`enc_params` is None, using `self.enc_params` instead.",
669 RuntimeWarning,
670 )
671 enc_params = self.enc_params
673 if pulse_params is None:
674 if gate_mode == "pulse":
675 warnings.warn(
676 "Explicit call to `_circuit` or `_variational` detected: "
677 "`pulse_params` is None, using `self.pulse_params` instead.",
678 RuntimeWarning,
679 )
680 pulse_params = self.pulse_params
682 if self.noise_params is not None:
683 self._apply_state_prep_noise()
685 # state preparation
686 for q in range(self.n_qubits):
687 for _sp, sp_pulse_params in zip(self._sp, self.sp_pulse_params):
688 _sp(
689 wires=q,
690 pulse_params=sp_pulse_params,
691 noise_params=self.noise_params,
692 gate_mode=gate_mode,
693 )
695 # circuit building
696 for layer in range(0, self.n_layers):
697 # ansatz layers
698 self.pqc(
699 params[layer],
700 self.n_qubits,
701 pulse_params=pulse_params[layer],
702 noise_params=self.noise_params,
703 gate_mode=gate_mode,
704 )
706 # encoding layers
707 self._iec(
708 inputs,
709 data_reupload=self.data_reupload[layer],
710 enc=self._enc,
711 enc_params=enc_params,
712 noise_params=self.noise_params,
713 )
715 # visual barrier
716 if self.degree > 1:
717 qml.Barrier(wires=list(range(self.n_qubits)), only_visual=True)
719 # final ansatz layer
720 if self.degree > 1: # same check as in init
721 self.pqc(
722 params[-1],
723 self.n_qubits,
724 pulse_params=pulse_params[-1],
725 noise_params=self.noise_params,
726 gate_mode=gate_mode,
727 )
729 # channel noise
730 if self.noise_params is not None:
731 self._apply_general_noise()
733 def _observable(self):
734 # run mixed simualtion and get density matrix
735 if self.execution_type == "density":
736 return qml.density_matrix(wires=list(range(self.n_qubits)))
737 elif self.execution_type == "state":
738 return qml.state()
739 # run default simulation and get expectation value
740 elif self.execution_type == "expval":
741 # n-local measurement
742 if self.output_qubit == -1:
743 return [qml.expval(qml.PauliZ(q)) for q in range(self.n_qubits)]
744 # local measurement(s)
745 elif isinstance(self.output_qubit, int):
746 return qml.expval(qml.PauliZ(self.output_qubit))
747 # parity measurenment
748 elif isinstance(self.output_qubit, list):
749 obs = qml.PauliZ(self.output_qubit[0])
750 for out_qubit in self.output_qubit[1:]:
751 obs = obs @ qml.PauliZ(out_qubit)
752 return qml.expval(obs)
753 else:
754 raise ValueError(
755 f"Invalid parameter `output_qubit`: {self.output_qubit}.\
756 Must be int, list or -1."
757 )
758 # run default simulation and get probs
759 elif self.execution_type == "probs":
760 if self.output_qubit == -1:
761 return qml.probs(wires=list(range(self.n_qubits)))
762 else:
763 return qml.probs(wires=self.output_qubit)
764 else:
765 raise ValueError(f"Invalid execution_type: {self.execution_type}.")
767 def _apply_state_prep_noise(self) -> None:
768 """
769 Applies a state preparation error on each qubit according to the
770 probability for StatePreparation provided in the noise_params.
771 """
772 p = self.noise_params.get("StatePreparation", 0.0)
773 for q in range(self.n_qubits):
774 if p > 0:
775 qml.BitFlip(p, wires=q)
777 def _apply_general_noise(self) -> None:
778 """
779 Applies general types of noise the full circuit (in contrast to gate
780 errors, applied directly at gate level, see Gates.Noise).
782 Possible types of noise are:
783 - AmplitudeDamping (specified through probability)
784 - PhaseDamping (specified through probability)
785 - ThermalRelaxation (specified through a dict, containing keys
786 "t1", "t2", "t_factor")
787 - Measurement (specified through probability)
788 """
789 amp_damp = self.noise_params.get("AmplitudeDamping", 0.0)
790 phase_damp = self.noise_params.get("PhaseDamping", 0.0)
791 thermal_relax = self.noise_params.get("ThermalRelaxation", 0.0)
792 meas = self.noise_params.get("Measurement", 0.0)
793 for q in range(self.n_qubits):
794 if amp_damp > 0:
795 qml.AmplitudeDamping(amp_damp, wires=q)
796 if phase_damp > 0:
797 qml.PhaseDamping(phase_damp, wires=q)
798 if meas > 0:
799 qml.BitFlip(meas, wires=q)
800 if isinstance(thermal_relax, dict):
801 t1 = thermal_relax["t1"]
802 t2 = thermal_relax["t2"]
803 t_factor = thermal_relax["t_factor"]
804 circuit_depth = self._get_circuit_depth()
805 tg = circuit_depth * t_factor
806 qml.ThermalRelaxationError(1.0, t1, t2, tg, q)
808 def _get_circuit_depth(self, inputs: Optional[np.ndarray] = None) -> int:
809 """
810 Obtain circuit depth for the model
812 Args:
813 inputs (Optional[np.ndarray]): The inputs, with which to call the
814 circuit. Defaults to None.
816 Returns:
817 int: Circuit depth (longest path of gates in circuit.)
818 """
819 inputs = self._inputs_validation(inputs)
820 spec_model = deepcopy(self)
821 spec_model.noise_params = None # remove noise
822 specs = qml.specs(spec_model.circuit)(self.params, inputs)
824 return specs["resources"].depth
826 def draw(self, inputs=None, figure="text", *args, **kwargs):
827 """
828 Draws the quantum circuit using the specified visualization method.
830 Args:
831 inputs (Optional[np.ndarray]): Input vector for the circuit. If None,
832 the default inputs are used.
833 figure (str, optional): The type of figure to generate. Must be one of
834 'text', 'mpl', or 'tikz'. Defaults to 'text'.
835 Returns:
836 Either a string, matplotlib figure or TikzFigure object (similar to string)
837 depending on the chosen visualization.
838 *args:
839 Additional arguments to be passed to the visualization method.
840 **kwargs:
841 Additional keyword arguments to be passed to the visualization method.
842 Can include `pulse_params`, `gate_mode`, `enc_params`, or `noise_params`.
844 Raises:
845 AssertionError: If the 'figure' argument is not one of the accepted values.
846 """
848 if not isinstance(self.circuit, qml.QNode):
849 # TODO: throws strange argument error if not catched
850 return ""
852 assert figure in [
853 "text",
854 "mpl",
855 "tikz",
856 ], f"Invalid figure: {figure}. Must be 'text', 'mpl' or 'tikz'."
858 inputs = self._inputs_validation(inputs)
860 if figure == "mpl":
861 result = qml.draw_mpl(self.circuit)(
862 params=self.params,
863 inputs=inputs,
864 *args,
865 **kwargs,
866 )
867 elif figure == "tikz":
868 result = QuanTikz.build(
869 self.circuit,
870 params=self.params,
871 inputs=inputs,
872 *args,
873 **kwargs,
874 )
875 else:
876 result = qml.draw(self.circuit)(params=self.params, inputs=inputs)
877 return result
879 def __repr__(self) -> str:
880 return self.draw(figure="text")
882 def __str__(self) -> str:
883 return self.draw(figure="text")
885 def _params_validation(self, params) -> np.ndarray:
886 """
887 Sets the parameters when calling the quantum circuit.
889 Args:
890 params (np.ndarray): The parameters used for the call.
892 Returns:
893 np.ndarray: Validated parameters.
894 """
895 if params is None:
896 params = self.params
897 else:
898 if numpy_boxes.ArrayBox == type(params):
899 self.params = params._value
900 else:
901 self.params = params
903 # Get rid of extra dimension
904 if len(params.shape) == 3 and params.shape[2] == 1:
905 params = params[:, :, 0]
907 return params
909 def _pulse_params_validation(self, pulse_params) -> np.ndarray:
910 """
911 Sets the pulse parameters when calling the quantum circuit.
913 Args:
914 pulse_params (np.ndarray): The pulse parameter scalers used for the call.
916 Returns:
917 np.ndarray: Validated pulse parameters, with `requires_grad` set according
918 to the current `gate_mode`.
919 """
920 if pulse_params is None:
921 pulse_params = self.pulse_params
922 else:
923 if isinstance(pulse_params, numpy_boxes.ArrayBox):
924 self.pulse_params = pulse_params._value
925 else:
926 self.pulse_params = pulse_params
928 # flip requires_grad depending on current gate_mode
929 if self.gate_mode == "pulse":
930 self.pulse_params = np.array(self.pulse_params, requires_grad=True)
931 else:
932 self.pulse_params = np.array(self.pulse_params, requires_grad=False)
934 return pulse_params
936 def _enc_params_validation(self, enc_params) -> np.ndarray:
937 """
938 Sets the encoding parameters when calling the quantum circuit
940 Args:
941 enc_params (np.ndarray): The encoding parameters used for the call
942 """
943 if enc_params is None:
944 enc_params = self.enc_params
945 else:
946 if isinstance(enc_params, numpy_boxes.ArrayBox):
947 if self.trainable_frequencies:
948 self.enc_params = enc_params._value
949 else:
950 self.enc_params = np.array(
951 enc_params._value, requires_grad=self.trainable_frequencies
952 )
953 else:
954 if self.trainable_frequencies:
955 self.enc_params = enc_params
956 else:
957 self.enc_params = np.array(
958 enc_params, requires_grad=self.trainable_frequencies
959 )
961 if len(enc_params.shape) == 1 and self.n_input_feat == 1:
962 enc_params = enc_params.reshape(-1, 1)
963 elif len(enc_params.shape) == 1 and self.n_input_feat > 1:
964 raise ValueError(
965 f"Input dimension {self.n_input_feat} >1 but \
966 `enc_params` has shape {enc_params.shape}"
967 )
969 return enc_params
971 def _inputs_validation(
972 self, inputs: Union[None, List, float, int, np.ndarray]
973 ) -> np.ndarray:
974 """
975 Validate the inputs to be a 2D numpy array of shape (batch_size, n_inputs).
977 Args:
978 inputs (Union[None, List, float, int, np.ndarray]): The input to validate.
980 Returns:
981 np.ndarray: The validated input.
982 """
983 if inputs is None:
984 # initialize to zero
985 inputs = np.array([[0] * self.n_input_feat])
986 elif isinstance(inputs, List):
987 inputs = np.stack(inputs)
988 elif isinstance(inputs, float) or isinstance(inputs, int):
989 inputs = np.array([inputs])
991 if len(inputs.shape) <= 1:
992 if self.n_input_feat == 1:
993 # add a batch dimension
994 inputs = inputs.reshape(-1, 1)
995 else:
996 if inputs.shape[0] == self.n_input_feat:
997 inputs = inputs.reshape(1, -1)
998 else:
999 inputs = inputs.reshape(-1, 1)
1000 inputs = inputs.repeat(self.n_input_feat, axis=1)
1001 warnings.warn(
1002 f"Expected {self.n_input_feat} inputs, but {inputs.shape[0]} "
1003 "was provided, replicating input for all input features.",
1004 UserWarning,
1005 )
1006 else:
1007 if inputs.shape[1] != self.n_input_feat:
1008 raise ValueError(
1009 f"Wrong number of inputs provided. Expected {self.n_input_feat} "
1010 f"inputs, but input has shape {inputs.shape}."
1011 )
1013 return inputs
1015 @staticmethod
1016 def _parallel_f(
1017 procnum,
1018 result,
1019 f,
1020 batch_size,
1021 params: np.ndarray,
1022 pulse_params: np.ndarray,
1023 inputs: np.ndarray,
1024 batch_shape,
1025 enc_params,
1026 gate_mode: str,
1027 ):
1028 """
1029 Helper function for parallelizing a function f over parameters.
1030 Sices the batch dimension based on the procnum and batch size.
1032 Args:
1033 procnum: The process number.
1034 result: The result array.
1035 f: The function to be parallelized.
1036 batch_size: The batch size.
1037 params: The parameters array.
1038 pulse_params (np.ndarray): Pulse parameter scalers for pulse-mode gates.
1039 inputs: The inputs array.
1040 enc_params: The encoding parameters array.
1041 gate_mode (str): Mode for gate execution ("unitary" or "pulse").
1042 """
1043 min_idx = max(procnum * batch_size, 0)
1045 if batch_shape[0] > 1:
1046 max_idx = min((procnum + 1) * batch_size, inputs.shape[0])
1047 inputs = inputs[min_idx:max_idx]
1048 if batch_shape[1] > 1:
1049 max_idx = min((procnum + 1) * batch_size, params.shape[2])
1050 params = params[:, :, min_idx:max_idx]
1052 result[procnum] = f(
1053 params=params,
1054 pulse_params=pulse_params,
1055 inputs=inputs,
1056 enc_params=enc_params,
1057 gate_mode=gate_mode,
1058 )
1060 def _mp_executor(self, f, params, pulse_params, inputs, enc_params, gate_mode):
1061 """
1062 Execute a function f in parallel over parameters.
1064 Args:
1065 f: A function that takes two arguments, params and inputs,
1066 and returns a numpy array.
1067 params: A 3D numpy array of parameters where the first dimension is
1068 the layer index, the second dimension is the parameter index in
1069 the layer, and the third dimension is the sample index.
1070 pulse_params (np.ndarray): array of pulse parameter scalers for pulse-mode
1071 gates.
1072 inputs: A 2D numpy array of inputs where the first dimension is
1073 the sample index and the second dimension is the input feature index.
1074 enc_params: A 1D numpy array of encoding parameters where the dimension is
1075 the qubit index.
1076 gate_mode (str): Mode for gate execution ("unitary" or "pulse").
1078 Returns:
1079 A numpy array of the output of f applied to each batch of
1080 samples in params, enc_params, and inputs.
1081 """
1082 n_processes = 1
1083 # batches available?
1084 combined_batch_size = math.prod(self.batch_shape)
1085 if (
1086 combined_batch_size > 1
1087 and self.mp_threshold > 0
1088 and combined_batch_size > self.mp_threshold
1089 ):
1090 n_processes = math.ceil(combined_batch_size / self.mp_threshold)
1091 # check if single process
1092 if n_processes == 1:
1093 if self.mp_threshold > 0:
1094 warnings.warn(
1095 f"Multiprocessing threshold {self.mp_threshold}>0, but using \
1096 single process, because {combined_batch_size} samples per batch.",
1097 )
1098 result = f(
1099 params=params,
1100 pulse_params=pulse_params,
1101 inputs=inputs,
1102 enc_params=enc_params,
1103 gate_mode=gate_mode,
1104 )
1105 else:
1106 log.info(f"Using {n_processes} processes")
1107 mpp = MultiprocessingPool(
1108 target=Model._parallel_f,
1109 n_processes=n_processes,
1110 cpu_scaler=self.cpu_scaler,
1111 batch_size=self.mp_threshold,
1112 f=f,
1113 params=params,
1114 pulse_params=pulse_params,
1115 enc_params=enc_params,
1116 inputs=inputs,
1117 gate_mode=gate_mode,
1118 batch_shape=self.batch_shape,
1119 )
1120 return_dict = mpp.spawn()
1122 # TODO: the following code could use some optimization
1123 result = [None] * len(return_dict)
1124 for k, v in return_dict.items():
1125 result[k] = v
1127 result = np.concat(result, axis=1 if self.execution_type == "expval" else 0)
1128 return result
1130 def _assimilate_batch(self, inputs, params, pulse_params):
1131 batch_shape = (
1132 inputs.shape[0],
1133 params.shape[2] if len(params.shape) == 3 else 1,
1134 )
1136 if (
1137 batch_shape[1] != 1
1138 and batch_shape[0] != batch_shape[1]
1139 and batch_shape[0] > 1
1140 ):
1141 # the following code does some dirty reshaping
1142 # TODO: optimize but be aware of the rabbit hole
1143 # key is to get the right "order" in which we repeat
1145 # [BI,D] -> [BPxBI,D]
1146 inputs = np.repeat(inputs, batch_shape[1], axis=0)
1148 # this is a tricky one, essentially we want to get
1149 # [L,Q,BP] -> [L,Q,BI,BP] -> [L,Q,BPxBI]
1150 params = np.repeat(
1151 params[:, :, np.newaxis, :], batch_shape[0], axis=2
1152 ).reshape([*params.shape[:-1], np.prod(batch_shape)])
1154 pulse_params = np.repeat(
1155 pulse_params[:, :, np.newaxis, :], batch_shape[0], axis=2
1156 ).reshape([*pulse_params.shape[:-1], np.prod(batch_shape)])
1158 return inputs, params, pulse_params, batch_shape
1160 def _requires_density(self):
1161 """
1162 Checks if the current model requires density matrix simulation or not
1163 based on the noise_params variable and the execution type
1165 Returns:
1166 bool: True if model requires density simulation
1167 """
1168 if self.execution_type == "density":
1169 return True
1171 if self.noise_params is not None:
1172 coherent_noise = ["GateError"]
1173 for k, v in self.noise_params.items():
1174 if k in coherent_noise:
1175 continue
1176 if v is not None and v > 0:
1177 return True
1178 return False
1180 def __call__(
1181 self,
1182 params: Optional[np.ndarray] = None,
1183 inputs: Optional[np.ndarray] = None,
1184 pulse_params: Optional[np.ndarray] = None,
1185 enc_params: Optional[np.ndarray] = None,
1186 noise_params: Optional[Dict[str, Union[float, Dict[str, float]]]] = None,
1187 cache: Optional[bool] = False,
1188 execution_type: Optional[str] = None,
1189 force_mean: bool = False,
1190 gate_mode: str = "unitary",
1191 ) -> np.ndarray:
1192 """
1193 Perform a forward pass of the quantum circuit with optional noise or
1194 pulse level simulation.
1196 Args:
1197 params (Optional[np.ndarray]): Weight vector of shape
1198 [n_layers, n_qubits*n_params_per_layer].
1199 If None, model internal parameters are used.
1200 inputs (Optional[np.ndarray]): Input vector of shape [1].
1201 If None, zeros are used.
1202 pulse_params (Optional[np.ndarray]): Pulse parameter scalers for pulse-mode
1203 gates.
1204 enc_params (Optional[np.ndarray]): Weight vector of shape
1205 [n_qubits, n_input_features]. If None, model internal encoding
1206 parameters are used.
1207 noise_params (Optional[Dict[str, float]], optional): The noise parameters.
1208 Defaults to None which results in the last
1209 set noise parameters being used.
1210 cache (Optional[bool], optional): Whether to cache the results.
1211 Defaults to False.
1212 execution_type (str, optional): The type of execution.
1213 Must be one of 'expval', 'density', or 'probs'.
1214 Defaults to None which results in the last set execution type
1215 being used.
1216 force_mean (bool, optional): Whether to average
1217 when performing n-local measurements.
1218 Defaults to False.
1219 gate_mode (str, optional): Gate backend mode ("unitary" or "pulse").
1220 Defaults to "unitary".
1222 Returns:
1223 np.ndarray: The output of the quantum circuit.
1224 The shape depends on the execution_type.
1225 - If execution_type is 'expval', returns an ndarray of shape
1226 (1,) if output_qubit is -1, else (len(output_qubit),).
1227 - If execution_type is 'density', returns an ndarray
1228 of shape (2**n_qubits, 2**n_qubits).
1229 - If execution_type is 'probs', returns an ndarray
1230 of shape (2**n_qubits,) if output_qubit is -1, else
1231 (2**len(output_qubit),).
1232 """
1233 # Call forward method which handles the actual caching etc.
1234 return self._forward(
1235 params=params,
1236 inputs=inputs,
1237 pulse_params=pulse_params,
1238 enc_params=enc_params,
1239 noise_params=noise_params,
1240 cache=cache,
1241 execution_type=execution_type,
1242 force_mean=force_mean,
1243 gate_mode=gate_mode,
1244 )
1246 def _forward(
1247 self,
1248 params: Optional[np.ndarray] = None,
1249 inputs: Optional[np.ndarray] = None,
1250 pulse_params: Optional[np.ndarray] = None,
1251 enc_params: Optional[np.ndarray] = None,
1252 noise_params: Optional[Dict[str, Union[float, Dict[str, float]]]] = None,
1253 cache: Optional[bool] = False,
1254 execution_type: Optional[str] = None,
1255 force_mean: bool = False,
1256 gate_mode: str = "unitary",
1257 ) -> np.ndarray:
1258 """
1259 Perform a forward pass of the quantum circuit.
1261 Args:
1262 params (Optional[np.ndarray]): Weight vector of shape
1263 [n_layers, n_qubits*n_params_per_layer].
1264 If None, model internal parameters are used.
1265 inputs (Optional[np.ndarray]): Input vector of shape [1].
1266 If None, zeros are used.
1267 pulse_params (Optional[np.ndarray]): Pulse parameter scalers for pulse-mode
1268 gates.
1269 enc_params (Optional[np.ndarray]): Weight vector of shape
1270 [n_qubits, n_input_features]. If None, model internal encoding
1271 parameters are used.
1272 noise_params (Optional[Dict[str, float]], optional): The noise parameters.
1273 Defaults to None which results in the last
1274 set noise parameters being used.
1275 cache (Optional[bool], optional): Whether to cache the results.
1276 Defaults to False.
1277 execution_type (str, optional): The type of execution.
1278 Must be one of 'expval', 'density', or 'probs'.
1279 Defaults to None which results in the last set execution type
1280 being used.
1281 force_mean (bool, optional): Whether to average
1282 when performing n-local measurements.
1283 Defaults to False.
1284 gate_mode (str, optional): Gate backend mode ("unitary" or "pulse").
1285 Defaults to "unitary".
1288 Returns:
1289 np.ndarray: The output of the quantum circuit.
1290 The shape depends on the execution_type.
1291 - If execution_type is 'expval', returns an ndarray of shape
1292 (1,) if output_qubit is -1, else (len(output_qubit),).
1293 - If execution_type is 'density', returns an ndarray
1294 of shape (2**n_qubits, 2**n_qubits).
1295 - If execution_type is 'probs', returns an ndarray
1296 of shape (2**n_qubits,) if output_qubit is -1, else
1297 (2**len(output_qubit),).
1299 Raises:
1300 NotImplementedError: If the number of shots is not None or if the
1301 expectation value is True.
1302 ValueError:
1303 - If `pulse_params` are provided but `gate_mode` is not "pulse".
1304 - If `noise_params` are provided while `gate_mode` is "pulse" (noise
1305 not supported in pulse mode).
1306 """
1307 # set the parameters as object attributes
1308 if noise_params is not None:
1309 self.noise_params = noise_params
1310 if execution_type is not None:
1311 self.execution_type = execution_type
1312 self.gate_mode = gate_mode
1314 # consistency checks
1315 if pulse_params is not None and gate_mode != "pulse":
1316 raise ValueError(
1317 "pulse_params were provided but gate_mode is not 'pulse'. "
1318 "Either switch gate_mode='pulse' or do not pass pulse_params."
1319 )
1321 if noise_params is not None and gate_mode == "pulse":
1322 raise ValueError(
1323 "Noise is not supported in 'pulse' gate_mode. "
1324 "Either remove noise_params or use gate_mode='unitary'."
1325 )
1327 params = self._params_validation(params)
1328 pulse_params = self._pulse_params_validation(pulse_params)
1329 inputs = self._inputs_validation(inputs)
1330 enc_params = self._enc_params_validation(enc_params)
1332 inputs, params, pulse_params, self.batch_shape = self._assimilate_batch(
1333 inputs,
1334 params,
1335 pulse_params,
1336 )
1337 # the qasm representation contains the bound parameters,
1338 # thus it is ok to hash that
1339 hs = hashlib.md5(
1340 repr(
1341 {
1342 "n_qubits": self.n_qubits,
1343 "n_layers": self.n_layers,
1344 "pqc": self.pqc.__class__.__name__,
1345 "dru": self.data_reupload,
1346 "params": self.params, # use safe-params
1347 "pulse_params": self.pulse_params,
1348 "enc_params": self.enc_params,
1349 "noise_params": self.noise_params,
1350 "execution_type": self.execution_type,
1351 "inputs": inputs,
1352 "output_qubit": self.output_qubit,
1353 }
1354 ).encode("utf-8")
1355 ).hexdigest()
1357 result: Optional[np.ndarray] = None
1358 if cache:
1359 name: str = f"pqc_{hs}.npy"
1361 cache_folder: str = ".cache"
1362 if not os.path.exists(cache_folder):
1363 os.mkdir(cache_folder)
1365 file_path: str = os.path.join(cache_folder, name)
1367 if os.path.isfile(file_path):
1368 result = np.load(file_path)
1370 if result is None:
1371 # if density matrix requested or noise params used
1372 if self._requires_density():
1373 result = self._mp_executor(
1374 f=self.circuit_mixed,
1375 params=params, # use arraybox params
1376 pulse_params=pulse_params,
1377 inputs=inputs,
1378 enc_params=enc_params,
1379 gate_mode=gate_mode,
1380 )
1381 else:
1382 if not isinstance(self.circuit, qml.QNode):
1383 result = self.circuit(
1384 inputs=inputs,
1385 )
1386 else:
1387 result = self._mp_executor(
1388 f=self.circuit,
1389 params=params, # use arraybox params
1390 pulse_params=pulse_params,
1391 inputs=inputs,
1392 enc_params=enc_params,
1393 gate_mode=gate_mode,
1394 )
1396 if isinstance(result, list):
1397 result = np.stack(result)
1399 if self.execution_type == "expval" and force_mean and self.output_qubit == -1:
1400 # exception for torch layer because it swaps batch and output dimension
1401 if not isinstance(self.circuit, qml.QNode):
1402 result = result.mean(axis=-1)
1403 else:
1404 result = result.mean(axis=0)
1405 elif self.execution_type == "probs" and force_mean and self.output_qubit == -1:
1406 # exception for torch layer because it swaps batch and output dimension
1407 if not isinstance(self.circuit, qml.QNode):
1408 result = result[..., -1].sum(axis=-1)
1409 else:
1410 result = result[1:, ...].sum(axis=0)
1412 if self.batch_shape[0] > 1 and self.batch_shape[1] > 1:
1413 result = result.reshape(-1, *self.batch_shape)
1415 result = result.squeeze()
1417 if cache:
1418 np.save(file_path, result)
1420 return result