From 76a0ca189db2493b0176785ce482d41932f55a21 Mon Sep 17 00:00:00 2001 From: smit Date: Tue, 9 Apr 2024 13:29:12 +0200 Subject: [PATCH 1/3] Added v0 of qnn_builder --- .../constructors/ufa_models/README.md | 67 ++ .../constructors/ufa_models/__init__.py | 4 + .../constructors/ufa_models/config.py | 263 +++++++ .../constructors/ufa_models/quantum_models.py | 706 ++++++++++++++++++ 4 files changed, 1040 insertions(+) create mode 100644 qadence_libs/constructors/ufa_models/README.md create mode 100644 qadence_libs/constructors/ufa_models/__init__.py create mode 100644 qadence_libs/constructors/ufa_models/config.py create mode 100644 qadence_libs/constructors/ufa_models/quantum_models.py diff --git a/qadence_libs/constructors/ufa_models/README.md b/qadence_libs/constructors/ufa_models/README.md new file mode 100644 index 0000000..358582d --- /dev/null +++ b/qadence_libs/constructors/ufa_models/README.md @@ -0,0 +1,67 @@ +## A submodule to quickly make different kinds of QNNs + +### Features: + +Independently specify the feature map, ansatz, and observable configurations. The `build_qnn` function will automatically build the QNN based on the configurations provided. + +### Feature Map: + +- `num_features`: Number of features to be encoded. +- `basis_set`: Basis set to be used for encoding the features. Fourier or Chebyshev. +- `reupload_scaling`: Scaling strategy for reuploading the features. Constant Tower or Exponential. +- `feature_range`: Range of data that the input data is assumed to come from. +- `target_range`: Range of data that the encoder assumes as natural range. +- `multivariate_strategy`: Strategy to be used for encoding multiple features. Series or Parallel. +- `feature_map_strategy`: Strategy to be used for encoding the features. Digital, Analog, or Rydberg. +- `param_prefix`: Prefix to be used for the parameters of the feature map. +- `num_repeats`: Number of times each feature is reuploaded. +- `operation`: Operation to be used for encoding the features. +- `inputs`: Inputs to be used for encoding the features. + +### Ansatz: + +- `num_layers`: Number of layers in the ansatz. +- `ansatz_type`: Type of ansatz to be used. HEA or IIA. +- `ansatz_strategy`: Strategy to be used for encoding the features. Digital, SDAQC or Rydberg. +- `strategy_args`: Arguments to be passed to the strategy. +- `param_prefix`: Prefix to be used for the parameters of the ansatz. + +### Observable: + +- `detuning`: The detuning term in the observable. +- `detuning_strength`: Strength of the detuning term in the observable. + +### Usage: + +```python + +from qadence.types import BasisSet, ReuploadScaling + +from config import AnsatzConfig, FeatureMapConfig, ObservableConfig +from quantum_models import build_qnn + +fm_config = FeatureMapConfig( + num_features=1, + basis_set=BasisSet.CHEBYSHEV, + reupload_scaling=ReuploadScaling.TOWER, + feature_range=(-1.2, 1.2), + feature_map_strategy="digital", + multivariate_strategy="series", +) + +ansatz_config = AnsatzConfig( + num_layers=2, + ansatz_type="hea", + ansatz_strategy="rydberg", +) + +obs_config = ObservableConfig(detuning_strength="z") + +f = build_qnn_model( + register=3, + fm_config=fm_config, + ansatz_config=ansatz_config, + observable_config=obs_config, +) + +``` diff --git a/qadence_libs/constructors/ufa_models/__init__.py b/qadence_libs/constructors/ufa_models/__init__.py new file mode 100644 index 0000000..d6ce833 --- /dev/null +++ b/qadence_libs/constructors/ufa_models/__init__.py @@ -0,0 +1,4 @@ +from __future__ import annotations + +from .config import AnsatzConfig, FeatureMapConfig, ObservableConfig # noqa F401 +from .quantum_models import build_qnn_model # noqa F401 diff --git a/qadence_libs/constructors/ufa_models/config.py b/qadence_libs/constructors/ufa_models/config.py new file mode 100644 index 0000000..fd27e01 --- /dev/null +++ b/qadence_libs/constructors/ufa_models/config.py @@ -0,0 +1,263 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from sympy import Basic +from typing import Callable, Type + +from qadence.blocks.primitive import ParametricBlock +from qadence.logger import get_logger +from qadence.blocks.analog import AnalogBlock +from qadence.operations import AnalogRX, RX, Z +from qadence.parameters import Parameter +from qadence.types import BasisSet, ReuploadScaling, TArray + +logger = get_logger(__file__) + + +@dataclass +class FeatureMapConfig: + num_features: int + """Number of feature parameters to be encoded.""" + + basis_set: BasisSet | list[BasisSet] + """ + Basis set for feature encoding. Takes qadence.BasisSet. + Give a single BasisSet to use the same for all features. + Give a list of BasisSet to apply each item for a corresponding variable. + BasisSet.FOURIER for Fourier encoding. + BasisSet.CHEBYSHEV for Chebyshev encoding. + """ + + reupload_scaling: ReuploadScaling | list[ReuploadScaling] + """ + Scaling for encoding the same feature on different qubits in the + same layer of the feature maps. Takes qadence.ReuploadScaling. + Give a single ReuploadScaling to use the same for all features. + Give a list of ReuploadScaling to apply each item for a corresponding variable. + ReuploadScaling.CONSTANT for constant scaling. + ReuploadScaling.TOWER for linearly increasing scaling. + ReuploadScaling.EXP for exponentially increasing scaling. + """ + + feature_range: tuple[float, float] | list[tuple[float, float]] + """ + Range of data that the input data is assumed to come from. + Give a single tuple to use the same range for all features. + Give a list of tuples to use each item for a corresponding variable. + """ + + target_range: tuple[float, float] | list[tuple[float, float]] | None = None + """ + Range of data the data encoder assumes as natural range. + Give a single tuple to use the same range for all features. + Give a list of tuples to use each item for a corresponding variable. + """ + + multivariate_strategy: str = "parallel" + """ + The encoding strategy in case of multi-variate function. If "parallel", + the features are encoded in one block of rotation gates. with each + feature given an equal number of qubits. If "serial", the features are + encoded sequentially, with a HEA block between. "parallel" is allowed + only for "digital" `feature_map_strategy`. + """ + + feature_map_strategy: str = "digital" + """Strategy for feature map. Accepts 'digital', 'analog' or 'rydberg'. Defaults to "digital".""" + + param_prefix: str = "phi" + """The base name of the feature map parameter. Defaults to `phi`.""" + + num_repeats: int | list[int] = 0 + """ + Number of feature map layers repeated in the data reuploadig step. If all are to be \ + repeated the same number of times, then can give a single `int`. For different number \ + of repeatitions for each feature, provide a list of `int`s corresponding to desired \ + number of repeatitions. This amounts to the number of additional reuploads. So if \ + `num_repeats` is N, the data gets uploaded N+1 times. Defaults to no repeatition. + """ + + operation: Callable[[Parameter | Basic], AnalogBlock] | Type[RX] | None = None + """ + type of operation. Choose among the analog or digital rotations or a custom + callable function returning an AnalogBlock instance + """ + + inputs: list[Basic | str] | None = None + """ + List that indicates the order of variables of the tensors that are passed + to the model. Given input tensors `xs = torch.rand(batch_size, input_size:=2)` a QNN + with `inputs=("t", "x")` will assign `t, x = xs[:,0], xs[:,1]`. + """ + + def __post_init__(self) -> None: + if self.multivariate_strategy == "parallel" and self.num_features > 1: + assert ( + self.feature_map_strategy == "digital" + ), "For `parallel` encoding of multiple features, the `feature_map_strategy` must be \ + of `digital` type." + + if self.operation is None: + if self.feature_map_strategy == "digital": + self.operation = RX + elif self.feature_map_strategy == "analog": + self.operation = AnalogRX # type: ignore[assignment] + + else: + if self.feature_map_strategy == "digital": + if isinstance(self.operation, AnalogBlock): + logger.warning( + "The `operation` is of type `AnalogBlock` but the `feature_map_strategy` is\ + `digital`. The `feature_map_strategy` will be modified and given operation\ + will be used." + ) + + self.feature_map_strategy = "analog" + + elif self.feature_map_strategy == "analog": + if isinstance(self.operation, ParametricBlock): + logger.warning( + "The `operation` is a digital gate but the `feature_map_strategy` is\ + `analog`. The `feature_map_strategy` will be modified and given operation\ + will be used." + ) + + self.feature_map_strategy = "digital" + + if isinstance(self.basis_set, BasisSet): + self.basis_set = [self.basis_set for i in range(self.num_features)] + else: + assert ( + len(self.basis_set) == self.num_features + ), f"Length of set of bases {len(self.feature_range)} must match the number \ + of features {self.num_features}. Or provide a single `BasisSet` to \ + use same basis for all features." + + if isinstance(self.reupload_scaling, ReuploadScaling): + self.reupload_scaling = [ + self.reupload_scaling for i in range(self.num_features) + ] + else: + assert ( + len(self.reupload_scaling) == self.num_features + ), f"Length of the reupload scalings {len(self.feature_range)} must match the number \ + of features {self.num_features}. Or provide a single `ReuploadScaling` for \ + same scaling for all features." + + if isinstance(self.feature_range, tuple): + self.feature_range = [self.feature_range for i in range(self.num_features)] + else: + assert ( + len(self.feature_range) == self.num_features + ), f"Length of the feature ranges {len(self.feature_range)} must match the number \ + of features {self.num_features}. Or provide a single tuple(float, float) for \ + same expected feature range for all features." + + if self.target_range is None: + self.target_range = [None for i in range(self.num_features)] # type: ignore[assignment] + elif isinstance(self.target_range, tuple): + self.target_range = [self.target_range for i in range(self.num_features)] + else: + assert ( + len(self.target_range) == self.num_features + ), f"Length of the feature ranges {len(self.feature_range)} must match the number \ + of features {self.num_features}. Or provide a single tuple(float, float) for \ + same expected feature range for all features." + + if isinstance(self.num_repeats, int): + self.num_repeats = [self.num_repeats for i in range(self.num_features)] + else: + assert ( + len(self.num_repeats) == self.num_features + ), f"Length of the repeat array {len(self.num_repeats)} must match the number \ + of features {self.num_features}. Or provide a single integer for same number \ + of repeatitions for all features." + + if self.inputs is None: + self.inputs = [f"phi_{i}" for i in range(self.num_features)] + + +@dataclass +class AnsatzConfig: + num_layers: int + """Number of layers of the ansatz.""" + + ansatz_type: str + """ + What type of ansatz. + "hea" for Hardware Efficient Ansatz. + "iia" for Identity intialized Ansatz. + """ + + ansatz_strategy: str + """ + Ansatz strategy. + "digital" for fully digital ansatz. Required if `ansatz_type` is `iia`. + "sdaqc" for analog entangling block. + "rydberg" for fully rydberg hea ansatz. + """ + + strategy_args: dict = field(default_factory=dict) + """ + A dictionary containing keyword arguments to the function creating the ansatz. + Details about each below. + + For "digital" strategy, accepts the following: + periodic (bool): if the qubits should be linked periodically. + periodic=False is not supported in emu-c. + operations (list): list of operations to cycle through in the + digital single-qubit rotations of each layer. + Defaults to [RX, RY, RX] for hea and [RX, RY] for iia. + entangler (AbstractBlock): 2-qubit entangling operation. + Supports CNOT, CZ, CRX, CRY, CRZ, CPHASE. Controlld rotations + will have variational parameters on the rotation angles. + Defaults to CNOT + + For "sdaqc" strategy, accepts the following: + operations (list): list of operations to cycle through in the + digital single-qubit rotations of each layer. + Defaults to [RX, RY, RX] for hea and [RX, RY] for iia. + entangler (AbstractBlock): Hamiltonian generator for the + analog entangling layer. Time parameter is considered variational. + Defaults to NN interaction. + + For "rydberg" strategy, accepts the following: + addressable_detuning: whether to turn on the trainable semi-local addressing pattern + on the detuning (n_i terms in the Hamiltonian). + Defaults to True. + addressable_drive: whether to turn on the trainable semi-local addressing pattern + on the drive (sigma_i^x terms in the Hamiltonian). + Defaults to False. + tunable_phase: whether to have a tunable phase to get both sigma^x and sigma^y rotations + in the drive term. If False, only a sigma^x term will be included in the drive part + of the Hamiltonian generator. + Defaults to False. + """ + # The default for a dataclass can not be a mutable object without using this default_factory. + + param_prefix: str = "theta" + """The base bame of the variational parameter.""" + + def __post_init__(self) -> None: + if self.ansatz_type == "iia": + assert ( + self.ansatz_strategy != "rydberg" + ), "Rydberg strategy not allowed for Identity-initialized ansatz." + + +@dataclass +class ObservableConfig: + detuning: Type[Z] = Z + """ + Single qubit detuning of the observable Hamiltonian. + Accepts single-qubit operator N, X, Y, or Z. + Defaults to Z. + """ + + detuning_strength: TArray | str | None = None + """ + list of values to be used as the detuning strength for each qubit. + Alternatively, some string "x" can be passed, which will create a parameterized + detuning for each qubit, each labelled as `"x_i"`. + Defaults to 1.0 for each qubit. + """ diff --git a/qadence_libs/constructors/ufa_models/quantum_models.py b/qadence_libs/constructors/ufa_models/quantum_models.py new file mode 100644 index 0000000..b317227 --- /dev/null +++ b/qadence_libs/constructors/ufa_models/quantum_models.py @@ -0,0 +1,706 @@ +from __future__ import annotations + +import numpy as np + +from qadence.blocks import chain, kron +from qadence.blocks.abstract import AbstractBlock +from qadence.blocks.composite import ChainBlock +from qadence.circuit import QuantumCircuit +from qadence.constructors import ( + analog_feature_map, + feature_map, + hamiltonian_factory, + identity_initialized_ansatz, + rydberg_feature_map, + rydberg_hea, + rydberg_tower_feature_map, +) +from qadence.constructors.ansatze import hea_digital, hea_sDAQC +from qadence.models import QNN +from qadence.operations import CNOT, RX, RY, H +from qadence.register import Register +from qadence.types import Interaction, ReuploadScaling, Strategy + +from eic_aero.ufa_models.config import AnsatzConfig, FeatureMapConfig, ObservableConfig + + +def _create_support_arrays( + num_qubits: int, + num_features: int, + multivariate_strategy: str, +) -> list[tuple[int, ...]]: + """ + Create the support arrays for the digital feature map. + + Args: + num_qubits (int): The number of qubits. + num_features (int): The number of features. + multivariate_strategy (str): The multivariate encoding strategy. + Either 'series' or 'parallel'. + + Returns: + list[tuple[int, ...]]: The list of support arrays. ith element of the list is the support + array for the ith feature. + + Raises: + ValueError: If the number of features is greater than the number of qubits + with parallel encoding. Not possible to encode these features in parallel. + ValueError: If the multivariate strategy is not 'series' or 'parallel'. + """ + if multivariate_strategy == "series": + return [tuple(range(num_qubits)) for i in range(num_features)] + elif multivariate_strategy == "parallel": + if num_features <= num_qubits: + return [ + tuple(x.tolist()) + for x in np.array_split(np.arange(num_qubits), num_features) + ] + else: + raise ValueError( + f"Number of features {num_features} must be less than or equal to the number of \ + qubits {num_qubits}. if the features are to be encoded is parallely." + ) + else: + raise ValueError( + f"Invalid encoding strategy {multivariate_strategy} provided. Only 'series' or \ + 'parallel' are allowed." + ) + + +def _encode_features_series_digital( + register: int | Register, + config: FeatureMapConfig, +) -> list[AbstractBlock]: + """ + Encode the features in series using digital feature map. + + Args: + register (int | Register): The number of qubits or the register object. + config (FeatureMapConfig): The feature map configuration. + + Returns: + list[AbstractBlock]: The list of digital feature map blocks. + """ + num_qubits = register if isinstance(register, int) else register.n_qubits + + support_arrays = _create_support_arrays( + num_qubits=num_qubits, + num_features=config.num_features, + multivariate_strategy=config.multivariate_strategy, + ) + + reuploads = ( + [config.num_repeats for i in range(config.num_features)] + if isinstance(config.num_repeats, int) + else config.num_repeats[:] + ) + + fm_blocks = [] + + for i in range(max(reuploads) + 1): + for j in range(config.num_features): + if i == 0 or reuploads[j] > 0: + fm_blocks.append( + feature_map( + num_qubits, + support=support_arrays[j], + param=f"{config.param_prefix}_{j}", + op=config.operation, # type: ignore[arg-type] + fm_type=config.basis_set[j], + reupload_scaling=config.reupload_scaling[j], + feature_range=config.feature_range[j], # type: ignore[arg-type] + target_range=config.target_range[j], # type: ignore[arg-type] + ) + ) + + if i != 0: + reuploads = [x - 1 if x > 0 else x for x in reuploads] + + return fm_blocks + + +def _encode_features_parallel_digital( + register: int | Register, + config: FeatureMapConfig, +) -> list[AbstractBlock]: + """ + Encode the features in parallel using digital feature map. + + Args: + register (int | Register): The number of qubits or the register object. + config (FeatureMapConfig): The feature map configuration. + + Returns: + list[AbstractBlock]: The list of digital feature map blocks. + """ + num_qubits = register if isinstance(register, int) else register.n_qubits + + support_arrays = _create_support_arrays( + num_qubits=num_qubits, + num_features=config.num_features, + multivariate_strategy=config.multivariate_strategy, + ) + + reuploads = ( + [config.num_repeats for i in range(config.num_features)] + if isinstance(config.num_repeats, int) + else config.num_repeats[:] + ) + + fm_blocks = [] + + for i in range(max(reuploads) + 1): + fm_layer = [] + for j in range(config.num_features): + if i == 0 or reuploads[j] > 0: + fm_layer.append( + feature_map( + len(support_arrays[j]), + support=support_arrays[j], + param=f"{config.param_prefix}_{j}", + op=config.operation, # type: ignore[arg-type] + fm_type=config.basis_set[j], + reupload_scaling=config.reupload_scaling[j], + feature_range=config.feature_range[j], # type: ignore[arg-type] + target_range=config.target_range[j], # type: ignore[arg-type] + ) + ) + + fm_blocks.append(kron(*fm_layer)) + + if i != 0: + reuploads = [x - 1 if x > 0 else x for x in reuploads] + + return fm_blocks + + +def _create_digital_fm( + register: int | Register, + config: FeatureMapConfig, +) -> list[AbstractBlock]: + """ + Create the digital feature map. + + Args: + register (int | Register): The number of qubits or the register object. + config (FeatureMapConfig): The feature map configuration. + + Returns: + list[AbstractBlock]: The list of digital feature map blocks. + + Raises: + ValueError: If the encoding strategy is invalid. Only 'series' or 'parallel' are allowed. + """ + if config.multivariate_strategy == "series": + fm_blocks = _encode_features_series_digital(register, config) + elif config.multivariate_strategy == "parallel": + fm_blocks = _encode_features_parallel_digital(register, config) + else: + raise ValueError( + f"Invalid encoding strategy {config.multivariate_strategy} provided. Only 'series' or \ + 'parallel' are allowed." + ) + + return fm_blocks + + +def _create_analog_fm( + register: int | Register, + config: FeatureMapConfig, +) -> list[AbstractBlock]: + """ + Create the analog feature map. + + Args: + register (int | Register): The number of qubits or the register object. + config (FeatureMapConfig): The feature map configuration. + + Returns: + list[AbstractBlock]: The list of analog feature map blocks. + """ + reuploads = ( + [config.num_repeats for i in range(config.num_features)] + if isinstance(config.num_repeats, int) + else config.num_repeats[:] + ) + + fm_blocks = [] + + for i in range(max(reuploads) + 1): + for j in range(config.num_features): + if ( + i == 0 or reuploads[j] > 0 + ): # If it is the first pass or if there are reuploads left to do + fm_blocks.append( + analog_feature_map( + param=f"{config.param_prefix}_{j}", + op=config.operation, # type: ignore[arg-type] + fm_type=config.basis_set[j], + reupload_scaling=config.reupload_scaling[j], + feature_range=config.feature_range[j], # type: ignore[arg-type] + target_range=config.target_range[j], # type: ignore[arg-type] + ) + ) + + if i != 0: + reuploads = [x - 1 if x > 0 else x for x in reuploads] + + return fm_blocks + + +def _encode_feature_rydberg( + num_qubits: int, + param: str, + reupload_scaling: ReuploadScaling, +) -> AbstractBlock: + """ + Encode features using a Rydberg feature map. + + Args: + num_qubits (int): The number of qubits to encode the features on. + param (str): The parameter prefix to use for the feature map parameter names. + reupload_scaling (ReuploadScaling): The scaling strategy for reuploads. + + Returns: + The Rydberg feature map. + + Raises: + NotImplementedError: If the reupload scaling strategy is not implemented. + Only `ReuploadScaling.CONSTANT` and `ReuploadScaling.TOWER` are supported. + """ + if reupload_scaling == ReuploadScaling.CONSTANT: + return rydberg_feature_map(n_qubits=num_qubits, param=param) + + elif reupload_scaling == ReuploadScaling.TOWER: + return rydberg_tower_feature_map(n_qubits=num_qubits, param=param) + + else: + raise NotImplementedError( + f"Rydberg feature map not implemented for {reupload_scaling}" + ) + + +def _create_rydberg_fm( + register: int | Register, + config: FeatureMapConfig, +) -> list[AbstractBlock]: + """ + Create a Rydberg feature map for the given configuration. + + Args: + register (int | Register): The number of qubits or the register to apply the feature map to. + config (FeatureMapConfig): The configuration for the feature map. + + Returns: + list: A list of Rydberg feature map blocks. + """ + num_qubits = register if isinstance(register, int) else register.n_qubits + + reuploads = ( + [config.num_repeats for i in range(config.num_features)] + if isinstance(config.num_repeats, int) + else config.num_repeats[:] + ) + + fm_blocks = [] + + for i in range(max(reuploads) + 1): + for j in range(config.num_features): + if i == 0 or reuploads[j] > 0: + fm_blocks.append( + _encode_feature_rydberg( + num_qubits=num_qubits, + param=f"{config.param_prefix}_{j}", + reupload_scaling=config.reupload_scaling[j], # type: ignore[arg-type] + ) + ) + + if i != 0: + reuploads = [x - 1 if x > 0 else x for x in reuploads] + + return fm_blocks + + +def _create_fm( + register: int | Register, + config: FeatureMapConfig, +) -> list[AbstractBlock]: + """ + Create the feature map based on the configuration. + + Args: + register (int | Register): The number of qubits or the register. + config (FeatureMapConfig): The configuration for the feature map. + + Returns: + list[AbstractBlock]: The feature map blocks. + + Raises: + ValueError: If the feature map strategy is not 'digital', 'analog' or 'rydberg'. + """ + if config.feature_map_strategy == "digital": + return _create_digital_fm(register=register, config=config) + elif config.feature_map_strategy == "analog": + return _create_analog_fm(register=register, config=config) + elif config.feature_map_strategy == "rydberg": + return _create_rydberg_fm(register=register, config=config) + else: + raise ValueError( + f"Wrong feature map type {config.feature_map_strategy} provided. Only 'digital', \ + 'analog' or 'rydberg' allowed." + ) + + +def _ansatz_layer( + register: int | Register, + ansatz_config: AnsatzConfig, + index: int, +) -> AbstractBlock: + """ + Create a layer of the ansatz based on the configuration. + + Args: + register (int | Register): The number of qubits or the register. + ansatz_config (AnsatzConfig): The configuration for the ansatz. + index (int): The index of the layer. + + Returns: + AbstractBlock: The layer of the ansatz. + """ + new_config = AnsatzConfig( + num_layers=1, + ansatz_type=ansatz_config.ansatz_type, + ansatz_strategy=ansatz_config.ansatz_strategy, + strategy_args=ansatz_config.strategy_args, + param_prefix=f"fm_{index}", + ) + + return _create_ansatz(register=register, config=new_config) + + +def _create_iia_digital( + num_qubits: int, + config: AnsatzConfig, +) -> AbstractBlock: + """ + Create the Digital Identity Initialized Ansatz based on the configuration. + + Args: + num_qubits (int): The number of qubits. + config (AnsatzConfig): The configuration for the ansatz. + + Returns: + AbstractBlock: The Identity Initialized Ansatz. + """ + operations = config.strategy_args.get("operations", [RX, RY]) + entangler = config.strategy_args.get("entangler", CNOT) + periodic = config.strategy_args.get("periodic", False) + + return identity_initialized_ansatz( + n_qubits=num_qubits, + depth=config.num_layers, + param_prefix=config.param_prefix, + strategy=Strategy.DIGITAL, + rotations=operations, + entangler=entangler, + periodic=periodic, + ) + + +def _create_iia_sdaqc( + num_qubits: int, + config: AnsatzConfig, +) -> AbstractBlock: + """ + Create the SDAQC Identity Initialized Ansatz based on the configuration. + + Args: + num_qubits (int): The number of qubits. + config (AnsatzConfig): The configuration for the ansatz. + + Returns: + AbstractBlock: The SDAQC Identity Initialized Ansatz. + """ + operations = config.strategy_args.get("operations", [RX, RY]) + entangler = config.strategy_args.get("entangler", CNOT) + periodic = config.strategy_args.get("periodic", False) + + return identity_initialized_ansatz( + n_qubits=num_qubits, + depth=config.num_layers, + param_prefix=config.param_prefix, + strategy=Strategy.SDAQC, + rotations=operations, + entangler=entangler, + periodic=periodic, + ) + + +def _create_iia( + num_qubits: int, + config: AnsatzConfig, +) -> AbstractBlock: + """ + Create the Identity Initialized Ansatz based on the configuration. + + Args: + num_qubits (int): The number of qubits. + config (AnsatzConfig): The configuration for the ansatz. + + Returns: + AbstractBlock: The Identity Initialized Ansatz. + + Raises: + ValueError: If the ansatz strategy is not supported. Only 'digital' and 'sdaqc' are allowed. + """ + if config.ansatz_strategy == "digital": + return _create_iia_digital(num_qubits=num_qubits, config=config) + elif config.ansatz_strategy == "sdaqc": + return _create_iia_sdaqc(num_qubits=num_qubits, config=config) + else: + raise ValueError( + f"Invalid ansatz strategy {config.ansatz_strategy} provided. Only 'digital', 'sdaqc', \ + allowed for IIA." + ) + + +def _create_hea_digital(num_qubits: int, config: AnsatzConfig) -> AbstractBlock: + """ + Create the Digital Hardware Efficient Ansatz based on the configuration. + + Args: + num_qubits (int): The number of qubits. + config (AnsatzConfig): The configuration for the ansatz. + + Returns: + AbstractBlock: The Digital Hardware Efficient Ansatz. + """ + operations = config.strategy_args.get("rotations", [RX, RY, RX]) + entangler = config.strategy_args.get("entangler", CNOT) + periodic = config.strategy_args.get("periodic", False) + + return hea_digital( + n_qubits=num_qubits, + depth=config.num_layers, + param_prefix=config.param_prefix, + operations=operations, + entangler=entangler, + periodic=periodic, + ) + + +def _create_hea_sdaqc(num_qubits: int, config: AnsatzConfig) -> AbstractBlock: + """ + Create the SDAQC Hardware Efficient Ansatz based on the configuration. + + Args: + num_qubits (int): The number of qubits. + config (AnsatzConfig): The configuration for the ansatz. + + Returns: + AbstractBlock: The SDAQC Hardware Efficient Ansatz. + """ + operations = config.strategy_args.get("rotations", [RX, RY, RX]) + entangler = config.strategy_args.get( + "entangler", hamiltonian_factory(num_qubits, interaction=Interaction.NN) + ) + + return hea_sDAQC( + n_qubits=num_qubits, + depth=config.num_layers, + param_prefix=config.param_prefix, + operations=operations, + entangler=entangler, + ) + + +def _create_hea_rydberg( + register: int | Register, + config: AnsatzConfig, +) -> AbstractBlock: + """ + Create the Rydberg Hardware Efficient Ansatz based on the configuration. + + Args: + register (int | Register): The number of qubits or the register object. + config (AnsatzConfig): The configuration for the ansatz. + + Returns: + AbstractBlock: The Rydberg Hardware Efficient Ansatz. + """ + register = ( + register + if isinstance(register, Register) + else Register.circle(n_qubits=register) + ) + + addressable_detuning = config.strategy_args.get("addressable_detuning", True) + addressable_drive = config.strategy_args.get("addressable_drive", False) + tunable_phase = config.strategy_args.get("tunable_phase", False) + + return rydberg_hea( + register=register, + n_layers=config.num_layers, + addressable_detuning=addressable_detuning, + addressable_drive=addressable_drive, + tunable_phase=tunable_phase, + additional_prefix=config.param_prefix, + ) + + +def _create_hea_ansatz( + register: int | Register, + config: AnsatzConfig, +) -> AbstractBlock: + """ + Create the Hardware Efficient Ansatz based on the configuration. + + Args: + register (int | Register): The number of qubits or the register to create the ansatz for. + config (AnsatzConfig): The configuration for the ansatz. + + Returns: + AbstractBlock: The hardware efficient ansatz block. + + Raises: + ValueError: If the ansatz strategy is not 'digital', 'sdaqc', or 'rydberg'. + """ + num_qubits = register if isinstance(register, int) else register.n_qubits + + if config.ansatz_strategy == "digital": + return _create_hea_digital(num_qubits=num_qubits, config=config) + elif config.ansatz_strategy == "sdaqc": + return _create_hea_sdaqc(num_qubits=num_qubits, config=config) + elif config.ansatz_strategy == "rydberg": + return _create_hea_rydberg(register=register, config=config) + else: + raise ValueError( + f"Invalid ansatz strategy {config.ansatz_strategy} provided. Only 'digital', 'sdaqc', \ + and 'rydberg' allowed" + ) + + +def _create_ansatz( + register: int | Register, + config: AnsatzConfig, +) -> AbstractBlock: + """ + Create the ansatz based on the configuration. + + Args: + register (int | Register): Number of qubits or a register object. + config (AnsatzConfig): Configuration for the ansatz. + + Returns: + AbstractBlock: The ansatz block. + + Raises: + NotImplementedError: If the ansatz type is not implemented. + """ + num_qubits = register if isinstance(register, int) else register.n_qubits + + if config.ansatz_type == "iia": + return _create_iia(num_qubits=num_qubits, config=config) + elif config.ansatz_type == "hea": + return _create_hea_ansatz(register=register, config=config) + else: + raise NotImplementedError( + f"Ansatz of type {config.ansatz_type} not implemented yet. Only 'hea' and\ + 'iia' available." + ) + + +def _interleave_ansatz_in_fm( + register: int | Register, + fm_blocks: list[AbstractBlock], + ansatz_config: AnsatzConfig, +) -> ChainBlock: + """ + Interleave the ansatz layers in between the feature map layers. + + Args: + register (int | Register): Number of qubits or a register object. + fm_blocks (list[AbstractBlock]): List of feature map blocks. + ansatz_config (AnsatzConfig): Ansatz configuration. + + Returns: + ChainBlock: A block containing feature map layers with interleaved ansatz layers. + """ + full_fm = [] + for idx, block in enumerate(fm_blocks): + full_fm.append(block) + if idx + 1 < len(fm_blocks): + full_fm.append(_ansatz_layer(register, ansatz_config, idx)) + + return chain(*full_fm) + + +def _create_observable( + register: int | Register, + config: ObservableConfig, +) -> AbstractBlock: + """ + Create an observable block. + + Args: + register (int | Register): Number of qubits or a register object. + config (ObservableConfig): Observable configuration. + + Returns: + AbstractBlock: The observable block. + """ + return hamiltonian_factory( + register=register, + detuning=config.detuning, + detuning_strength=config.detuning_strength, + ) + + +def build_qnn_model( + register: int | Register, + fm_config: FeatureMapConfig, + ansatz_config: AnsatzConfig, + observable_config: ObservableConfig, +) -> QNN: + """ + Build a QNN model. + + Args: + register (int | Register): Number of qubits or a register object. + fm_config (FeatureMapConfig): Feature map configuration. + ansatz_config (AnsatzConfig): Ansatz configuration. + observable_config (ObservableConfig): Observable configuration. + + Returns: + QNN: A QNN model. + """ + fm_blocks = _create_fm(register=register, config=fm_config) + full_fm = _interleave_ansatz_in_fm( + register=register, + fm_blocks=fm_blocks, + ansatz_config=ansatz_config, + ) + + ansatz = _create_ansatz(register=register, config=ansatz_config) + + # Add a block before the Featuer Map to move from 0 state to an + # equal superposition of all states. This needs to be here only for rydberg + # feature map and only as long as the feature map is not updated to include + # a driving term in the Hamiltonian. + + if ansatz_config.ansatz_strategy == "rydberg": + num_qubits = register if isinstance(register, int) else register.n_qubits + mixing_block = kron(*[H(i) for i in range(num_qubits)]) + full_fm = chain(mixing_block, full_fm) + + circ = QuantumCircuit( + register, + full_fm, + ansatz, + ) + + observable = _create_observable(register=register, config=observable_config) + + ufa = QNN(circ, observable, inputs=fm_config.inputs) + + return ufa From 460182d22a6be370c8180362d7e4c8436775f4d4 Mon Sep 17 00:00:00 2001 From: smit Date: Tue, 9 Apr 2024 13:49:52 +0200 Subject: [PATCH 2/3] Linting fixed --- .../constructors/ufa_models/config.py | 71 ++++++++++++------- .../constructors/ufa_models/quantum_models.py | 19 ++--- 2 files changed, 50 insertions(+), 40 deletions(-) diff --git a/qadence_libs/constructors/ufa_models/config.py b/qadence_libs/constructors/ufa_models/config.py index fd27e01..e90d20e 100644 --- a/qadence_libs/constructors/ufa_models/config.py +++ b/qadence_libs/constructors/ufa_models/config.py @@ -1,15 +1,15 @@ from __future__ import annotations from dataclasses import dataclass, field -from sympy import Basic from typing import Callable, Type +from qadence.blocks.analog import AnalogBlock from qadence.blocks.primitive import ParametricBlock from qadence.logger import get_logger -from qadence.blocks.analog import AnalogBlock -from qadence.operations import AnalogRX, RX, Z +from qadence.operations import RX, AnalogRX, Z from qadence.parameters import Parameter from qadence.types import BasisSet, ReuploadScaling, TArray +from sympy import Basic logger = get_logger(__file__) @@ -21,7 +21,9 @@ class FeatureMapConfig: basis_set: BasisSet | list[BasisSet] """ - Basis set for feature encoding. Takes qadence.BasisSet. + Basis set for feature encoding. + + Takes qadence.BasisSet. Give a single BasisSet to use the same for all features. Give a list of BasisSet to apply each item for a corresponding variable. BasisSet.FOURIER for Fourier encoding. @@ -30,8 +32,10 @@ class FeatureMapConfig: reupload_scaling: ReuploadScaling | list[ReuploadScaling] """ - Scaling for encoding the same feature on different qubits in the - same layer of the feature maps. Takes qadence.ReuploadScaling. + Scaling for data reupload in the feature map. + + This for encoding the same feature on different qubits in the same layer of + the feature maps. Takes qadence.ReuploadScaling. Give a single ReuploadScaling to use the same for all features. Give a list of ReuploadScaling to apply each item for a corresponding variable. ReuploadScaling.CONSTANT for constant scaling. @@ -42,20 +46,24 @@ class FeatureMapConfig: feature_range: tuple[float, float] | list[tuple[float, float]] """ Range of data that the input data is assumed to come from. + Give a single tuple to use the same range for all features. Give a list of tuples to use each item for a corresponding variable. """ - target_range: tuple[float, float] | list[tuple[float, float]] | None = None + target_range: tuple[float, float] | list[tuple[float, float] | None] | None = None """ Range of data the data encoder assumes as natural range. + Give a single tuple to use the same range for all features. Give a list of tuples to use each item for a corresponding variable. """ multivariate_strategy: str = "parallel" """ - The encoding strategy in case of multi-variate function. If "parallel", + The encoding strategy in case of multi-variate function. + + If "parallel", the features are encoded in one block of rotation gates. with each feature given an equal number of qubits. If "serial", the features are encoded sequentially, with a HEA block between. "parallel" is allowed @@ -63,29 +71,41 @@ class FeatureMapConfig: """ feature_map_strategy: str = "digital" - """Strategy for feature map. Accepts 'digital', 'analog' or 'rydberg'. Defaults to "digital".""" + """Strategy for feature map. + + Accepts 'digital', 'analog' or 'rydberg'. Defaults to "digital". + """ param_prefix: str = "phi" - """The base name of the feature map parameter. Defaults to `phi`.""" + """The base name of the feature map parameter. + + Defaults to `phi`. + """ num_repeats: int | list[int] = 0 """ - Number of feature map layers repeated in the data reuploadig step. If all are to be \ - repeated the same number of times, then can give a single `int`. For different number \ - of repeatitions for each feature, provide a list of `int`s corresponding to desired \ - number of repeatitions. This amounts to the number of additional reuploads. So if \ - `num_repeats` is N, the data gets uploaded N+1 times. Defaults to no repeatition. + Number of feature map layers repeated in the data reuploadig step. + + If all features are to be repeated the same number of times, then can give + a single `int`. For different number of repeatitions for each feature, + provide a list of `int`s corresponding to desired number of repeatitions. + This amounts to the number of additional reuploads. So if `num_repeats` is + N, the data gets uploaded N+1 times. Defaults to no repeatition. """ operation: Callable[[Parameter | Basic], AnalogBlock] | Type[RX] | None = None """ - type of operation. Choose among the analog or digital rotations or a custom + Type of operation. + + Choose among the analog or digital rotations or a custom callable function returning an AnalogBlock instance """ inputs: list[Basic | str] | None = None """ - List that indicates the order of variables of the tensors that are passed + Order of variables in the input tensor. + + List that indicates the order of variables of the tensors that are passed. to the model. Given input tensors `xs = torch.rand(batch_size, input_size:=2)` a QNN with `inputs=("t", "x")` will assign `t, x = xs[:,0], xs[:,1]`. """ @@ -134,9 +154,7 @@ def __post_init__(self) -> None: use same basis for all features." if isinstance(self.reupload_scaling, ReuploadScaling): - self.reupload_scaling = [ - self.reupload_scaling for i in range(self.num_features) - ] + self.reupload_scaling = [self.reupload_scaling for i in range(self.num_features)] else: assert ( len(self.reupload_scaling) == self.num_features @@ -183,15 +201,15 @@ class AnsatzConfig: """Number of layers of the ansatz.""" ansatz_type: str - """ - What type of ansatz. + """What type of ansatz. + "hea" for Hardware Efficient Ansatz. "iia" for Identity intialized Ansatz. """ ansatz_strategy: str - """ - Ansatz strategy. + """Ansatz strategy. + "digital" for fully digital ansatz. Required if `ansatz_type` is `iia`. "sdaqc" for analog entangling block. "rydberg" for fully rydberg hea ansatz. @@ -200,6 +218,7 @@ class AnsatzConfig: strategy_args: dict = field(default_factory=dict) """ A dictionary containing keyword arguments to the function creating the ansatz. + Details about each below. For "digital" strategy, accepts the following: @@ -250,13 +269,15 @@ class ObservableConfig: detuning: Type[Z] = Z """ Single qubit detuning of the observable Hamiltonian. + Accepts single-qubit operator N, X, Y, or Z. Defaults to Z. """ detuning_strength: TArray | str | None = None """ - list of values to be used as the detuning strength for each qubit. + List of values to be used as the detuning strength for each qubit. + Alternatively, some string "x" can be passed, which will create a parameterized detuning for each qubit, each labelled as `"x_i"`. Defaults to 1.0 for each qubit. diff --git a/qadence_libs/constructors/ufa_models/quantum_models.py b/qadence_libs/constructors/ufa_models/quantum_models.py index b317227..ad0ecb8 100644 --- a/qadence_libs/constructors/ufa_models/quantum_models.py +++ b/qadence_libs/constructors/ufa_models/quantum_models.py @@ -1,7 +1,7 @@ from __future__ import annotations import numpy as np - +from eic_aero.ufa_models.config import AnsatzConfig, FeatureMapConfig, ObservableConfig from qadence.blocks import chain, kron from qadence.blocks.abstract import AbstractBlock from qadence.blocks.composite import ChainBlock @@ -21,8 +21,6 @@ from qadence.register import Register from qadence.types import Interaction, ReuploadScaling, Strategy -from eic_aero.ufa_models.config import AnsatzConfig, FeatureMapConfig, ObservableConfig - def _create_support_arrays( num_qubits: int, @@ -51,10 +49,7 @@ def _create_support_arrays( return [tuple(range(num_qubits)) for i in range(num_features)] elif multivariate_strategy == "parallel": if num_features <= num_qubits: - return [ - tuple(x.tolist()) - for x in np.array_split(np.arange(num_qubits), num_features) - ] + return [tuple(x.tolist()) for x in np.array_split(np.arange(num_qubits), num_features)] else: raise ValueError( f"Number of features {num_features} must be less than or equal to the number of \ @@ -275,9 +270,7 @@ def _encode_feature_rydberg( return rydberg_tower_feature_map(n_qubits=num_qubits, param=param) else: - raise NotImplementedError( - f"Rydberg feature map not implemented for {reupload_scaling}" - ) + raise NotImplementedError(f"Rydberg feature map not implemented for {reupload_scaling}") def _create_rydberg_fm( @@ -528,11 +521,7 @@ def _create_hea_rydberg( Returns: AbstractBlock: The Rydberg Hardware Efficient Ansatz. """ - register = ( - register - if isinstance(register, Register) - else Register.circle(n_qubits=register) - ) + register = register if isinstance(register, Register) else Register.circle(n_qubits=register) addressable_detuning = config.strategy_args.get("addressable_detuning", True) addressable_drive = config.strategy_args.get("addressable_drive", False) From f840e5575ab394cce24e8391b419fb83d4fe5710 Mon Sep 17 00:00:00 2001 From: smit Date: Mon, 15 Apr 2024 09:27:15 +0200 Subject: [PATCH 3/3] Fixed relative imports of configs --- qadence_libs/constructors/ufa_models/quantum_models.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/qadence_libs/constructors/ufa_models/quantum_models.py b/qadence_libs/constructors/ufa_models/quantum_models.py index ad0ecb8..0dd5fb0 100644 --- a/qadence_libs/constructors/ufa_models/quantum_models.py +++ b/qadence_libs/constructors/ufa_models/quantum_models.py @@ -1,7 +1,6 @@ from __future__ import annotations import numpy as np -from eic_aero.ufa_models.config import AnsatzConfig, FeatureMapConfig, ObservableConfig from qadence.blocks import chain, kron from qadence.blocks.abstract import AbstractBlock from qadence.blocks.composite import ChainBlock @@ -21,6 +20,8 @@ from qadence.register import Register from qadence.types import Interaction, ReuploadScaling, Strategy +from .config import AnsatzConfig, FeatureMapConfig, ObservableConfig + def _create_support_arrays( num_qubits: int,