diff --git a/README.md b/README.md index bc46c6ee..51598c44 100644 --- a/README.md +++ b/README.md @@ -220,19 +220,13 @@ The _QPAM_ scheme's encoding only preserves **num_samples** (_int_) and the norm > The essential keys required for decoding with any scheme can be checked from the scheme's `.keys` attribute. ### Using Custom Functions -The `decode` and `stream` operations can be configured with the following custom functions. They require few mandatory arguments followed by custom preceding keyword arguments (denoted as `**kwargs`). +The `stream` operation can be configured with a custom process function: - **Process Function**: The default process function of `stream()` simply encodes and decodes a chunk of data with default parameters. It can be overriden by passing a custom function to the `process_function=` parameter. The mandatory arguments for the custom process function are `data=` and `scheme=`. ```python processed_data = process_function(data, scheme, **kwargs) ``` -- **Execute Function** : -The default execute function for `decode()` can be overriden by passing a custom function to the `execute_function=` parameter. The mandatory argument for the custom execute function is `circuit=`. (QPAM also expects `shots=` since it's a metadata) -```python -result = execute_function(circuit, **kwargs) -``` -**Example**: An optional execute function is included in the package which uses [Sampler Primitive](https://docs.quantum.ibm.com/api/qiskit-ibm-runtime/qiskit_ibm_runtime.SamplerV2): `quantumaudio.utils.execute_with_sampler` that can be passed to the `decode()` method. It requires the dependency `pip install qiskit-ibm-runtime`.
- + ## 📘 Additional Resources ### Notebook Examples For examples of circuit preparation, signal reconstruction, and interactive demonstrations, please check the [Demo Notebook](https://github.com/moth-quantum/quantum-audio/blob/main/demos/1_Basics_Walkthrough.ipynb). It combines the core package with additional functions from the `demos/tools` folder to go through Visual and Digital Audio examples. diff --git a/pyproject.toml b/pyproject.toml index 2b9aae41..176b1928 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,6 +40,7 @@ Repository = "https://github.com/moth-quantum/quantum-audio" Documentation = "https://quantumaudio.readthedocs.io/" [project.optional-dependencies] +cirq = ["cirq-core"] demos = [ "soundfile==0.12.1", "librosa==0.10.2.post1", diff --git a/quantumaudio/__init__.py b/quantumaudio/__init__.py index d4c8055f..1b1de485 100644 --- a/quantumaudio/__init__.py +++ b/quantumaudio/__init__.py @@ -117,6 +117,7 @@ def __dir__(): "schemes", "utils", "tools", + "backends", "load_scheme", "encode", "decode", diff --git a/quantumaudio/backends/__init__.py b/quantumaudio/backends/__init__.py new file mode 100644 index 00000000..f9f76b52 --- /dev/null +++ b/quantumaudio/backends/__init__.py @@ -0,0 +1,58 @@ +# Copyright 2024 Moth Quantum +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========================================================================== + +"""Framework-agnostic backend layer for quantumaudio. + +This subpackage provides a backend abstraction that allows quantum +circuits built by quantumaudio schemes to be executed on different +quantum frameworks (Qiskit, Cirq, and others). +""" + +from quantumaudio.backends._optional import is_available, require +from quantumaudio.backends.core import ( + CircuitSpec, + GateOp, + GateType, + UnifiedResult, + Backend, + registry, +) + +# Trigger auto-registration of installed backends. +import quantumaudio.backends.providers # noqa: F401 + + +def available_backends() -> list[str]: + """Return names of all backends whose dependencies are installed.""" + return registry.available() + + +def get_backend(name: str = "qiskit") -> Backend: + """Instantiate and return a backend by name.""" + return registry.get(name) + + +__all__ = [ + "Backend", + "CircuitSpec", + "GateOp", + "GateType", + "UnifiedResult", + "available_backends", + "get_backend", + "is_available", + "registry", + "require", +] diff --git a/quantumaudio/backends/_optional.py b/quantumaudio/backends/_optional.py new file mode 100644 index 00000000..477f055d --- /dev/null +++ b/quantumaudio/backends/_optional.py @@ -0,0 +1,56 @@ +# Copyright 2024 Moth Quantum +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========================================================================== + +"""Helpers for guarding optional backend dependencies.""" + +from __future__ import annotations + +import importlib +import importlib.util +from types import ModuleType + + +def is_available(package: str) -> bool: + """Check whether *package* can be imported without importing it.""" + return importlib.util.find_spec(package) is not None + + +def require( + package: str, extras_name: str | None = None +) -> ModuleType: + """Import and return *package*, raising a helpful error if missing. + + Only a missing top-level package is converted into the "install + quantumaudio[]" hint. Import errors raised inside an + installed dependency (e.g. due to a broken transitive dep) are + re-raised unchanged so the original traceback is preserved. + + Args: + package: Dotted module name, e.g. ``"cirq"`` or ``"qiskit"``. + extras_name: Optional pip extras hint used in the error message. + """ + top_level = package.split(".", 1)[0] + try: + return importlib.import_module(package) + except ModuleNotFoundError as e: + # Only convert "the requested package itself is missing" into + # the install hint; let nested missing modules propagate. + if e.name in (package, top_level): + hint = extras_name or top_level + raise ImportError( + f"{package!r} is required but not installed. " + f"Install it with: pip install quantumaudio[{hint}]" + ) from None + raise diff --git a/quantumaudio/backends/core/__init__.py b/quantumaudio/backends/core/__init__.py new file mode 100644 index 00000000..33ca420a --- /dev/null +++ b/quantumaudio/backends/core/__init__.py @@ -0,0 +1,33 @@ +# Copyright 2024 Moth Quantum +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========================================================================== + +from quantumaudio.backends.core.types import GateType +from quantumaudio.backends.core.circuit import CircuitSpec, GateOp +from quantumaudio.backends.core.result import UnifiedResult +from quantumaudio.backends.core.backend import ( + Backend, + BackendRegistry, + registry, +) + +__all__ = [ + "Backend", + "BackendRegistry", + "CircuitSpec", + "GateOp", + "GateType", + "UnifiedResult", + "registry", +] diff --git a/quantumaudio/backends/core/backend.py b/quantumaudio/backends/core/backend.py new file mode 100644 index 00000000..0f3d9e03 --- /dev/null +++ b/quantumaudio/backends/core/backend.py @@ -0,0 +1,86 @@ +# Copyright 2024 Moth Quantum +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========================================================================== + +"""Backend abstract base class and registry.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import Any + +import numpy as np + +from quantumaudio.backends.core.circuit import CircuitSpec +from quantumaudio.backends.core.result import UnifiedResult + + +class Backend(ABC): + """Interface that every quantum backend must implement.""" + + name: str + + @abstractmethod + def build_circuit(self, spec: CircuitSpec) -> Any: + """Translate a CircuitSpec into the backend's native circuit.""" + + @abstractmethod + def run( + self, native_circuit: Any, shots: int = 1024 + ) -> UnifiedResult: + """Execute a native circuit and return a UnifiedResult.""" + + @abstractmethod + def statevector(self, native_circuit: Any) -> np.ndarray: + """Return the exact statevector (simulator only).""" + + def run_spec( + self, spec: CircuitSpec, shots: int = 1024 + ) -> UnifiedResult: + """Convenience: build and run in one call.""" + native = self.build_circuit(spec) + return self.run(native, shots) + + +class BackendRegistry: + """Singleton that tracks available backends.""" + + _instance: BackendRegistry | None = None + _backends: dict[str, type[Backend]] + + def __new__(cls) -> BackendRegistry: + if cls._instance is None: + cls._instance = super().__new__(cls) + cls._instance._backends = {} + return cls._instance + + def register(self, name: str, cls_: type[Backend]) -> None: + self._backends[name] = cls_ + + def get(self, name: str) -> Backend: + """Instantiate and return a backend by name.""" + if name not in self._backends: + available = ", ".join(self._backends) or "(none)" + raise KeyError( + f"Backend {name!r} not available. " + f"Installed: {available}" + ) + return self._backends[name]() + + def available(self) -> list[str]: + """Return names of all registered backends.""" + return list(self._backends) + + +registry = BackendRegistry() diff --git a/quantumaudio/backends/core/circuit.py b/quantumaudio/backends/core/circuit.py new file mode 100644 index 00000000..edc0e8ad --- /dev/null +++ b/quantumaudio/backends/core/circuit.py @@ -0,0 +1,308 @@ +# Copyright 2024 Moth Quantum +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========================================================================== + +"""Framework-agnostic quantum circuit description.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Sequence + +import numpy as np + +from quantumaudio.backends.core.types import GateType + + +@dataclass(frozen=True) +class GateOp: + """A single gate operation. + + ``params`` carries gate angles (e.g. ``RY(theta)``); ``clbits`` + carries the classical-bit targets for measurements. Keeping the + two separate avoids overloading ``params`` for non-angle data. + """ + + gate: GateType + qubits: tuple[int, ...] + params: tuple[float, ...] = () + clbits: tuple[int, ...] = () + + +@dataclass +class CircuitSpec: + """Framework-agnostic quantum circuit specification. + + This is a pure data structure. Backends translate it into their + native circuit representation for execution. The builder API + mirrors the gate-method style used by Qiskit and MicroMoth so + that circuits read naturally. + """ + + num_qubits: int + num_clbits: int = 0 + ops: list[GateOp] = field(default_factory=list) + metadata: dict = field(default_factory=dict) + name: str = "" + + # -- single-qubit gates ----------------------------------------------- + + def h(self, qubit: int) -> CircuitSpec: + self.ops.append(GateOp(GateType.H, (qubit,))) + return self + + def x(self, qubit: int) -> CircuitSpec: + self.ops.append(GateOp(GateType.X, (qubit,))) + return self + + def y(self, qubit: int) -> CircuitSpec: + self.ops.append(GateOp(GateType.Y, (qubit,))) + return self + + def z(self, qubit: int) -> CircuitSpec: + self.ops.append(GateOp(GateType.Z, (qubit,))) + return self + + def s(self, qubit: int) -> CircuitSpec: + self.ops.append(GateOp(GateType.S, (qubit,))) + return self + + def t(self, qubit: int) -> CircuitSpec: + self.ops.append(GateOp(GateType.T, (qubit,))) + return self + + def rx(self, theta: float, qubit: int) -> CircuitSpec: + self.ops.append(GateOp(GateType.RX, (qubit,), (theta,))) + return self + + def ry(self, theta: float, qubit: int) -> CircuitSpec: + self.ops.append(GateOp(GateType.RY, (qubit,), (theta,))) + return self + + def rz(self, theta: float, qubit: int) -> CircuitSpec: + self.ops.append(GateOp(GateType.RZ, (qubit,), (theta,))) + return self + + # -- two-qubit gates --------------------------------------------------- + + def cx(self, control: int, target: int) -> CircuitSpec: + self.ops.append(GateOp(GateType.CX, (control, target))) + return self + + def cz(self, control: int, target: int) -> CircuitSpec: + self.ops.append(GateOp(GateType.CZ, (control, target))) + return self + + def crx( + self, theta: float, control: int, target: int + ) -> CircuitSpec: + self.ops.append( + GateOp(GateType.CRX, (control, target), (theta,)) + ) + return self + + def cry( + self, theta: float, control: int, target: int + ) -> CircuitSpec: + self.ops.append( + GateOp(GateType.CRY, (control, target), (theta,)) + ) + return self + + def crz( + self, theta: float, control: int, target: int + ) -> CircuitSpec: + self.ops.append( + GateOp(GateType.CRZ, (control, target), (theta,)) + ) + return self + + def swap(self, qubit1: int, qubit2: int) -> CircuitSpec: + self.ops.append(GateOp(GateType.SWAP, (qubit1, qubit2))) + return self + + # -- multi-controlled gates -------------------------------------------- + + def mcx( + self, controls: Sequence[int], target: int + ) -> CircuitSpec: + """Multi-controlled X gate.""" + self.ops.append( + GateOp(GateType.MCX, (*tuple(controls), target)) + ) + return self + + def mcry( + self, theta: float, controls: Sequence[int], target: int + ) -> CircuitSpec: + """Multi-controlled RY gate.""" + self.ops.append( + GateOp( + GateType.MCRY, + (*tuple(controls), target), + (theta,), + ) + ) + return self + + # -- state preparation ------------------------------------------------- + + def initialize(self, state: Sequence[float]) -> CircuitSpec: + """Prepare the register in the given non-negative real state. + + Decomposes the state preparation into RY and CX gates using + the Mottonen method so that all backends only need to handle + elementary gates. The decomposition uses magnitudes via + ``arctan2`` and therefore only encodes non-negative real + amplitudes (e.g. QPAM probability amplitudes); signed or + complex states are not supported. + """ + state = np.asarray(state, dtype=float) + n = self.num_qubits + expected = 2**n + if len(state) != expected: + raise ValueError( + f"State vector length {len(state)} does not match " + f"the expected number of amplitudes " + f"2**{n} = {expected}" + ) + norm = np.linalg.norm(state) + if norm < 1e-15: + return self + state = state / norm + _apply_mottonen(self, state, n) + return self + + # -- measurement ------------------------------------------------------- + + def measure(self, qubit: int, clbit: int) -> CircuitSpec: + self.ops.append( + GateOp(GateType.MEASURE, (qubit,), clbits=(clbit,)) + ) + self.num_clbits = max(self.num_clbits, clbit + 1) + return self + + def measure_all(self) -> CircuitSpec: + """Measure every qubit into a classical bit of the same index.""" + for q in range(self.num_qubits): + self.measure(q, q) + return self + + # -- visual separator -------------------------------------------------- + + def barrier(self) -> CircuitSpec: + self.ops.append(GateOp(GateType.BARRIER, ())) + return self + + +# ====================================================================== +# Mottonen state-preparation decomposition (non-negative real states) +# ====================================================================== + + +def _apply_mottonen(spec: CircuitSpec, state: np.ndarray, n: int): + """Decompose a non-negative real state vector into RY and CX gates. + + Uses the Mottonen method: compute a tree of rotation angles, + then apply uniformly controlled RY rotations from the most + significant qubit down to the least significant. + + The qubit convention follows Qiskit's little-endian ordering: + qubit 0 is the least significant bit of the state index. + + Angles are derived from amplitude magnitudes, so signed inputs + are encoded as their absolute values. + """ + angles_tree = _compute_angles(state, n) + for level in range(n): + target = n - 1 - level + controls = list(range(target + 1, n)) + # Reverse controls so the UCR's recursive even/odd split + # matches the angle indexing order. + controls.reverse() + thetas = angles_tree[level] + _apply_ucry(spec, thetas, controls, target) + + +def _compute_angles(state: np.ndarray, n: int) -> list[list[float]]: + """Compute the rotation angle tree for Mottonen decomposition. + + Level 0 targets the MSB (qubit n-1) with 1 angle; level k + targets qubit n-1-k with 2^k angles. Each angle splits a + pair of amplitude sub-blocks via + theta = 2 * arctan2(norm_lower, norm_upper). + """ + all_angles = [] + for level in range(n): + k = n - 1 - level + stride = 2 ** (k + 1) + half = stride // 2 + num_pairs = len(state) // stride + angles = [] + for j in range(num_pairs): + start = j * stride + upper = state[start : start + half] + lower = state[start + half : start + stride] + norm_upper = np.linalg.norm(upper) + norm_lower = np.linalg.norm(lower) + if norm_upper + norm_lower < 1e-15: + angles.append(0.0) + else: + angles.append( + 2.0 * np.arctan2(norm_lower, norm_upper) + ) + all_angles.append(angles) + return all_angles + + +def _apply_ucry( + spec: CircuitSpec, + thetas: list[float], + controls: list[int], + target: int, +): + """Apply a uniformly controlled RY rotation. + + Recursively decomposes into CX and RY gates: + UCR_Y(theta_0, ..., theta_{2^k-1}) = + UCR_Y(alpha_even) @ CX(last_ctrl, target) + @ UCR_Y(alpha_odd) @ CX(last_ctrl, target) + + where alpha_even[i] = (theta[2i] + theta[2i+1]) / 2 + and alpha_odd[i] = (theta[2i] - theta[2i+1]) / 2. + + Base case (no controls): a single RY(theta, target). + """ + if not controls: + # Base case: single rotation. + if abs(thetas[0]) > 1e-15: + spec.ry(thetas[0], target) + return + + n = len(thetas) + even = [ + (thetas[2 * i] + thetas[2 * i + 1]) / 2 + for i in range(n // 2) + ] + odd = [ + (thetas[2 * i] - thetas[2 * i + 1]) / 2 + for i in range(n // 2) + ] + last_ctrl = controls[-1] + remaining = controls[:-1] + + _apply_ucry(spec, even, remaining, target) + spec.cx(last_ctrl, target) + _apply_ucry(spec, odd, remaining, target) + spec.cx(last_ctrl, target) diff --git a/quantumaudio/backends/core/result.py b/quantumaudio/backends/core/result.py new file mode 100644 index 00000000..4c55bf43 --- /dev/null +++ b/quantumaudio/backends/core/result.py @@ -0,0 +1,123 @@ +# Copyright 2024 Moth Quantum +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========================================================================== + +"""Unified result type that normalises output across backends.""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +import numpy as np + +from quantumaudio.backends._optional import require + + +@dataclass +class UnifiedResult: + """Backend-agnostic execution result. + + Every backend produces one of these so that downstream code never + has to care which framework ran the circuit. + """ + + counts: dict[str, int] + shots: int + backend_name: str + metadata: dict = field(default_factory=dict) + + def __post_init__(self) -> None: + if self.shots <= 0: + raise ValueError( + f"shots must be positive, got {self.shots}. " + "Statevector-only execution should not produce a " + "UnifiedResult." + ) + + def probabilities(self) -> dict[str, float]: + """Return the probability distribution over bitstrings.""" + return {k: v / self.shots for k, v in self.counts.items()} + + def probabilities_array(self) -> np.ndarray: + """Return probabilities as a dense array indexed by integer + bitstring value.""" + if not self.counts: + return np.array([]) + n_bits = len(next(iter(self.counts))) + probs = np.zeros(2**n_bits) + for bitstring, count in self.counts.items(): + probs[int(bitstring, 2)] = count / self.shots + return probs + + def marginal(self, qubits: list[int]) -> UnifiedResult: + """Marginalise the result over the given qubit indices.""" + if not self.counts: + return UnifiedResult( + {}, self.shots, self.backend_name, self.metadata + ) + n_bits = len(next(iter(self.counts))) + new_counts: dict[str, int] = {} + for bitstring, count in self.counts.items(): + # Qubits are indexed from the right in the bitstring. + new_key = "".join( + bitstring[n_bits - 1 - q] + for q in sorted(qubits, reverse=True) + ) + new_counts[new_key] = ( + new_counts.get(new_key, 0) + count + ) + return UnifiedResult( + new_counts, self.shots, self.backend_name, self.metadata + ) + + def to_qiskit_result(self): + """Construct a ``qiskit.result.Result`` for compatibility with + code that does isinstance checks on Qiskit result types.""" + require("qiskit", extras_name="qiskit") + from qiskit.result import Result # noqa: PLC0415 + from qiskit.result.models import ( # noqa: PLC0415 + ExperimentResult, + ExperimentResultData, + ) + + if self.counts: + n_bits = len(next(iter(self.counts))) + hex_width = (n_bits + 3) // 4 + hex_counts = { + "0x" + + format(int(k, 2), f"0{hex_width}x"): v + for k, v in self.counts.items() + } + else: + hex_counts = {} + n_bits = 0 + + exp_data = ExperimentResultData(counts=hex_counts) + exp_result = ExperimentResult( + shots=self.shots, + success=True, + data=exp_data, + header={ + "metadata": self.metadata, + "memory_slots": n_bits, + }, + ) + return Result( + backend_name=self.backend_name, + backend_version="0.0.0", + qobj_id="quantumaudio-backends", + job_id="quantumaudio-backends", + success=True, + results=[exp_result], + ) diff --git a/quantumaudio/backends/core/types.py b/quantumaudio/backends/core/types.py new file mode 100644 index 00000000..90dbacd0 --- /dev/null +++ b/quantumaudio/backends/core/types.py @@ -0,0 +1,55 @@ +# Copyright 2024 Moth Quantum +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========================================================================== + +"""Gate types for framework-agnostic circuit descriptions.""" + +from enum import Enum, auto + + +class GateType(Enum): + """Universal gate set for quantum audio circuits. + + Covers every operation used by the quantumaudio scheme + implementations across all supported backends. + """ + + # Single-qubit gates. + H = auto() + X = auto() + Y = auto() + Z = auto() + S = auto() + T = auto() + RX = auto() + RY = auto() + RZ = auto() + + # Two-qubit gates. + CX = auto() + CZ = auto() + CRX = auto() + CRY = auto() + CRZ = auto() + SWAP = auto() + + # Multi-controlled gates. + MCX = auto() + MCRY = auto() + + # Measurement. + MEASURE = auto() + + # Visual separator (no-op for execution). + BARRIER = auto() diff --git a/quantumaudio/backends/providers/__init__.py b/quantumaudio/backends/providers/__init__.py new file mode 100644 index 00000000..a74e5ef2 --- /dev/null +++ b/quantumaudio/backends/providers/__init__.py @@ -0,0 +1,33 @@ +# Copyright 2024 Moth Quantum +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========================================================================== + +"""Auto-registration of available backend providers.""" + +from quantumaudio.backends._optional import is_available +from quantumaudio.backends.core.backend import registry + +if is_available("qiskit"): + from quantumaudio.backends.providers.qiskit_backend import ( + QiskitBackend, + ) + + registry.register("qiskit", QiskitBackend) + +if is_available("cirq"): + from quantumaudio.backends.providers.cirq_backend import ( + CirqBackend, + ) + + registry.register("cirq", CirqBackend) diff --git a/quantumaudio/backends/providers/cirq_backend.py b/quantumaudio/backends/providers/cirq_backend.py new file mode 100644 index 00000000..e5bb625b --- /dev/null +++ b/quantumaudio/backends/providers/cirq_backend.py @@ -0,0 +1,174 @@ +# Copyright 2024 Moth Quantum +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========================================================================== + +"""Cirq backend for quantumaudio.""" + +from __future__ import annotations + +import numpy as np + +from quantumaudio.backends._optional import require +from quantumaudio.backends.core.backend import Backend +from quantumaudio.backends.core.circuit import CircuitSpec, GateOp +from quantumaudio.backends.core.result import UnifiedResult +from quantumaudio.backends.core.types import GateType + +cirq = require("cirq", extras_name="cirq") + + +def _qubits(n: int): + """Create a list of Cirq LineQubits.""" + return cirq.LineQubit.range(n) + + +_SINGLE_GATE = { + GateType.H: lambda: cirq.H, + GateType.X: lambda: cirq.X, + GateType.Y: lambda: cirq.Y, + GateType.Z: lambda: cirq.Z, + GateType.S: lambda: cirq.S, + GateType.T: lambda: cirq.T, +} + +_PARAM_GATE = { + GateType.RX: cirq.rx, + GateType.RY: cirq.ry, + GateType.RZ: cirq.rz, +} + +_TWO_GATE = { + GateType.CX: lambda: cirq.CNOT, + GateType.CZ: lambda: cirq.CZ, + GateType.SWAP: lambda: cirq.SWAP, +} + +_CTRL_PARAM = { + GateType.CRX: cirq.rx, + GateType.CRY: cirq.ry, + GateType.CRZ: cirq.rz, +} + + +def _apply_op(qubits, op: GateOp) -> list: # noqa: PLR0911 + """Return a list of Cirq operations for a single GateOp.""" + g = op.gate + idx = op.qubits + + if g in _SINGLE_GATE: + return [_SINGLE_GATE[g]()(qubits[idx[0]])] + if g in _PARAM_GATE: + return [_PARAM_GATE[g](op.params[0])(qubits[idx[0]])] + if g in _TWO_GATE: + return [ + _TWO_GATE[g]()(qubits[idx[0]], qubits[idx[1]]) + ] + if g in _CTRL_PARAM: + return [ + _CTRL_PARAM[g](op.params[0])( + qubits[idx[1]] + ).controlled_by(qubits[idx[0]]) + ] + if g == GateType.MCX: + ctrls = [qubits[i] for i in idx[:-1]] + return [cirq.X(qubits[idx[-1]]).controlled_by(*ctrls)] + if g == GateType.MCRY: + ctrls = [qubits[i] for i in idx[:-1]] + return [ + cirq.ry(op.params[0])( + qubits[idx[-1]] + ).controlled_by(*ctrls) + ] + if g == GateType.MEASURE: + return [ + cirq.measure(qubits[idx[0]], key=str(idx[0])) + ] + # BARRIER and anything else: no-op. + return [] + + +class CirqBackend(Backend): + """Google Cirq simulator backend.""" + + name = "cirq" + + def build_circuit( + self, spec: CircuitSpec + ) -> dict: + """Build a Cirq circuit from a CircuitSpec. + + Returns a dict with the circuit, qubits, and metadata + for use by :meth:`run` and :meth:`statevector`. + """ + n = spec.num_qubits + qubits = _qubits(n) + moments_ops = [] + for op in spec.ops: + ops = _apply_op(qubits, op) + moments_ops.extend(ops) + circuit = cirq.Circuit(moments_ops) + return { + "circuit": circuit, + "qubits": qubits, + "num_qubits": n, + "metadata": { + k: v + for k, v in spec.metadata.items() + if k != "registers" + }, + } + + def run( + self, native_circuit: dict, shots: int = 1024 + ) -> UnifiedResult: + """Execute the circuit and return UnifiedResult.""" + circuit = native_circuit["circuit"] + n = native_circuit["num_qubits"] + metadata = native_circuit["metadata"] + + sim = cirq.Simulator() + result = sim.run(circuit, repetitions=shots) + + # Build bitstrings from per-qubit measurement keys. + # Reverse to match Qiskit's little-endian convention: + # qubit 0 is the rightmost bit. + counts: dict[str, int] = {} + for i in range(shots): + bits = [] + for q in range(n - 1, -1, -1): + key = str(q) + bits.append( + str(result.measurements[key][i, 0]) + ) + bitstring = "".join(bits) + counts[bitstring] = counts.get(bitstring, 0) + 1 + + return UnifiedResult(counts, shots, self.name, metadata) + + def statevector(self, native_circuit: dict) -> np.ndarray: + """Return the exact statevector.""" + circuit = native_circuit["circuit"] + + # Strip measurement gates for statevector computation. + clean_ops = [ + op + for moment in circuit.moments + for op in moment.operations + if not cirq.is_measurement(op) + ] + clean = cirq.Circuit(clean_ops) + + sim = cirq.Simulator() + sv_result = sim.simulate(clean) + return sv_result.final_state_vector diff --git a/quantumaudio/backends/providers/qiskit_backend.py b/quantumaudio/backends/providers/qiskit_backend.py new file mode 100644 index 00000000..243495ab --- /dev/null +++ b/quantumaudio/backends/providers/qiskit_backend.py @@ -0,0 +1,184 @@ +# Copyright 2024 Moth Quantum +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========================================================================== + +"""Qiskit backend for quantumaudio.""" + +from __future__ import annotations + +import numpy as np +import qiskit +import qiskit_aer +from qiskit import ClassicalRegister +from qiskit.circuit.library import RYGate +from qiskit.quantum_info import Statevector +from qiskit.transpiler.preset_passmanagers import ( + generate_preset_pass_manager, +) + +from quantumaudio.backends.core.backend import Backend +from quantumaudio.backends.core.circuit import CircuitSpec, GateOp +from quantumaudio.backends.core.result import UnifiedResult +from quantumaudio.backends.core.types import GateType + + +_SINGLE_QUBIT = { + GateType.H: "h", + GateType.X: "x", + GateType.Y: "y", + GateType.Z: "z", + GateType.S: "s", + GateType.T: "t", +} + +_SINGLE_QUBIT_PARAM = { + GateType.RX: "rx", + GateType.RY: "ry", + GateType.RZ: "rz", +} + +_TWO_QUBIT = { + GateType.CX: "cx", + GateType.CZ: "cz", + GateType.SWAP: "swap", +} + +_TWO_QUBIT_PARAM = { + GateType.CRX: "crx", + GateType.CRY: "cry", + GateType.CRZ: "crz", +} + + +def _apply_op(qc, qubits, op: GateOp): + """Apply a single GateOp to a Qiskit QuantumCircuit.""" + g = op.gate + idx = op.qubits + + if g in _SINGLE_QUBIT: + getattr(qc, _SINGLE_QUBIT[g])(qubits[idx[0]]) + elif g in _SINGLE_QUBIT_PARAM: + getattr(qc, _SINGLE_QUBIT_PARAM[g])( + op.params[0], qubits[idx[0]] + ) + elif g in _TWO_QUBIT: + getattr(qc, _TWO_QUBIT[g])( + qubits[idx[0]], qubits[idx[1]] + ) + elif g in _TWO_QUBIT_PARAM: + getattr(qc, _TWO_QUBIT_PARAM[g])( + op.params[0], qubits[idx[0]], qubits[idx[1]] + ) + elif g == GateType.MCX: + ctrls = [qubits[i] for i in idx[:-1]] + qc.mcx(ctrls, qubits[idx[-1]]) + elif g == GateType.MCRY: + ctrls = [qubits[i] for i in idx[:-1]] + controlled_ry = RYGate(op.params[0]).control( + len(ctrls) + ) + qc.append(controlled_ry, [*ctrls, qubits[idx[-1]]]) + elif g == GateType.MEASURE: + qc.measure(qubits[idx[0]], op.clbits[0]) + elif g == GateType.BARRIER: + qc.barrier() + + +class QiskitBackend(Backend): + """Qiskit / AerSimulator backend.""" + + name = "qiskit" + + def build_circuit(self, spec: CircuitSpec) -> qiskit.QuantumCircuit: + """Translate CircuitSpec to a Qiskit QuantumCircuit. + + If register metadata is present, named QuantumRegisters are + created for better circuit visualisation. + """ + regs_meta = spec.metadata.get("registers", {}) + if regs_meta: + regs = [] + qubit_list = [None] * spec.num_qubits + for reg_name, (start, size) in regs_meta.items(): + if size > 0: + reg = qiskit.QuantumRegister(size, reg_name) + regs.append(reg) + for i in range(size): + qubit_list[start + i] = reg[i] + qc = qiskit.QuantumCircuit( + *regs, name=spec.name + ) + # Map integer qubit indices to Qiskit qubit objects. + qubits = qc.qubits + else: + qc = qiskit.QuantumCircuit( + spec.num_qubits, spec.num_clbits, name=spec.name + ) + qubits = qc.qubits + + # Copy non-register metadata to the Qiskit circuit. + qc.metadata = { + k: v + for k, v in spec.metadata.items() + if k != "registers" + } + + for op in spec.ops: + if op.gate == GateType.MEASURE and not qc.cregs: + qc.add_register( + ClassicalRegister(spec.num_clbits) + ) + _apply_op(qc, qubits, op) + + return qc + + def run( + self, native_circuit, shots: int = 1024 + ) -> UnifiedResult: + """Execute on AerSimulator and return UnifiedResult.""" + backend = qiskit_aer.AerSimulator() + pm = generate_preset_pass_manager( + optimization_level=1, backend=backend + ) + transpiled = pm.run(native_circuit) + job = backend.run(transpiled, shots=shots) + result = job.result() + raw_counts = result.get_counts() + + # Normalise keys to zero-padded binary strings. + n_bits = native_circuit.num_clbits + counts: dict[str, int] = {} + for raw_key, val in raw_counts.items(): + key = raw_key.replace(" ", "") + if key.startswith("0x"): + bits = bin(int(key, 16))[2:].zfill(n_bits) + else: + bits = key.zfill(n_bits) + counts[bits] = counts.get(bits, 0) + val + + metadata = ( + native_circuit.metadata + if native_circuit.metadata + else {} + ) + return UnifiedResult(counts, shots, self.name, metadata) + + def statevector(self, native_circuit) -> np.ndarray: + """Return the exact statevector.""" + # Strip measurements for statevector computation. + qc = native_circuit.remove_final_measurements( + inplace=False + ) + sv = Statevector.from_instruction(qc) + return np.asarray(sv) diff --git a/quantumaudio/schemes/base_scheme.py b/quantumaudio/schemes/base_scheme.py index 58477ebb..31238e45 100644 --- a/quantumaudio/schemes/base_scheme.py +++ b/quantumaudio/schemes/base_scheme.py @@ -15,7 +15,8 @@ from abc import ABC, abstractmethod import numpy as np -import qiskit + +from quantumaudio.backends import CircuitSpec class Scheme(ABC): @@ -29,23 +30,23 @@ class Scheme(ABC): """ @abstractmethod - def encode(self, data: np.ndarray) -> qiskit.QuantumCircuit: + def encode(self, data: np.ndarray) -> CircuitSpec: """Encode the input data using the scheme. Args: data: A `numpy` array containing the data to be encoded. Returns: - A `qiskit.QuantumCircuit` representing the encoded data. + A `CircuitSpec` representing the encoded data. """ pass @abstractmethod - def decode(self, circuit: qiskit.QuantumCircuit) -> np.ndarray: + def decode(self, circuit: CircuitSpec) -> np.ndarray: """Decode the quantum circuit using the scheme. Args: - circuit: A `qiskit.QuantumCircuit` that contains the encoded data. + circuit: A `CircuitSpec` that contains the encoded data. Returns: A `numpy` array containing the decoded data. diff --git a/quantumaudio/schemes/mqsm.py b/quantumaudio/schemes/mqsm.py index ac43ef0c..9bb7f42d 100644 --- a/quantumaudio/schemes/mqsm.py +++ b/quantumaudio/schemes/mqsm.py @@ -13,13 +13,13 @@ # limitations under the License. # ========================================================================== -from typing import Optional, Union, Callable, Any, Tuple +from typing import Optional, Union, Tuple import numpy as np -import qiskit from bitstring import BitArray from quantumaudio import utils +from quantumaudio.backends import CircuitSpec, get_backend from .base_scheme import Scheme @@ -55,13 +55,11 @@ def __init__( n_fold: Term for a fixed number of indexed registers used. labels: Name of the Quantum registers positions: Index position of Quantum registers - (In a Qiskit circuit the registers are arranged - from Top to Bottom) convert: Function that applies a mathematical conversion of input at Encoding. restore: Function that restores the conversion at Decoding. - + keys: Reference to essential metadata keys for decoding. Args: @@ -119,15 +117,15 @@ def calculate( # y-axis num_channels = ( 1 if data.ndim == 1 else data.shape[0] - ) # data-dependent channels + ) if self.num_channels: - num_channels = self.num_channels # override with pre-set channels + num_channels = self.num_channels data_shape = (num_channels, num_samples) num_channel_qubits = utils.get_qubit_count( max(2, num_channels) - ) # apply constraint of minimum 2 channels + ) num_value_qubits = ( utils.get_bit_depth(data) if not self.qubit_depth @@ -135,7 +133,6 @@ def calculate( ) qubit_shape = (num_index_qubits, num_channel_qubits, num_value_qubits) - # print if verbose: utils.print_num_qubits(qubit_shape, labels=self.labels) return data_shape, qubit_shape @@ -175,7 +172,7 @@ def initialize_circuit( num_index_qubits: int, num_channel_qubits: int, num_value_qubits: int, - ) -> qiskit.QuantumCircuit: + ) -> CircuitSpec: """Initializes the circuit with Index, Channel and Value Registers. Args: @@ -184,31 +181,37 @@ def initialize_circuit( num_value_qubits: Number of qubits used to encode the sample values. Returns: - Qiskit Circuit with the registers + CircuitSpec with the appropriate number of qubits. """ - index_register = qiskit.QuantumRegister( - num_index_qubits, self.labels[0] - ) - channel_register = qiskit.QuantumRegister( - num_channel_qubits, self.labels[1] - ) - value_register = qiskit.QuantumRegister( - num_value_qubits, self.labels[2] + total = num_value_qubits + num_channel_qubits + num_index_qubits + spec = CircuitSpec( + num_qubits=total, name=self.__class__.__name__ ) - circuit = qiskit.QuantumCircuit( - value_register, - channel_register, - index_register, - name=self.__class__.__name__, - ) - circuit.h(channel_register) - circuit.h(index_register) - return circuit + v_end = num_value_qubits + c_end = v_end + num_channel_qubits + self._value_range = (0, num_value_qubits) + self._channel_range = (v_end, num_channel_qubits) + self._index_range = (c_end, num_index_qubits) + + spec.metadata["registers"] = { + self.labels[2]: self._value_range, + self.labels[1]: self._channel_range, + self.labels[0]: self._index_range, + } + + # Apply Hadamard to channel and index registers. + ch_start, ch_size = self._channel_range + for q in range(ch_start, ch_start + ch_size): + spec.h(q) + idx_start, idx_size = self._index_range + for q in range(idx_start, idx_start + idx_size): + spec.h(q) + return spec @utils.with_indexing def value_setting( - self, circuit: qiskit.QuantumCircuit, index: int, value: float + self, circuit: CircuitSpec, index: int, value: float ) -> None: """Encodes the prepared, converted values to the initialised circuit. @@ -217,31 +220,35 @@ def value_setting( corresponding to the given index. Args: - circuit: Initialized Qiskit Circuit - index: position to set the value - value: value to be set at the index + circuit: Initialized CircuitSpec. + index: position to set the value. + value: value to be set at the index. Note: This method is used in a loop where each value is iterated and set at its corresponding index. """ - value_register, channel_register, index_register = circuit.qregs - for i, areg_qubit in enumerate(value_register): + v_start, v_size = self._value_range + ch_start, ch_size = self._channel_range + idx_start, idx_size = self._index_range + controls = ( + list(range(ch_start, ch_start + ch_size)) + + list(range(idx_start, idx_start + idx_size)) + ) + for i in range(v_size): a_bit = (value >> i) & 1 if a_bit: - circuit.mcx( - channel_register[:] + index_register[:], areg_qubit - ) + circuit.mcx(controls, v_start + i) - def measure(self, circuit: qiskit.QuantumCircuit) -> None: - """Adds classical measurements to all registers of the Quantum Circuit + def measure(self, circuit: CircuitSpec) -> None: + """Adds classical measurements to all qubits of the circuit if the circuit is not already measured. Args: - circuit: Encoded Qiskit Circuit + circuit: Encoded CircuitSpec. """ - if not circuit.cregs: + if circuit.num_clbits == 0: circuit.measure_all() # ----- Default Encode Function ----- @@ -251,8 +258,8 @@ def encode( data: np.ndarray, measure: bool = True, verbose: Union[int, bool] = 1, - ) -> qiskit.QuantumCircuit: - """Given audio data, prepares a Qiskit Circuit representing it. + ) -> CircuitSpec: + """Given audio data, prepares a CircuitSpec representing it. Args: data: Array representing Digital Audio Samples @@ -263,7 +270,7 @@ def encode( - >2: Displays the encoded circuit. Returns: - A Qiskit Circuit representing the Digital Audio + A CircuitSpec representing the Digital Audio. """ utils.validate_data(data) @@ -286,12 +293,12 @@ def encode( self.value_setting(circuit=circuit, index=i, value=sample) # additional information for decoding - circuit.metadata = { + circuit.metadata.update({ "num_samples": num_samples, "num_channels": num_channels, "qubit_shape": qubit_shape, "scheme": circuit.name, - } + }) # measure if measure: @@ -304,7 +311,7 @@ def encode( def decode_components( self, - counts: Union[dict, qiskit.result.Counts], + counts: dict, qubit_shape: Tuple[int, int, int], ) -> np.ndarray: """The first stage of decoding is extracting the required components from @@ -337,20 +344,19 @@ def decode_components( def reconstruct_data( self, - counts: Union[dict, qiskit.result.Counts], + counts: dict, qubit_shape: Tuple[int, int, int], ) -> np.ndarray: - """Given counts, Extract components and restore the conversion at + """Given counts, extract components and restore the conversion at encoding stage. Args: counts: a dictionary with the outcome of measurements performed on the quantum circuit. qubit_shape: Tuple to determine the number of (channels, samples) to get. - qubit_depth : number of qubits in amplitude register. Return: - Array of restored values + Array of restored values. """ data = self.decode_components(counts, qubit_shape) data = self.restore(data, bit_depth=qubit_shape[-1]) @@ -358,45 +364,33 @@ def reconstruct_data( def decode_counts( self, - counts: Union[dict, qiskit.result.Counts], + counts: dict, metadata: dict, - keep_padding: Tuple[int, int] = (False, False), + keep_padding: Tuple[bool, bool] = (False, False), ) -> np.ndarray: - """Given a Qiskit counts object or Dictionary, Extract components and restore the + """Given a counts dictionary, extract components and restore the conversion did at encoding stage. Args: - counts: a qiskit Counts object or Dictionary obtained from a job result. - metadata: metadata required for decoding. + counts: a counts dictionary obtained from a job result. + metadata: metadata required for decoding. keep_padding: Undo the padding set at Encoding stage if set to False. - Dimension 0 for Channels. - Dimension 1 for Time. Return: - Array of restored values with original dimensions + Array of restored values with original dimensions. """ - index_position, channel_position, amplitude_position = self.positions - - # decoding x-axis qubit_shape = metadata["qubit_shape"] - num_index_qubits = qubit_shape[0] num_channel_qubits = qubit_shape[1] - - num_samples = 2**num_index_qubits num_channels = 2**num_channel_qubits - num_components = (num_channels, num_samples) original_num_samples = metadata["num_samples"] original_num_channels = metadata["num_channels"] - # decoding y-axis - qubit_depth = qubit_shape[2] - - # decoding data data = self.reconstruct_data(counts=counts, qubit_shape=qubit_shape) - # reconstruct data = utils.restore_channels(data, num_channels) if not keep_padding[0]: @@ -409,15 +403,15 @@ def decode_counts( def decode_result( self, - result: qiskit.result.Result, + result, metadata: Optional[dict] = None, - keep_padding: Tuple[int, int] = (False, False), + keep_padding: Tuple[bool, bool] = (False, False), ) -> np.ndarray: - """Given a result object. Extract components and restore the conversion - did in the encoding stage. + """Given a result object, extract components and restore the + conversion did in the encoding stage. Args: - result: a qiskit Result object that contains counts along + result: a result object that contains counts along with metadata that was held by the original circuit. metadata: optionally pass metadata as argument. keep_padding: Undo the padding set at Encoding stage if set to False. @@ -426,7 +420,7 @@ def decode_result( - Dimension 1 for Time. Return: - Array of restored values with original dimensions + Array of restored values with original dimensions. """ counts = utils.get_counts(result) metadata = utils.get_metadata(result) if not metadata else metadata @@ -439,30 +433,31 @@ def decode_result( def decode( self, - circuit: qiskit.QuantumCircuit, + circuit: CircuitSpec, metadata: Optional[dict] = None, - keep_padding: Tuple[int, int] = (False, False), - execute_function: Callable[ - [qiskit.QuantumCircuit, dict], Any - ] = utils.execute, + keep_padding: Tuple[bool, bool] = (False, False), + backend=None, **kwargs, ) -> np.ndarray: - """Given a qiskit circuit, decodes and returns the Original Audio Array. + """Given a CircuitSpec, decodes and returns the Original Audio Array. Args: - circuit: A Qiskit Circuit representing the Digital Audio. + circuit: A CircuitSpec representing the Digital Audio. metadata: optionally pass metadata as argument. keep_padding: Undo the padding set at Encoding stage if set False. - execute_function: Function to execute the circuit for decoding. - - - Defaults to :ref:`utils.execute ` which accepts any additional `**kwargs`. + backend: Backend name to execute on (default: "qiskit"). Return: - Array of decoded values + Array of decoded values. """ self.measure(circuit) - result = execute_function(circuit=circuit, **kwargs) - data = self.decode_result( - result=result, metadata=metadata, keep_padding=keep_padding + be = get_backend(backend or "qiskit") + result = be.run_spec(circuit, shots=kwargs.get("shots", 8000)) + if metadata is None: + metadata = circuit.metadata + data = self.decode_counts( + counts=result.counts, + metadata=metadata, + keep_padding=keep_padding, ) return data diff --git a/quantumaudio/schemes/msqpam.py b/quantumaudio/schemes/msqpam.py index 821d7be3..3fd7d863 100644 --- a/quantumaudio/schemes/msqpam.py +++ b/quantumaudio/schemes/msqpam.py @@ -13,12 +13,12 @@ # limitations under the License. # ========================================================================== -from typing import Optional, Union, Callable, Any, Tuple +from typing import Optional, Union, Tuple import numpy as np -import qiskit from quantumaudio import utils +from quantumaudio.backends import CircuitSpec, get_backend from .base_scheme import Scheme @@ -50,13 +50,11 @@ def __init__(self, num_channels: Optional[int] = None) -> None: n_fold: Term for a fixed number of indexed registers used. labels: Name of the Quantum registers positions: Index position of Quantum registers - (In a Qiskit circuit the registers are arranged - from Top to Bottom) convert: Function that applies a mathematical conversion of input at Encoding. restore: Function that restores the conversion at Decoding. - + keys: Reference to essential metadata keys for decoding. Args: @@ -116,19 +114,18 @@ def calculate( # y-axis num_channels = ( 1 if data.ndim == 1 else data.shape[0] - ) # data-dependent channels + ) if self.num_channels: - num_channels = self.num_channels # override with pre-set channels + num_channels = self.num_channels data_shape = (num_channels, num_samples) num_channel_qubits = utils.get_qubit_count( max(2, num_channels) - ) # apply constraint of minimum 2 channels + ) num_value_qubits = self.qubit_depth qubit_shape = (num_index_qubits, num_channel_qubits, num_value_qubits) - # print if verbose: utils.print_num_qubits(qubit_shape, labels=self.labels) return data_shape, qubit_shape @@ -166,7 +163,7 @@ def initialize_circuit( num_index_qubits: int, num_channel_qubits: int, num_value_qubits: int, - ) -> qiskit.QuantumCircuit: + ) -> CircuitSpec: """Initializes the circuit with Index, Channel and Value Registers. Args: @@ -175,31 +172,37 @@ def initialize_circuit( num_value_qubits: Number of qubits used to encode the sample values. Returns: - Qiskit Circuit with the registers + CircuitSpec with the appropriate number of qubits. """ - index_register = qiskit.QuantumRegister( - num_index_qubits, self.labels[0] - ) - channel_register = qiskit.QuantumRegister( - num_channel_qubits, self.labels[1] - ) - value_register = qiskit.QuantumRegister( - num_value_qubits, self.labels[2] + total = num_value_qubits + num_channel_qubits + num_index_qubits + spec = CircuitSpec( + num_qubits=total, name=self.__class__.__name__ ) - circuit = qiskit.QuantumCircuit( - value_register, - channel_register, - index_register, - name=self.__class__.__name__, - ) - circuit.h(channel_register) - circuit.h(index_register) - return circuit + v_end = num_value_qubits + c_end = v_end + num_channel_qubits + self._value_range = (0, num_value_qubits) + self._channel_range = (v_end, num_channel_qubits) + self._index_range = (c_end, num_index_qubits) + + spec.metadata["registers"] = { + self.labels[2]: self._value_range, + self.labels[1]: self._channel_range, + self.labels[0]: self._index_range, + } + + # Apply Hadamard to channel and index registers. + ch_start, ch_size = self._channel_range + for q in range(ch_start, ch_start + ch_size): + spec.h(q) + idx_start, idx_size = self._index_range + for q in range(idx_start, idx_start + idx_size): + spec.h(q) + return spec @utils.with_indexing def value_setting( - self, circuit: qiskit.QuantumCircuit, index: int, value: float + self, circuit: CircuitSpec, index: int, value: float ) -> None: """Encodes the prepared, converted values to the initialised circuit. This function is used to set a single value at a single index. The @@ -207,39 +210,27 @@ def value_setting( corresponding to the given index. Args: - circuit: Initialized Qiskit Circuit - index: position to set the value - value: value to be set at the index + circuit: Initialized CircuitSpec. + index: position to set the value. + value: value to be set at the index. """ - value_register, channel_register, index_register = circuit.qregs - - # initialise sub-circuit - sub_circuit = qiskit.QuantumCircuit( - name=f"Sample {index} (CH {index%(2**channel_register.size)})" - ) - sub_circuit.add_register(value_register) - - # rotate qubits with values - sub_circuit.ry(2 * value, 0) - - # entangle with index qubits - sub_circuit = sub_circuit.control( - channel_register.size + index_register.size + v_start, _ = self._value_range + ch_start, ch_size = self._channel_range + idx_start, idx_size = self._index_range + controls = ( + list(range(ch_start, ch_start + ch_size)) + + list(range(idx_start, idx_start + idx_size)) ) + circuit.mcry(2 * value, controls, v_start) - # attach sub-circuit - circuit.append( - sub_circuit, list(i for i in range(circuit.num_qubits - 1, -1, -1)) - ) - - def measure(self, circuit: qiskit.QuantumCircuit) -> None: - """Adds classical measurements to all registers of the Quantum Circuit + def measure(self, circuit: CircuitSpec) -> None: + """Adds classical measurements to all qubits of the circuit if the circuit is not already measured. Args: - circuit: Encoded Qiskit Circuit + circuit: Encoded CircuitSpec. """ - if not circuit.cregs: + if circuit.num_clbits == 0: circuit.measure_all() # ----- Default Encode Function ----- @@ -249,8 +240,8 @@ def encode( data: np.ndarray, measure: bool = True, verbose: Union[int, bool] = 1, - ) -> qiskit.QuantumCircuit: - """Given audio data, prepares a Qiskit Circuit representing it. + ) -> CircuitSpec: + """Given audio data, prepares a CircuitSpec representing it. Args: data: Array representing Digital Audio Samples @@ -261,7 +252,7 @@ def encode( - >2: Displays the encoded circuit. Returns: - A Qiskit Circuit representing the Digital Audio + A CircuitSpec representing the Digital Audio. """ utils.validate_data(data) @@ -284,26 +275,26 @@ def encode( self.value_setting(circuit=circuit, index=i, value=sample) # additional information for decoding - circuit.metadata = { + circuit.metadata.update({ "num_samples": num_samples, "num_channels": num_channels, "qubit_shape": qubit_shape, "scheme": circuit.name, - } + }) # measure if measure: self.measure(circuit) if verbose == 2: - utils.draw_circuit(circuit, decompose=1) + utils.draw_circuit(circuit) return circuit # ------------------- Decoding Helpers --------------------------- def decode_components( self, - counts: Union[dict, qiskit.result.Counts], - qubit_shape: Tuple[int, int], + counts: dict, + qubit_shape: Tuple[int, int, int], ) -> np.ndarray: """The first stage of decoding is extracting required components from counts. @@ -317,7 +308,6 @@ def decode_components( 2-D Array of shape (num_channels, num_samples) for further decoding. """ - # initialising components num_index_qubits = qubit_shape[0] num_channel_qubits = qubit_shape[1] @@ -328,7 +318,6 @@ def decode_components( cosine_amps = np.zeros(num_components) sine_amps = np.zeros(num_components) - # getting components from counts for state in counts: index_bits, channel_bits, value_bits = utils.split_string( state, qubit_shape @@ -345,11 +334,11 @@ def decode_components( def reconstruct_data( self, - counts: Union[dict, qiskit.result.Counts], - qubit_shape: Tuple[int, int], + counts: dict, + qubit_shape: Tuple[int, int, int], inverted: bool = False, ) -> np.ndarray: - """Given counts, Extract components and restore the conversion did at + """Given counts, extract components and restore the conversion did at encoding stage. Args: @@ -359,7 +348,7 @@ def reconstruct_data( inverted : retrieves cosine components of the signal. Return: - Array of restored values + Array of restored values. """ cosine_amps, sine_amps = self.decode_components(counts, qubit_shape) data = self.restore(cosine_amps, sine_amps, inverted) @@ -367,16 +356,16 @@ def reconstruct_data( def decode_counts( self, - counts: Union[dict, qiskit.result.Counts], + counts: dict, metadata: dict, inverted: bool = False, - keep_padding: Tuple[int, int] = (False, False), + keep_padding: Tuple[bool, bool] = (False, False), ) -> np.ndarray: - """Given a Qiskit counts object or Dictionary, Extract components and restore the + """Given a counts dictionary, extract components and restore the conversion did at encoding stage. Args: - counts: a qiskit Counts object or Dictionary obtained from a job result. + counts: a counts dictionary obtained from a job result. metadata: metadata required for decoding. inverted : retrieves cosine components of the signal. keep_padding: Undo the padding set at Encoding stage if set to False. @@ -385,10 +374,8 @@ def decode_counts( - Dimension 1 for Time. Return: - Array of restored values with original dimensions + Array of restored values with original dimensions. """ - # decoding x-axis - index_position, channel_position, _ = self.positions qubit_shape = metadata["qubit_shape"] num_channel_qubits = qubit_shape[1] @@ -397,7 +384,6 @@ def decode_counts( original_num_samples = metadata["num_samples"] original_num_channels = metadata["num_channels"] - # decoding y-axis data = self.reconstruct_data( counts=counts, qubit_shape=qubit_shape, @@ -417,16 +403,16 @@ def decode_counts( def decode_result( self, - result: qiskit.result.Result, + result, metadata: Optional[dict] = None, inverted: bool = False, - keep_padding: Tuple[int, int] = (False, False), + keep_padding: Tuple[bool, bool] = (False, False), ) -> np.ndarray: - """Given a result object. Extract components and restore the conversion - did in the encoding stage. + """Given a result object, extract components and restore the + conversion did in the encoding stage. Args: - result: a qiskit Result object that contains counts along + result: a result object that contains counts along with metadata that was held by the original circuit. metadata: optionally pass metadata as argument. inverted : retrieves cosine components of the signal. @@ -436,7 +422,7 @@ def decode_result( - Dimension 1 for Time. Return: - Array of restored values with original dimensions + Array of restored values with original dimensions. """ counts = utils.get_counts(result) metadata = utils.get_metadata(result) if not metadata else metadata @@ -452,36 +438,35 @@ def decode_result( def decode( self, - circuit: qiskit.QuantumCircuit, + circuit: CircuitSpec, metadata: Optional[dict] = None, inverted: bool = False, - keep_padding: Tuple[int, int] = (False, False), - execute_function: Callable[ - [qiskit.QuantumCircuit, dict], Any - ] = utils.execute, + keep_padding: Tuple[bool, bool] = (False, False), + backend=None, **kwargs, ) -> np.ndarray: - """Given a qiskit circuit, decodes and returns the Original Audio Array. + """Given a CircuitSpec, decodes and returns the Original Audio Array. Args: - circuit: A Qiskit Circuit representing the Digital Audio. + circuit: A CircuitSpec representing the Digital Audio. metadata: optionally pass metadata as argument. inverted: retrieves cosine components of the signal. keep_padding: Undo the padding set at Encoding stage if set to False. - Dimension 0 for Channels. - Dimension 1 for Time. - execute_function: Function to execute the circuit for decoding. - - - Defaults to :ref:`utils.execute ` which accepts any additional `**kwargs`. + backend: Backend name to execute on (default: "qiskit"). Return: - Array of decoded values + Array of decoded values. """ self.measure(circuit) - result = utils.execute(circuit=circuit, **kwargs) - data = self.decode_result( - result=result, + be = get_backend(backend or "qiskit") + result = be.run_spec(circuit, shots=kwargs.get("shots", 8000)) + if metadata is None: + metadata = circuit.metadata + data = self.decode_counts( + counts=result.counts, metadata=metadata, inverted=inverted, keep_padding=keep_padding, diff --git a/quantumaudio/schemes/qpam.py b/quantumaudio/schemes/qpam.py index 98122eaf..93c962e4 100644 --- a/quantumaudio/schemes/qpam.py +++ b/quantumaudio/schemes/qpam.py @@ -13,12 +13,12 @@ # limitations under the License. # ========================================================================== -from typing import Optional, Union, Callable, Any, Tuple +from typing import Optional, Union, Tuple import numpy as np -import qiskit from quantumaudio import utils +from quantumaudio.backends import CircuitSpec, get_backend from .base_scheme import Scheme @@ -27,8 +27,8 @@ class QPAM(Scheme): QPAM class implements encoding and decoding of Digital Audio as Quantum Probability Amplitudes. It's the simplest of Schemes and - uses Qiskit circuit's `initialize` method to set the Quantum States - based on provided values. The values are normalized before encoding + uses state initialisation to set the Quantum States based on + provided values. The values are normalized before encoding using the `convert` method. """ @@ -135,7 +135,7 @@ def prepare_data( def initialize_circuit( self, num_index_qubits: int, num_value_qubits: int - ) -> qiskit.QuantumCircuit: + ) -> CircuitSpec: """Initializes the circuit with Index and Value Registers. Args: @@ -143,39 +143,44 @@ def initialize_circuit( num_value_qubits: Number of qubits used to encode the sample values. Returns: - Qiskit Circuit with the registers + CircuitSpec with the appropriate number of qubits. """ - index_register = qiskit.QuantumRegister( - num_index_qubits, self.labels[0] + total = num_value_qubits + num_index_qubits + spec = CircuitSpec( + num_qubits=total, name=self.__class__.__name__ ) - value_register = qiskit.QuantumRegister( - num_value_qubits, self.labels[1] - ) - # Arranging Registers from Top to Bottom - circuit = qiskit.QuantumCircuit( - value_register, index_register, name=self.__class__.__name__ - ) - return circuit + self._value_range = (0, num_value_qubits) + self._index_range = (num_value_qubits, num_index_qubits) + if num_value_qubits > 0: + spec.metadata["registers"] = { + self.labels[1]: self._value_range, + self.labels[0]: self._index_range, + } + else: + spec.metadata["registers"] = { + self.labels[0]: self._index_range, + } + return spec def value_setting( - self, circuit: qiskit.QuantumCircuit, values: np.ndarray + self, circuit: CircuitSpec, values: np.ndarray ) -> None: """Encodes the prepared, converted values to the initialised circuit. Args: - circuit: Initialized Qiskit Circuit - values: Array of probability amplitudes to encode + circuit: Initialized CircuitSpec. + values: Array of probability amplitudes to encode. """ - circuit.initialize(values) + circuit.initialize(list(values)) - def measure(self, circuit: qiskit.QuantumCircuit) -> None: - """Adds classical measurements to all qubits of the Quantum Circuit if + def measure(self, circuit: CircuitSpec) -> None: + """Adds classical measurements to all qubits of the circuit if the circuit is not already measured. Args: - circuit: Encoded Qiskit Circuit + circuit: Encoded CircuitSpec. """ - if not circuit.cregs: + if circuit.num_clbits == 0: circuit.measure_all() # ----- Default Encode Function ----- @@ -185,8 +190,8 @@ def encode( data: np.ndarray, measure: bool = True, verbose: Union[int, bool] = 1, - ) -> qiskit.QuantumCircuit: - """Given audio data, prepares a Qiskit Circuit representing it. + ) -> CircuitSpec: + """Given audio data, prepares a CircuitSpec representing it. Args: data: Array representing Digital Audio Samples @@ -197,7 +202,7 @@ def encode( - >2: Displays the encoded circuit. Returns: - A Qiskit Circuit representing the Digital Audio + A CircuitSpec representing the Digital Audio. """ utils.validate_data(data) @@ -213,11 +218,11 @@ def encode( # encode values self.value_setting(circuit=circuit, values=values) # additional information for decoding - circuit.metadata = { + circuit.metadata.update({ "num_samples": num_samples, "norm_factor": norm, "scheme": circuit.name, - } + }) if measure: self.measure(circuit) if verbose == 2: @@ -227,7 +232,7 @@ def encode( # ------------------- Decoding Helpers --------------------------- def decode_components( - self, counts: Union[dict, qiskit.result.Counts] + self, counts: dict ) -> np.ndarray: """The first stage of decoding is extracting required components from counts. @@ -244,7 +249,7 @@ def decode_components( def reconstruct_data( self, - counts: Union[dict, qiskit.result.Counts], + counts: dict, shots: int, norm: float, ) -> np.ndarray: @@ -258,7 +263,7 @@ def reconstruct_data( norm : the norm factor used to normalize the decoding in QPAM. Return: - Array of restored values + Array of restored values. """ probabilities = self.decode_components(counts) data = self.restore(probabilities, norm, shots) @@ -266,24 +271,24 @@ def reconstruct_data( def decode_counts( self, - counts: Union[dict, qiskit.result.Counts], + counts: dict, metadata: dict, shots: Optional[int] = 4000, norm: Optional[float] = None, keep_padding: bool = False, ) -> np.ndarray: - """Given a Qiskit counts object or Dictionary, Extract components and restore the + """Given a counts dictionary, extract components and restore the conversion did at encoding stage. Args: - counts: a qiskit Counts object or Dictionary obtained from a job result. + counts: a counts dictionary obtained from a job result. metadata: metadata required for decoding. shots : total number of times the quantum circuit is measured. norm : Override the norm factor used to normalize the decoding. keep_padding: Undos the padding set at Encoding stage if set to False. Return: - Array of restored values with original dimensions + Array of restored values with original dimensions. """ shots = metadata.get("shots", shots) norm = norm if norm else metadata["norm_factor"] @@ -304,17 +309,17 @@ def decode_counts( def decode_result( self, - result: qiskit.result.Result, + result, metadata: Optional[dict] = None, shots: Optional[int] = 8000, norm: Optional[float] = None, keep_padding: bool = False, ) -> np.ndarray: - """Given a Qiskit Result object, Extract components and restore the + """Given a result object, extract components and restore the conversion did at encoding stage. Args: - result: a qiskit Result object that contains counts along + result: a result object that contains counts along with metadata that was held by the original circuit. metadata: optionally pass metadata as argument. shots : total number of times the quantum circuit is measured. @@ -322,7 +327,7 @@ def decode_result( keep_padding: Undos the padding set at Encoding stage if set to False. Return: - Array of restored values with original dimensions + Array of restored values with original dimensions. """ counts = utils.get_counts(result) metadata = utils.get_metadata(result) if not metadata else metadata @@ -340,38 +345,35 @@ def decode_result( def decode( self, - circuit: qiskit.QuantumCircuit, + circuit: CircuitSpec, metadata: Optional[dict] = None, shots: Optional[int] = 8000, norm: Optional[float] = None, keep_padding: bool = False, - execute_function: Callable[ - [qiskit.QuantumCircuit, dict], Any - ] = utils.execute, + backend=None, **kwargs, ) -> np.ndarray: - """Given a qiskit circuit, decodes and returns back the Original Audio Array. + """Given a CircuitSpec, decodes and returns back the Original Audio Array. Args: - circuit: A Qiskit Circuit representing the Digital Audio. + circuit: A CircuitSpec representing the Digital Audio. metadata: optionally pass metadata as argument. shots : Total number of times the quantum circuit is measured. norm : The norm factor used to normalize the decoding in QPAM. keep_padding: Undo the padding set at Encoding stage if set to False. - execute_function: Function to execute the circuit for decoding. - - - Defaults to :ref:`utils.execute ` which accepts any additional `**kwargs`. - - The keyword argument **shots** (int) is a metadata for QPAM decoding and accepted - by `execute_function`. (Defaults to **8000**) + backend: Backend name to execute on (default: "qiskit"). Return: - Array of decoded values + Array of decoded values. """ self.measure(circuit) - kwargs["shots"] = shots - result = execute_function(circuit=circuit, **kwargs) - data = self.decode_result( - result=result, + be = get_backend(backend or "qiskit") + result = be.run_spec(circuit, shots=shots) + if metadata is None: + metadata = circuit.metadata + metadata["shots"] = shots + data = self.decode_counts( + counts=result.counts, metadata=metadata, shots=shots, norm=norm, diff --git a/quantumaudio/schemes/qsm.py b/quantumaudio/schemes/qsm.py index ca0b037a..7a6a5ff9 100644 --- a/quantumaudio/schemes/qsm.py +++ b/quantumaudio/schemes/qsm.py @@ -13,13 +13,13 @@ # limitations under the License. # ========================================================================== -from typing import Optional, Union, Callable, Any, Tuple +from typing import Optional, Union, Tuple import numpy as np -import qiskit from bitstring import BitArray from quantumaudio import utils +from quantumaudio.backends import CircuitSpec, get_backend from .base_scheme import Scheme @@ -46,13 +46,11 @@ def __init__(self, qubit_depth: Optional[int] = None) -> None: n_fold: Term for a fixed number of indexed registers used. labels: Name of the Quantum registers positions: Index position of Quantum registers - (In Qiskit circuit the registers are arranged - from Top to Bottom) convert: Function that applies a mathematical conversion of input at Encoding. restore: Function that restores the conversion at Decoding. - + keys: Reference to essential metadata keys for decoding. Args: @@ -144,7 +142,7 @@ def prepare_data( def initialize_circuit( self, num_index_qubits: int, num_value_qubits: int - ) -> qiskit.QuantumCircuit: + ) -> CircuitSpec: """Initializes the circuit with Index and Value Registers. Args: @@ -152,24 +150,27 @@ def initialize_circuit( num_value_qubits: Number of qubits used to encode the sample values. Returns: - Qiskit Circuit with the registers + CircuitSpec with the appropriate number of qubits. """ - index_register = qiskit.QuantumRegister( - num_index_qubits, self.labels[0] - ) - value_register = qiskit.QuantumRegister( - num_value_qubits, self.labels[1] - ) - # Arranging Registers from Top to Bottom - circuit = qiskit.QuantumCircuit( - value_register, index_register, name=self.__class__.__name__ + total = num_value_qubits + num_index_qubits + spec = CircuitSpec( + num_qubits=total, name=self.__class__.__name__ ) - circuit.h(index_register) - return circuit + self._value_range = (0, num_value_qubits) + self._index_range = (num_value_qubits, num_index_qubits) + spec.metadata["registers"] = { + self.labels[1]: self._value_range, + self.labels[0]: self._index_range, + } + # Apply Hadamard to index register. + idx_start, idx_size = self._index_range + for q in range(idx_start, idx_start + idx_size): + spec.h(q) + return spec @utils.with_indexing def value_setting( - self, circuit: qiskit.QuantumCircuit, index: int, value: float + self, circuit: CircuitSpec, index: int, value: float ) -> None: """Encodes the prepared, converted values to the initialised circuit. @@ -178,24 +179,26 @@ def value_setting( corresponding to the given index. Args: - circuit: Initialized Qiskit Circuit - index: position to set the value - value: value to be set at the index + circuit: Initialized CircuitSpec. + index: position to set the value. + value: value to be set at the index. """ - value_register, index_register = circuit.qregs - for i, areg_qubit in enumerate(value_register): + v_start, v_size = self._value_range + idx_start, idx_size = self._index_range + controls = list(range(idx_start, idx_start + idx_size)) + for i in range(v_size): a_bit = (value >> i) & 1 if a_bit: - circuit.mcx(index_register, areg_qubit) + circuit.mcx(controls, v_start + i) - def measure(self, circuit: qiskit.QuantumCircuit) -> None: - """Adds classical measurements to all registers of the Quantum Circuit + def measure(self, circuit: CircuitSpec) -> None: + """Adds classical measurements to all qubits of the circuit if the circuit is not already measured. Args: - circuit: Encoded Qiskit Circuit + circuit: Encoded CircuitSpec. """ - if not circuit.cregs: + if circuit.num_clbits == 0: circuit.barrier() circuit.measure_all() @@ -206,8 +209,8 @@ def encode( data: np.ndarray, measure: bool = True, verbose: Union[int, bool] = 1, - ) -> qiskit.QuantumCircuit: - """Given an audio data, prepares a Qiskit Circuit representing it. + ) -> CircuitSpec: + """Given an audio data, prepares a CircuitSpec representing it. Args: data: Array representing Digital Audio Samples @@ -218,7 +221,7 @@ def encode( - >2: Displays the encoded circuit. Returns: - A Qiskit Circuit representing the Digital Audio + A CircuitSpec representing the Digital Audio. """ utils.validate_data(data) @@ -236,11 +239,11 @@ def encode( self.value_setting(circuit=circuit, index=i, value=sample) # additional information for decoding - circuit.metadata = { + circuit.metadata.update({ "num_samples": num_samples, "qubit_shape": (num_index_qubits, num_value_qubits), "scheme": circuit.name, - } + }) # measure, print and return if measure: @@ -253,8 +256,8 @@ def encode( def decode_components( self, - counts: Union[dict, qiskit.result.Counts], - qubit_shape: [int, int], + counts: dict, + qubit_shape: Tuple[int, int], ) -> np.ndarray: """The first stage of decoding is extracting required components from counts. @@ -280,7 +283,7 @@ def decode_components( return data def reconstruct_data( - self, counts: Union[dict, qiskit.result.Counts], qubit_shape: int + self, counts: dict, qubit_shape: Tuple[int, int] ) -> np.ndarray: """Given counts, Extract components and restore the conversion did at encoding stage. @@ -289,10 +292,9 @@ def reconstruct_data( counts: a dictionary with the outcome of measurements performed on the quantum circuit. qubit_shape: Tuple to determine the number of components to get. - qubit_depth : number of qubits in amplitude register. Return: - Array of restored values + Array of restored values. """ data = self.decode_components(counts, qubit_shape) data = self.restore(data, bit_depth=qubit_shape[-1]) @@ -300,28 +302,24 @@ def reconstruct_data( def decode_counts( self, - counts: Union[dict, qiskit.result.Counts], + counts: dict, metadata: dict, keep_padding: bool = False, ) -> np.ndarray: - """Given a result object. Extract components and restore the conversion - did in encoding stage. + """Given a counts dictionary, extract components and restore the + conversion did in encoding stage. Args: - counts: a qiskit Counts object or Dictionary obtained from a job result. + counts: a counts dictionary obtained from a job result. metadata: metadata required for decoding. keep_padding: Undo the padding set at Encoding stage if set False. Return: - Array of restored values with original dimensions + Array of restored values with original dimensions. """ - index_position, amplitude_position = self.positions qubit_shape = metadata["qubit_shape"] - - # decoding x-axis original_num_samples = metadata["num_samples"] - # decoding y-axis data = self.reconstruct_data(counts, qubit_shape) # undo padding @@ -331,21 +329,21 @@ def decode_counts( def decode_result( self, - result: qiskit.result.Result, + result, metadata: Optional[dict] = None, keep_padding: bool = False, ) -> np.ndarray: - """Given a result object. Extract components and restore the conversion - did in encoding stage. + """Given a result object, extract components and restore the + conversion did in encoding stage. Args: - result: a qiskit Result object that contains counts along + result: a result object that contains counts along with metadata that was held by the original circuit. metadata: optionally pass metadata as argument. keep_padding: Undo the padding set at Encoding stage if set False. Return: - Array of restored values with original dimensions + Array of restored values with original dimensions. """ counts = utils.get_counts(result) metadata = utils.get_metadata(result) if not metadata else metadata @@ -358,30 +356,31 @@ def decode_result( def decode( self, - circuit: qiskit.QuantumCircuit, + circuit: CircuitSpec, metadata: Optional[dict] = None, keep_padding: bool = False, - execute_function: Callable[ - [qiskit.QuantumCircuit, dict], Any - ] = utils.execute, + backend=None, **kwargs, ) -> np.ndarray: - """Given a qiskit circuit, decodes and returns back the Original Audio Array. + """Given a CircuitSpec, decodes and returns back the Original Audio Array. Args: - circuit: A Qiskit Circuit representing the Digital Audio. + circuit: A CircuitSpec representing the Digital Audio. metadata: optionally pass metadata as argument. keep_padding: Undo the padding set at Encoding stage if set False. - execute_function: Function to execute the circuit for decoding. - - - Defaults to :ref:`utils.execute ` which accepts any additional `**kwargs`. + backend: Backend name to execute on (default: "qiskit"). Return: - Array of decoded values + Array of decoded values. """ self.measure(circuit) - result = execute_function(circuit=circuit, **kwargs) - data = self.decode_result( - result=result, metadata=metadata, keep_padding=keep_padding + be = get_backend(backend or "qiskit") + result = be.run_spec(circuit, shots=kwargs.get("shots", 8000)) + if metadata is None: + metadata = circuit.metadata + data = self.decode_counts( + counts=result.counts, + metadata=metadata, + keep_padding=keep_padding, ) return data diff --git a/quantumaudio/schemes/sqpam.py b/quantumaudio/schemes/sqpam.py index e127f989..7b18a708 100644 --- a/quantumaudio/schemes/sqpam.py +++ b/quantumaudio/schemes/sqpam.py @@ -13,12 +13,12 @@ # limitations under the License. # ========================================================================== -from typing import Optional, Union, Callable, Any, Tuple +from typing import Optional, Union, Tuple import numpy as np -import qiskit from quantumaudio import utils +from quantumaudio.backends import CircuitSpec, get_backend from .base_scheme import Scheme @@ -45,12 +45,10 @@ def __init__(self) -> None: n_fold: Term for a fixed number of indexed registers used. labels: Name of the Quantum registers positions: Index position of Quantum registers - (In Qiskit circuit the registers are arranged - from Top to Bottom) convert: Function that applies a mathematical conversion of input at Encoding. restore: Function that restores the conversion at Decoding. - + keys: Reference to essential metadata keys for decoding. """ self.name = "Single-Qubit Probability Amplitude Modulation" @@ -132,7 +130,7 @@ def prepare_data( def initialize_circuit( self, num_index_qubits: int, num_value_qubits: int - ) -> qiskit.QuantumCircuit: + ) -> CircuitSpec: """Initializes the circuit with Index and Value Registers. Args: @@ -140,24 +138,27 @@ def initialize_circuit( num_value_qubits: Number of qubits used to encode the sample values. Returns: - Qiskit Circuit with the registers + CircuitSpec with the appropriate number of qubits. """ - index_register = qiskit.QuantumRegister( - num_index_qubits, self.labels[0] - ) - value_register = qiskit.QuantumRegister( - num_value_qubits, self.labels[1] + total = num_value_qubits + num_index_qubits + spec = CircuitSpec( + num_qubits=total, name=self.__class__.__name__ ) - # Arranging Registers from Top to Bottom - circuit = qiskit.QuantumCircuit( - value_register, index_register, name=self.__class__.__name__ - ) - circuit.h(index_register) - return circuit + self._value_range = (0, num_value_qubits) + self._index_range = (num_value_qubits, num_index_qubits) + spec.metadata["registers"] = { + self.labels[1]: self._value_range, + self.labels[0]: self._index_range, + } + # Apply Hadamard to index register. + idx_start, idx_size = self._index_range + for q in range(idx_start, idx_start + idx_size): + spec.h(q) + return spec @utils.with_indexing def value_setting( - self, circuit: qiskit.QuantumCircuit, index: int, value: float + self, circuit: CircuitSpec, index: int, value: float ) -> None: """Encodes the prepared, converted values to the initialised circuit. This function is used to set a single value at a single index. The @@ -165,36 +166,24 @@ def value_setting( corresponding to the given index. Args: - circuit: Initialized Qiskit Circuit - index: position to set the value - value: value to be set at the index + circuit: Initialized CircuitSpec. + index: position to set the value. + value: value to be set at the index. """ - value_register, index_register = circuit.qregs - - # initialise sub-circuit - sub_circuit = qiskit.QuantumCircuit(name=f"Sample {index}") - sub_circuit.add_register(value_register) - - # rotate qubits with values - for i in range(value_register.size): - sub_circuit.ry(2 * value, i) - - # entangle with index qubits - sub_circuit = sub_circuit.control(index_register.size) - - # attach sub-circuit - circuit.append( - sub_circuit, list(i for i in range(circuit.num_qubits - 1, -1, -1)) - ) - - def measure(self, circuit: qiskit.QuantumCircuit) -> None: - """Adds classical measurements to all registers of the Quantum Circuit + v_start, v_size = self._value_range + idx_start, idx_size = self._index_range + controls = list(range(idx_start, idx_start + idx_size)) + for i in range(v_size): + circuit.mcry(2 * value, controls, v_start + i) + + def measure(self, circuit: CircuitSpec) -> None: + """Adds classical measurements to all qubits of the circuit if the circuit is not already measured. Args: - circuit: Encoded Qiskit Circuit + circuit: Encoded CircuitSpec. """ - if not circuit.cregs: + if circuit.num_clbits == 0: circuit.measure_all() # ----- Default Encode Function ----- @@ -204,8 +193,8 @@ def encode( data: np.ndarray, measure: bool = True, verbose: Union[int, bool] = 1, - ) -> qiskit.QuantumCircuit: - """Given an audio data, prepares a Qiskit Circuit representing it. + ) -> CircuitSpec: + """Given an audio data, prepares a CircuitSpec representing it. Args: data: Array representing Digital Audio Samples @@ -216,7 +205,7 @@ def encode( - >2: Displays the encoded circuit. Returns: - A Qiskit Circuit representing the Digital Audio + A CircuitSpec representing the Digital Audio. """ utils.validate_data(data) @@ -233,24 +222,24 @@ def encode( for i, value in enumerate(values): self.value_setting(circuit=circuit, index=i, value=value) # additional information for decoding - circuit.metadata = { + circuit.metadata.update({ "num_samples": num_samples, "qubit_shape": (num_index_qubits, num_value_qubits), "scheme": circuit.name, - } + }) # measure, print and return if measure: self.measure(circuit) if verbose == 2: - utils.draw_circuit(circuit, decompose=1) + utils.draw_circuit(circuit) return circuit # ------------------- Decoding Helpers --------------------------- def decode_components( self, - counts: Union[dict, qiskit.result.Counts], - qubit_shape: [int, int], + counts: dict, + qubit_shape: Tuple[int, int], ) -> np.ndarray: """The first stage of decoding is extracting required components from counts. @@ -283,8 +272,8 @@ def decode_components( def reconstruct_data( self, - counts: Union[dict, qiskit.result.Counts], - qubit_shape: [int, int], + counts: dict, + qubit_shape: Tuple[int, int], inverted: bool = False, ) -> np.ndarray: """Given counts, Extract components and restore the conversion did at @@ -297,7 +286,7 @@ def reconstruct_data( inverted : retrieves cosine components of the signal. Return: - Array of restored values + Array of restored values. """ cosine_amps, sine_amps = self.decode_components(counts, qubit_shape) data = self.restore(cosine_amps, sine_amps, inverted) @@ -305,32 +294,28 @@ def reconstruct_data( def decode_counts( self, - counts: Union[dict, qiskit.result.Counts], + counts: dict, metadata: dict, inverted: bool = False, keep_padding: bool = False, ) -> np.ndarray: - """Given a Qiskit counts object or Dictionary, Extract components and restore the + """Given a counts dictionary, extract components and restore the conversion did at encoding stage. Args: - counts: a qiskit Counts object or Dictionary obtained from a job result. + counts: a counts dictionary obtained from a job result. metadata: metadata required for decoding. inverted: retrieves cosine components of the signal. keep_padding: Undo the padding set at Encoding stage if set False. Return: - Array of restored values with original dimensions + Array of restored values with original dimensions. """ - # decoding x-axis - index_position, _ = self.positions qubit_shape = metadata["qubit_shape"] - original_num_samples = metadata["num_samples"] - # decoding y-axis data = self.reconstruct_data( - counts=counts, qubit_shape=qubit_shape, inverted=False + counts=counts, qubit_shape=qubit_shape, inverted=inverted ) # undo padding @@ -341,23 +326,23 @@ def decode_counts( def decode_result( self, - result: qiskit.result.Result, + result, metadata: Optional[dict] = None, inverted: bool = False, keep_padding: bool = False, ) -> np.ndarray: - """Given a result object. Extract components and restore the conversion - did in encoding stage. + """Given a result object, extract components and restore the + conversion did in encoding stage. Args: - result: a qiskit Result object that contains counts along + result: a result object that contains counts along with metadata that was held by the original circuit. metadata: optionally pass metadata as argument. inverted: retrieves cosine components of the signal. keep_padding: Undo the padding set at Encoding stage if set False. Return: - Array of restored values with original dimensions + Array of restored values with original dimensions. """ counts = utils.get_counts(result) metadata = utils.get_metadata(result) if not metadata else metadata @@ -374,33 +359,32 @@ def decode_result( def decode( self, - circuit: qiskit.QuantumCircuit, + circuit: CircuitSpec, metadata: Optional[dict] = None, inverted: bool = False, keep_padding: bool = False, - execute_function: Callable[ - [qiskit.QuantumCircuit, dict], Any - ] = utils.execute, + backend=None, **kwargs, ) -> np.ndarray: - """Given a qiskit circuit, decodes and returns back the Original Audio Array. + """Given a CircuitSpec, decodes and returns back the Original Audio Array. Args: - circuit: A Qiskit Circuit representing the Digital Audio. + circuit: A CircuitSpec representing the Digital Audio. metadata: optionally pass metadata as argument. inverted: retrieves cosine components of the signal. keep_padding: Undo the padding set at Encoding stage if set False. - execute_function: Function to execute the circuit for decoding. - - - Defaults to :ref:`utils.execute ` which accepts any additional `**kwargs`. + backend: Backend name to execute on (default: "qiskit"). Return: - Array of decoded values + Array of decoded values. """ self.measure(circuit) - result = execute_function(circuit=circuit, **kwargs) - data = self.decode_result( - result=result, + be = get_backend(backend or "qiskit") + result = be.run_spec(circuit, shots=kwargs.get("shots", 8000)) + if metadata is None: + metadata = circuit.metadata + data = self.decode_counts( + counts=result.counts, metadata=metadata, inverted=inverted, keep_padding=keep_padding, diff --git a/quantumaudio/utils/circuit.py b/quantumaudio/utils/circuit.py index 4411457c..83230518 100644 --- a/quantumaudio/utils/circuit.py +++ b/quantumaudio/utils/circuit.py @@ -16,29 +16,38 @@ from functools import wraps from typing import Callable -import qiskit +from quantumaudio.backends.core.circuit import CircuitSpec # ========================= # Circuit Preparation Utils # ========================= -def apply_x_at_index(qc: qiskit.QuantumCircuit, i: int) -> None: - """This function is used to encode an index value into control qubits of a circuit. +def apply_x_at_index( + spec: CircuitSpec, scheme, i: int +) -> None: + """Encode an index value into control qubits of a circuit. + + Applies X gates to each control qubit whose corresponding + bit in *i* is 0, preparing the control state for a + multi-controlled operation at index *i*. Args: - qc: Qiskit Circuit - i: Index position + spec: The circuit specification. + scheme: The scheme instance (provides register ranges). + i: Index position. """ - if len(qc.qregs) != 2: - _, creg, treg = qc.qregs - else: - _, treg = qc.qregs - creg = [] - for reg_index, reg_qubit in enumerate(creg[:] + treg[:]): + control_qubits = [] + if hasattr(scheme, "_channel_range"): + ch_start, ch_size = scheme._channel_range + control_qubits.extend(range(ch_start, ch_start + ch_size)) + idx_start, idx_size = scheme._index_range + control_qubits.extend(range(idx_start, idx_start + idx_size)) + + for reg_index, qubit in enumerate(control_qubits): bit = (i >> reg_index) & 1 if not bit: - qc.x(reg_qubit) + spec.x(qubit) def with_indexing(func: Callable) -> Callable: @@ -51,13 +60,13 @@ def with_indexing(func: Callable) -> Callable: The wrapped function with time indexing applied. """ - @wraps(func) # added to fix docstrings not printing func - def wrapper(*args, **kwargs): - qc = kwargs.get("circuit") + @wraps(func) + def wrapper(self, *args, **kwargs): + spec = kwargs.get("circuit") i = kwargs.get("index") - qc.barrier() - apply_x_at_index(qc, i) - func(*args, **kwargs) - apply_x_at_index(qc, i) + spec.barrier() + apply_x_at_index(spec, self, i) + func(self, *args, **kwargs) + apply_x_at_index(spec, self, i) return wrapper diff --git a/quantumaudio/utils/convert.py b/quantumaudio/utils/convert.py index 838f5ef0..970ccd6c 100644 --- a/quantumaudio/utils/convert.py +++ b/quantumaudio/utils/convert.py @@ -13,6 +13,8 @@ # limitations under the License. # ========================================================================== +import numbers + import numpy as np # ====================== @@ -58,15 +60,39 @@ def convert_to_angles(array: np.ndarray) -> np.ndarray: def quantize(array: np.ndarray, qubit_depth: int) -> np.ndarray: """Quantizes the array to a given qubit depth. + Values outside the representable signed-integer range + ``[-2**(qubit_depth - 1), 2**(qubit_depth - 1) - 1]`` are saturated + rather than wrapped, mirroring the standard audio-clipping + convention. Without this, an input of ``+1.0`` at ``qubit_depth=3`` + would produce integer ``+4``, which does not fit in a 3-bit signed + register and wraps to ``-4`` (decoded ``-1.0``): a sign-flip glitch + on every peak sample. + Args: array: The input array. - qubit_depth: The number of bits to quantize to. + qubit_depth: The number of bits to quantize to. Must be an + integer >= 1. Returns: The quantized array as integers. + + Raises: + ValueError: If ``qubit_depth`` is not an integer >= 1. """ - values = array * (2 ** (qubit_depth - 1)) - return values.astype(int) + if ( + not isinstance(qubit_depth, numbers.Integral) + or isinstance(qubit_depth, bool) + or qubit_depth < 1 + ): + raise ValueError( + f"qubit_depth must be an integer >= 1, got {qubit_depth!r}." + ) + # Coerce to Python int so the arithmetic below cannot overflow a + # narrow NumPy integer dtype (e.g. ``np.uint8``). + qubit_depth = int(qubit_depth) + scale = 2 ** (qubit_depth - 1) + values = array * scale + return np.clip(values, -scale, scale - 1).astype(np.int64) def convert_from_probability_amplitudes( diff --git a/quantumaudio/utils/execute.py b/quantumaudio/utils/execute.py index e23afed5..3de236dc 100644 --- a/quantumaudio/utils/execute.py +++ b/quantumaudio/utils/execute.py @@ -37,6 +37,7 @@ def execute( backend: Any = None, keep_memory: bool = False, optimization_level: int = 3, + **kwargs: Any, ): """ Executes a quantum circuit on a given backend and return the results. @@ -47,6 +48,9 @@ def execute( shots: Total number of times the quantum circuit is measured. keep_memory: Whether to return the memory (quantum state) of each shot. optimization_level: Optimization level for transpiling the circuit. + **kwargs: Accepted for compatibility with the :class:`ExecuteFunction` + protocol (additional keyword arguments forwarded by scheme + ``decode`` calls); ignored by this implementation. Returns: Result: The result of the execution, containing the counts and other metadata. @@ -74,6 +78,7 @@ def execute_with_sampler( backend: Any = None, shots: int = 8000, optimization_level: int = 3, + **kwargs: Any, ): """ Executes a quantum circuit on a given backend using `Sampler Primitive` and return the results. @@ -83,6 +88,9 @@ def execute_with_sampler( backend: The backend on which to run the circuit. If None, the default backend `qiskit_aer.AerSimulator()` is used. shots: Total number of times the quantum circuit is measured. optimization_level: Optimization level for transpiling the circuit. + **kwargs: Accepted for compatibility with the :class:`ExecuteFunction` + protocol (additional keyword arguments forwarded by scheme + ``decode`` calls); ignored by this implementation. Returns: Result: The result of the execution, containing the counts and other metadata. diff --git a/quantumaudio/utils/preview.py b/quantumaudio/utils/preview.py index 0be8b84d..975f53ac 100644 --- a/quantumaudio/utils/preview.py +++ b/quantumaudio/utils/preview.py @@ -14,7 +14,8 @@ # ========================================================================== import matplotlib.pyplot as plt -import qiskit + +from quantumaudio.backends import CircuitSpec, get_backend # ====================== # Preview Functions @@ -36,20 +37,24 @@ def print_num_qubits( print(f"{qubits} qubits for {labels[i]}") -def draw_circuit(circuit: qiskit.QuantumCircuit, decompose: int = 0) -> None: +def draw_circuit(circuit, decompose: int = 0) -> None: """Draws a quantum circuit diagram. Args: - circuit: The quantum circuit to draw. + circuit: The quantum circuit or CircuitSpec to draw. decompose: Number of times to decompose the circuit. Defaults to 0. """ + if isinstance(circuit, CircuitSpec): + backend = get_backend("qiskit") + circuit = backend.build_circuit(circuit) + for _i in range(decompose): circuit = circuit.decompose() fig = circuit.draw("mpl", style="clifford") - try: # Check if the code is running in Jupyter Notebook + try: # Check if the code is running in Jupyter Notebook. display(fig) except NameError: plt.show() diff --git a/quantumaudio/utils/results.py b/quantumaudio/utils/results.py index 4b6be2dd..a4944e85 100644 --- a/quantumaudio/utils/results.py +++ b/quantumaudio/utils/results.py @@ -14,9 +14,13 @@ # ========================================================================== from typing import Union + import qiskit from qiskit.primitives import PrimitiveResult, SamplerPubResult +from quantumaudio.backends.core.circuit import CircuitSpec +from quantumaudio.backends.core.result import UnifiedResult + # ====================== # Post-processing # ====================== @@ -44,7 +48,8 @@ def get_counts(results_obj, result_id=0): Extract counts from a results object. Args: - results_obj: An instance of `PrimitiveResult` or `Result` object from which to extract counts. + results_obj: An instance of `UnifiedResult`, `PrimitiveResult`, + or `Result` object from which to extract counts. result_id: The index of the result to extract if the results object contains multiple results. Returns: @@ -52,7 +57,10 @@ def get_counts(results_obj, result_id=0): """ counts = {} - if isinstance(results_obj, PrimitiveResult): + if isinstance(results_obj, UnifiedResult): + counts = results_obj.counts + + elif isinstance(results_obj, PrimitiveResult): results_obj = results_obj[result_id] if isinstance(results_obj, SamplerPubResult): @@ -61,7 +69,7 @@ def get_counts(results_obj, result_id=0): elif isinstance(results_obj, qiskit.result.Result): counts = results_obj.get_counts() - else: + elif not counts: raise TypeError("Unsupported result object type.") return counts @@ -72,7 +80,8 @@ def get_metadata(results_obj, result_id=0): Extract metadata from a results object. Args: - results_obj: An instance of `PrimitiveResult` or `Result` object from which to extract metadata. + results_obj: An instance of `UnifiedResult`, `PrimitiveResult`, + or `Result` object from which to extract metadata. result_id: The index of the result to extract if the results object contains multiple results. Returns: @@ -80,6 +89,9 @@ def get_metadata(results_obj, result_id=0): """ metadata = {} + if isinstance(results_obj, UnifiedResult): + return dict(results_obj.metadata) + if isinstance(results_obj, PrimitiveResult): metadata.update(results_obj.metadata) results_obj = results_obj[result_id] @@ -134,22 +146,26 @@ def pick_key_from_instance(instance, key): """Search for given key in an instance used at decoding. Args: - instance: Can be Qiskit Circuit or Result object. + instance: Can be CircuitSpec, Qiskit Circuit, or Result object. key: Key to find in the encoded metadata. """ - if isinstance(instance, qiskit.circuit.QuantumCircuit): - if key == "scheme" and instance.name.upper() in [ - "QPAM", - "SQPAM", - "QSM", - "MSQPAM", - "MQSM", - ]: + _SCHEME_NAMES = ["QPAM", "SQPAM", "QSM", "MSQPAM", "MQSM"] + + # CircuitSpec path. + if isinstance(instance, CircuitSpec): + if key == "scheme" and instance.name.upper() in _SCHEME_NAMES: return instance.name.upper() elif key in instance.metadata: return instance.metadata[key] + # Legacy Qiskit QuantumCircuit path. + elif isinstance(instance, qiskit.circuit.QuantumCircuit): + if key == "scheme" and instance.name.upper() in _SCHEME_NAMES: + return instance.name.upper() + elif instance.metadata and key in instance.metadata: + return instance.metadata[key] + elif isinstance( instance, (qiskit.result.Result, PrimitiveResult, SamplerPubResult) ): @@ -157,9 +173,9 @@ def pick_key_from_instance(instance, key): if key in metadata: return metadata[key] - # If the key was not found in the instance + # If the key was not found in the instance. if key == "scheme": - raise ValueError(f"{key} is missing") # Scheme is essential + raise ValueError(f"{key} is missing") # Scheme is essential. return None diff --git a/tests/_helpers.py b/tests/_helpers.py new file mode 100644 index 00000000..60dd63bf --- /dev/null +++ b/tests/_helpers.py @@ -0,0 +1,31 @@ +# Copyright 2024 Moth Quantum +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========================================================================== + +"""Shared test helpers.""" + +from qiskit.result.counts import Counts + + +def counts_from_spaced(d: dict) -> Counts: + """Build a Qiskit ``Counts`` from a dict whose keys may include spaces + between register segments. + + The backend layer's ``UnifiedResult.counts`` uses flat bitstrings, but + spaces in fixture literals make the register boundaries visible at a + glance (e.g. ``"101 100"`` for amplitude/time). This helper strips + those spaces so the fixture reads naturally while still producing the + flat-string format the schemes expect. + """ + return Counts({k.replace(" ", ""): v for k, v in d.items()}) diff --git a/tests/test_backends.py b/tests/test_backends.py new file mode 100644 index 00000000..3b6eb76e --- /dev/null +++ b/tests/test_backends.py @@ -0,0 +1,585 @@ +# Copyright 2024 Moth Quantum +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========================================================================== + +"""Tests for the framework-agnostic backend abstractions.""" + +from __future__ import annotations + +import numpy as np +import pytest + +from quantumaudio.backends import ( + CircuitSpec, + GateOp, + GateType, + UnifiedResult, + available_backends, + get_backend, + is_available, + registry, + require, +) + + +# ====================================================================== +# UnifiedResult +# ====================================================================== + + +def test_probabilities_basic(): + res = UnifiedResult( + counts={"00": 30, "01": 10, "11": 60}, + shots=100, + backend_name="test", + ) + probs = res.probabilities() + assert probs == {"00": 0.3, "01": 0.1, "11": 0.6} + + +def test_probabilities_array_indexed_by_int(): + res = UnifiedResult( + counts={"00": 25, "01": 25, "10": 25, "11": 25}, + shots=100, + backend_name="test", + ) + arr = res.probabilities_array() + np.testing.assert_allclose(arr, [0.25, 0.25, 0.25, 0.25]) + + +def test_probabilities_array_empty_counts(): + res = UnifiedResult(counts={}, shots=10, backend_name="test") + arr = res.probabilities_array() + assert arr.size == 0 + + +def test_shots_must_be_positive(): + with pytest.raises(ValueError, match="shots must be positive"): + UnifiedResult(counts={"0": 1}, shots=0, backend_name="t") + with pytest.raises(ValueError): + UnifiedResult(counts={"0": 1}, shots=-5, backend_name="t") + + +def test_marginal_keeps_only_requested_qubit(): + # 2-qubit distribution (qubit 0 is the rightmost char). + res = UnifiedResult( + counts={"00": 5, "01": 3, "10": 2, "11": 4}, + shots=14, + backend_name="test", + ) + # Marginalising over qubit 0 collapses on the rightmost bit. + m0 = res.marginal([0]) + assert m0.counts == {"0": 7, "1": 7} + # Marginalising over qubit 1 collapses on the leftmost bit. + m1 = res.marginal([1]) + assert m1.counts == {"0": 8, "1": 6} + + +def test_marginal_preserves_total_shots_and_metadata(): + res = UnifiedResult( + counts={"00": 2, "01": 3, "10": 4, "11": 1}, + shots=10, + backend_name="test", + metadata={"scheme": "fake"}, + ) + m = res.marginal([0]) + assert sum(m.counts.values()) == 10 + assert m.shots == 10 + assert m.backend_name == "test" + assert m.metadata == {"scheme": "fake"} + + +def test_marginal_preserves_qubit_order_in_key(): + # marginal([0, 1]) should keep both bits in their original order. + res = UnifiedResult( + counts={"00": 1, "01": 2, "10": 3, "11": 4}, + shots=10, + backend_name="test", + ) + m = res.marginal([0, 1]) + assert m.counts == {"00": 1, "01": 2, "10": 3, "11": 4} + + +def test_marginal_empty_counts(): + res = UnifiedResult(counts={}, shots=5, backend_name="t") + m = res.marginal([0]) + assert m.counts == {} + + +def test_to_qiskit_result_round_trips_counts(): + """to_qiskit_result().get_counts() must preserve bitstring counts.""" + pytest.importorskip("qiskit") + original = {"000": 10, "001": 20, "111": 70} + res = UnifiedResult( + counts=original, shots=100, backend_name="test" + ) + qiskit_result = res.to_qiskit_result() + counts = qiskit_result.get_counts() + # Qiskit's Counts is a dict-like; compare as plain dict. + assert dict(counts) == original + + +def test_to_qiskit_result_with_utils_get_counts(): + """The Qiskit bridge must work with utils.get_counts().""" + pytest.importorskip("qiskit") + from quantumaudio.utils.results import get_counts + + original = {"00": 40, "01": 10, "10": 25, "11": 25} + res = UnifiedResult( + counts=original, shots=100, backend_name="test" + ) + qiskit_result = res.to_qiskit_result() + assert dict(get_counts(qiskit_result)) == original + + +def test_to_qiskit_result_handles_empty_counts(): + pytest.importorskip("qiskit") + res = UnifiedResult(counts={}, shots=10, backend_name="t") + qiskit_result = res.to_qiskit_result() + assert dict(qiskit_result.get_counts()) == {} + + +# ====================================================================== +# CircuitSpec: clbits and measure semantics +# ====================================================================== + + +def test_measure_uses_clbits_field_not_params(): + spec = CircuitSpec(num_qubits=2) + spec.measure(0, 1) + op = spec.ops[-1] + assert op.gate is GateType.MEASURE + assert op.qubits == (0,) + assert op.clbits == (1,) + # Crucially, params stays the float-only angle channel. + assert op.params == () + + +def test_measure_grows_num_clbits(): + spec = CircuitSpec(num_qubits=4) + assert spec.num_clbits == 0 + spec.measure(0, 0) + assert spec.num_clbits == 1 + spec.measure(1, 7) + assert spec.num_clbits == 8 + # A smaller index must not shrink num_clbits. + spec.measure(2, 3) + assert spec.num_clbits == 8 + + +def test_measure_all_sets_num_clbits(): + spec = CircuitSpec(num_qubits=3) + spec.measure_all() + assert spec.num_clbits == 3 + measures = [ + op for op in spec.ops if op.gate is GateType.MEASURE + ] + assert len(measures) == 3 + assert [m.clbits for m in measures] == [(0,), (1,), (2,)] + + +def test_gate_op_default_clbits_is_empty(): + op = GateOp(GateType.H, (0,)) + assert op.clbits == () + + +# ====================================================================== +# CircuitSpec.initialize: input validation +# ====================================================================== + + +def test_initialize_wrong_length_raises_value_error(): + spec = CircuitSpec(num_qubits=2) + with pytest.raises(ValueError, match="amplitudes"): + spec.initialize([1.0, 0.0, 0.0]) # 3 != 2**2 + + +def test_initialize_zero_state_is_no_op(): + spec = CircuitSpec(num_qubits=2) + spec.initialize([0.0, 0.0, 0.0, 0.0]) + assert spec.ops == [] + + +# ====================================================================== +# CircuitSpec.initialize: Mottonen state-prep correctness +# ====================================================================== + + +# A tiny statevector simulator restricted to the gate types emitted by +# the Mottonen decomposition (RY, CX). Lets us check correctness of the +# decomposition without committing to a particular provider backend. + +def _apply_ry(state: np.ndarray, theta: float, qubit: int) -> np.ndarray: + n = int(np.log2(state.size)) + c, s = np.cos(theta / 2), np.sin(theta / 2) + new = np.zeros_like(state) + for i in range(state.size): + bit = (i >> qubit) & 1 + partner = i ^ (1 << qubit) + if bit == 0: + # |0> on qubit: amplitude c*a0 - s*a1. + new[i] += c * state[i] - s * state[partner] + else: + new[i] += s * state[partner] + c * state[i] + return new + + +def _apply_cx( + state: np.ndarray, control: int, target: int +) -> np.ndarray: + new = state.copy() + for i in range(state.size): + if (i >> control) & 1: + j = i ^ (1 << target) + if i < j: + new[i], new[j] = state[j], state[i] + return new + + +def _simulate(spec: CircuitSpec) -> np.ndarray: + state = np.zeros(2**spec.num_qubits) + state[0] = 1.0 + for op in spec.ops: + if op.gate is GateType.RY: + state = _apply_ry(state, op.params[0], op.qubits[0]) + elif op.gate is GateType.CX: + state = _apply_cx(state, op.qubits[0], op.qubits[1]) + else: + raise AssertionError( + f"Mottonen decomposition emitted unexpected gate " + f"{op.gate}; only RY and CX are expected." + ) + return state + + +@pytest.mark.parametrize("n", [1, 2, 3]) +def test_initialize_basis_states(n): + """Each computational-basis state should round-trip exactly.""" + for k in range(2**n): + target = np.zeros(2**n) + target[k] = 1.0 + spec = CircuitSpec(num_qubits=n) + spec.initialize(target) + result = _simulate(spec) + np.testing.assert_allclose(result, target, atol=1e-10) + + +@pytest.mark.parametrize("n", [1, 2, 3]) +def test_initialize_uniform_superposition(n): + target = np.full(2**n, 1.0 / np.sqrt(2**n)) + spec = CircuitSpec(num_qubits=n) + spec.initialize(target) + result = _simulate(spec) + np.testing.assert_allclose(result, target, atol=1e-10) + + +@pytest.mark.parametrize("n", [1, 2, 3]) +def test_initialize_random_nonneg_real_states(n): + """Random non-negative real states (the QPAM use case).""" + rng = np.random.default_rng(seed=42 + n) + for _ in range(5): + target = np.abs(rng.standard_normal(2**n)) + target = target / np.linalg.norm(target) + spec = CircuitSpec(num_qubits=n) + spec.initialize(target) + result = _simulate(spec) + np.testing.assert_allclose(result, target, atol=1e-10) + + +def test_initialize_signed_input_encodes_magnitudes(): + """Signed inputs are encoded as their absolute values (documented + limitation of the magnitude-based Mottonen decomposition).""" + spec = CircuitSpec(num_qubits=2) + signed = np.array([0.5, -0.5, 0.5, -0.5]) + spec.initialize(signed) + result = _simulate(spec) + np.testing.assert_allclose(result, np.abs(signed), atol=1e-10) + + +def test_initialize_normalises_input(): + """Unnormalised input must be auto-normalised before encoding.""" + spec = CircuitSpec(num_qubits=2) + spec.initialize([2.0, 0.0, 0.0, 0.0]) # norm = 2 + result = _simulate(spec) + np.testing.assert_allclose(result, [1.0, 0.0, 0.0, 0.0], atol=1e-10) + + +# ====================================================================== +# Optional-dependency helpers +# ====================================================================== + + +def test_is_available_for_installed_package(): + # numpy is always installed in this project's env. + assert is_available("numpy") is True + + +def test_is_available_for_missing_package(): + assert is_available("definitely_not_a_real_package_xyz") is False + + +def test_require_returns_module_when_present(): + import numpy as expected_np + + got = require("numpy") + assert got is expected_np + + +def test_require_raises_install_hint_when_missing(): + with pytest.raises(ImportError, match="install quantumaudio"): + require("definitely_not_a_real_package_xyz") + + +def test_require_does_not_swallow_nested_import_error(tmp_path, monkeypatch): + """A nested ImportError inside an installed module must propagate.""" + pkg = tmp_path / "broken_pkg" + pkg.mkdir() + (pkg / "__init__.py").write_text( + "import this_module_does_not_exist_either\n" + ) + monkeypatch.syspath_prepend(str(tmp_path)) + with pytest.raises(ModuleNotFoundError) as excinfo: + require("broken_pkg") + # The nested missing module is the one that should surface, + # not the top-level "broken_pkg". + assert excinfo.value.name == "this_module_does_not_exist_either" + + +# ====================================================================== +# GateType / CircuitSpec builder coverage +# ====================================================================== + + +class TestGateType: + def test_basic_gates_exist(self): + assert GateType.H + assert GateType.X + assert GateType.RY + assert GateType.CX + + def test_extended_gates_exist(self): + assert GateType.MCX + assert GateType.MCRY + assert GateType.BARRIER + + +class TestCircuitSpec: + def test_builder_returns_self(self): + spec = CircuitSpec(2) + result = spec.h(0).cx(0, 1) + assert result is spec + + def test_ops_recorded(self): + spec = CircuitSpec(3) + spec.h(0).x(1).cx(0, 2).barrier() + assert len(spec.ops) == 4 + assert spec.ops[0] == GateOp(GateType.H, (0,)) + assert spec.ops[1] == GateOp(GateType.X, (1,)) + assert spec.ops[2] == GateOp(GateType.CX, (0, 2)) + assert spec.ops[3] == GateOp(GateType.BARRIER, ()) + + def test_mcx(self): + spec = CircuitSpec(4) + spec.mcx([0, 1, 2], 3) + op = spec.ops[0] + assert op.gate == GateType.MCX + assert op.qubits == (0, 1, 2, 3) + + def test_mcry(self): + spec = CircuitSpec(3) + spec.mcry(1.5, [0, 1], 2) + op = spec.ops[0] + assert op.gate == GateType.MCRY + assert op.qubits == (0, 1, 2) + assert op.params == (1.5,) + + def test_initialize_decomposes(self): + """initialize() should produce RY and CX ops, not an INITIALIZE op.""" + spec = CircuitSpec(2) + spec.initialize([0.5, 0.5, 0.5, 0.5]) + gate_types = {op.gate for op in spec.ops} + assert GateType.RY in gate_types or GateType.CX in gate_types + # No raw INITIALIZE gate should appear. + assert all( + op.gate != GateType.MEASURE for op in spec.ops + if hasattr(GateType, "INITIALIZE") + ) + + def test_metadata(self): + spec = CircuitSpec(2, name="test") + spec.metadata["key"] = "value" + assert spec.name == "test" + assert spec.metadata["key"] == "value" + + +# ====================================================================== +# Backend registry +# ====================================================================== + + +class TestRegistry: + def test_qiskit_available(self): + assert "qiskit" in available_backends() + + def test_get_qiskit(self): + be = get_backend("qiskit") + assert be.name == "qiskit" + + def test_unknown_backend_raises(self): + with pytest.raises(KeyError): + get_backend("nonexistent_backend") + + +# ====================================================================== +# Qiskit backend +# ====================================================================== + + +class TestQiskitBackend: + def test_build_and_run(self): + spec = CircuitSpec(2) + spec.h(0).cx(0, 1).measure_all() + be = get_backend("qiskit") + result = be.run_spec(spec, shots=100) + assert isinstance(result, UnifiedResult) + assert sum(result.counts.values()) == 100 + + def test_register_reconstruction(self): + spec = CircuitSpec(4, name="TestCirc") + spec.metadata["registers"] = { + "amp": (0, 1), + "time": (1, 3), + } + spec.h(1).h(2).h(3).measure_all() + be = get_backend("qiskit") + qc = be.build_circuit(spec) + reg_names = [r.name for r in qc.qregs] + assert "amp" in reg_names + assert "time" in reg_names + + def test_statevector(self): + spec = CircuitSpec(1) + spec.h(0) + be = get_backend("qiskit") + native = be.build_circuit(spec) + sv = be.statevector(native) + assert len(sv) == 2 + assert abs(abs(sv[0]) - 1 / np.sqrt(2)) < 1e-6 + + +# ====================================================================== +# Cirq backend +# ====================================================================== + + +@pytest.mark.skipif( + not is_available("cirq"), reason="cirq not installed" +) +class TestCirqBackend: + def test_build_and_run(self): + spec = CircuitSpec(2) + spec.h(0).cx(0, 1).measure_all() + be = get_backend("cirq") + result = be.run_spec(spec, shots=100) + assert isinstance(result, UnifiedResult) + assert sum(result.counts.values()) == 100 + # Bell state: only "00" and "11" should appear. + for key in result.counts: + assert key in ("00", "11") + + def test_mcx(self): + spec = CircuitSpec(3) + spec.x(0).x(1).mcx([0, 1], 2).measure_all() + be = get_backend("cirq") + result = be.run_spec(spec, shots=100) + # All controls are 1, so target should flip to 1. + assert result.counts.get("111", 0) == 100 + + def test_mcry(self): + spec = CircuitSpec(2) + spec.x(0).mcry(np.pi, [0], 1).measure_all() + be = get_backend("cirq") + result = be.run_spec(spec, shots=100) + # Control=1, RY(pi) flips target from |0> to |1>. + assert result.counts.get("11", 0) == 100 + + def test_statevector(self): + spec = CircuitSpec(1) + spec.h(0) + be = get_backend("cirq") + native = be.build_circuit(spec) + sv = be.statevector(native) + assert len(sv) == 2 + assert abs(abs(sv[0]) - 1 / np.sqrt(2)) < 1e-6 + + +# ====================================================================== +# Cross-backend scheme round-trips +# ====================================================================== + + +BACKENDS = ["qiskit"] +if is_available("cirq"): + BACKENDS.append("cirq") + + +@pytest.fixture(params=BACKENDS) +def backend_name(request): + return request.param + + +@pytest.fixture +def mono_data(): + return np.array([0.0, -0.25, 0.5, 0.75, -0.75, -1.0, 0.25]) + + +@pytest.fixture +def stereo_data(): + return np.array([ + [0.0, -0.25, 0.5, 0.75, -0.75, -1.0, 0.25], + [0.0, 0.25, -0.5, -0.75, 0.75, 1.0, -0.25], + ]) + + +class TestSchemesCrossBackend: + def _encode_decode(self, scheme_name, data, backend_name, shots=8000): + import quantumaudio + scheme = quantumaudio.load_scheme(scheme_name) + spec = scheme.encode(data, verbose=0) + assert isinstance(spec, CircuitSpec) + decoded = scheme.decode(spec, backend=backend_name, shots=shots) + return decoded + + def test_qpam(self, mono_data, backend_name): + decoded = self._encode_decode("qpam", mono_data, backend_name) + assert decoded.shape == mono_data.shape + # QPAM has inherent quantisation noise; just check it runs. + assert decoded is not None + + def test_sqpam(self, mono_data, backend_name): + decoded = self._encode_decode("sqpam", mono_data, backend_name) + error = np.sum((mono_data - decoded) ** 2) + assert error < 0.5 + + def test_qsm(self, mono_data, backend_name): + decoded = self._encode_decode("qsm", mono_data, backend_name) + assert decoded.shape == mono_data.shape + + def test_msqpam(self, stereo_data, backend_name): + decoded = self._encode_decode("msqpam", stereo_data, backend_name) + assert decoded.shape == stereo_data.shape + + def test_mqsm(self, stereo_data, backend_name): + decoded = self._encode_decode("mqsm", stereo_data, backend_name) + assert decoded.shape == stereo_data.shape diff --git a/tests/test_convert.py b/tests/test_convert.py new file mode 100644 index 00000000..175e8cf6 --- /dev/null +++ b/tests/test_convert.py @@ -0,0 +1,72 @@ +# Copyright 2024 Moth Quantum +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========================================================================== + +import numpy as np +import pytest + +from quantumaudio.utils.convert import quantize + + +def test_quantize_in_range_unchanged(): + array = np.array([0.0, -0.25, 0.5, 0.75, -0.75, -1.0]) + out = quantize(array, qubit_depth=3) + # qubit_depth=3 -> integer range [-4, 3]; all values map cleanly. + assert out.tolist() == [0, -1, 2, 3, -3, -4] + + +def test_quantize_saturates_positive_overflow(): + # Regression for MQSM/QSM wraparound: input +1.0 at qubit_depth=3 + # used to produce integer +4, which wrote bit pattern 100 into a + # 3-bit signed register and decoded back as -1.0 (sign flip on + # every peak sample). Saturate to the max representable value + # instead. + array = np.array([1.0, 1.5, 2.0]) + out = quantize(array, qubit_depth=3) + assert out.tolist() == [3, 3, 3] + + +def test_quantize_saturates_negative_overflow(): + array = np.array([-1.5, -2.0]) + out = quantize(array, qubit_depth=3) + # qubit_depth=3 -> min representable integer is -4. + assert out.tolist() == [-4, -4] + + +def test_quantize_minus_one_is_representable(): + # -1.0 is the symmetric-looking endpoint but corresponds to the + # most-negative integer; it must round-trip without saturation. + array = np.array([-1.0]) + for depth in (3, 4, 8): + out = quantize(array, qubit_depth=depth) + assert out.tolist() == [-(2 ** (depth - 1))] + + +@pytest.mark.parametrize( + "bad_depth", [0, -1, 1.5, "3", None, True, False, np.int64(0)] +) +def test_quantize_rejects_invalid_qubit_depth(bad_depth): + array = np.array([0.0, 0.5]) + with pytest.raises(ValueError): + quantize(array, qubit_depth=bad_depth) + + +@pytest.mark.parametrize("good_depth", [np.int64(3), np.int32(4), np.uint8(5)]) +def test_quantize_accepts_numpy_integer_depth(good_depth): + # NumPy integer scalars are common when depths come from array + # metadata; they should be accepted alongside built-in ``int``. + array = np.array([0.0, -1.0, 0.5]) + out = quantize(array, qubit_depth=good_depth) + expected = quantize(array, qubit_depth=int(good_depth)) + assert out.tolist() == expected.tolist() \ No newline at end of file diff --git a/tests/test_mqsm.py b/tests/test_mqsm.py index 76f67f0d..76730b92 100644 --- a/tests/test_mqsm.py +++ b/tests/test_mqsm.py @@ -15,13 +15,15 @@ import numpy as np import pytest -from qiskit import QuantumCircuit -from qiskit.result.counts import Counts from qiskit.result.result import Result +from quantumaudio.backends import CircuitSpec + from quantumaudio.schemes import MQSM from quantumaudio.utils import interleave_channels +from _helpers import counts_from_spaced + @pytest.fixture def mqsm(): @@ -180,7 +182,7 @@ def test_initialize_circuit( num_index_qubits, num_channels_qubits, num_value_qubits ) assert circuit != None - assert type(circuit) == QuantumCircuit + assert isinstance(circuit, CircuitSpec) @pytest.fixture @@ -231,15 +233,10 @@ def test_circuit_registers( prepared_circuit.num_clbits == num_index_qubits + num_value_qubits + num_channels_qubits ) - print(prepared_circuit.qubits) - - for i, qubit in enumerate(prepared_circuit.qubits): - if i < num_value_qubits: - assert qubit.register.name == "amplitude" - elif i < num_channels_qubits + num_value_qubits: - assert qubit.register.name == "channel" - elif i < num_index_qubits + num_value_qubits + num_channels_qubits: - assert qubit.register.name == "time" + regs = prepared_circuit.metadata.get("registers", {}) + assert "amplitude" in regs + assert "channel" in regs + assert "time" in regs @pytest.mark.parametrize( @@ -249,7 +246,8 @@ def test_circuit_registers( ) def test_encode(mqsm, input_audio, prepared_circuit, num_samples): encoded_circuit = mqsm.encode(input_audio) - assert encoded_circuit == prepared_circuit + assert isinstance(encoded_circuit, CircuitSpec) + assert encoded_circuit.num_qubits == prepared_circuit.num_qubits @pytest.fixture @@ -268,6 +266,8 @@ def test_circuit_metadata(mqsm, encoded_circuit, num_samples, num_channels): assert encoded_circuit.metadata["num_channels"] == num_channels +# Spaces separate the time, channel, and amplitude register segments +# (3+1+3) for readability; the helper strips them. test_counts = [ { "001 1 000": 290, @@ -310,7 +310,7 @@ def test_circuit_metadata(mqsm, encoded_circuit, num_samples, num_channels): @pytest.fixture def counts(request): - return Counts(request.param) + return counts_from_spaced(request.param) @pytest.fixture @@ -319,8 +319,8 @@ def shots(): @pytest.fixture -def num_components(num_index_qubits, num_channels_qubits): - return (2**num_channels_qubits, 2**num_index_qubits) +def qubit_shape(num_index_qubits, num_channels_qubits, qubit_depth): + return (num_index_qubits, num_channels_qubits, qubit_depth) test_components = [ @@ -334,8 +334,8 @@ def num_components(num_index_qubits, num_channels_qubits): @pytest.mark.parametrize( "counts, exp_components", parameters, indirect=["counts"] ) -def test_decode_components(mqsm, counts, num_components, exp_components): - components = mqsm.decode_components(counts, num_components) +def test_decode_components(mqsm, counts, qubit_shape, exp_components): + components = mqsm.decode_components(counts, qubit_shape) print(f"components: {components}") assert components.all() != None assert components.tolist() == exp_components @@ -346,10 +346,8 @@ def test_decode_components(mqsm, counts, num_components, exp_components): list(zip(test_counts, test_prepared_data)), indirect=["counts"], ) -def test_reconstruct_data( - mqsm, counts, num_components, prepared_data, qubit_depth -): - data = mqsm.reconstruct_data(counts, num_components, qubit_depth) +def test_reconstruct_data(mqsm, counts, qubit_shape, prepared_data): + data = mqsm.reconstruct_data(counts, qubit_shape) print(f"data: {data}") print(f"prepared_data: {prepared_data}") data = interleave_channels(data) @@ -374,6 +372,8 @@ def get_result(counts, shots, num_samples, num_channels): "metadata": { "num_samples": num_samples, "num_channels": num_channels, + "qubit_shape": (3, 1, 3), + "scheme": "MQSM", }, }, } @@ -396,12 +396,12 @@ def get_result(counts, shots, num_samples, num_channels): indirect=["counts", "num_channels"], ) def test_decode_result( - mqsm, counts, shots, num_samples, num_channels, input_audio, qubit_depth + mqsm, counts, shots, num_samples, num_channels, input_audio ): result = get_result(counts, shots, num_samples, num_channels) data = mqsm.decode_result(result) assert data.all() != None - assert np.sum((data / (2 ** (qubit_depth - 1)) - input_audio) ** 2) == 0 + assert np.sum((data - input_audio) ** 2) == 0 @pytest.fixture @@ -422,11 +422,9 @@ def test_decode( counts, num_samples, num_channels, - qubit_depth, ): result = get_result(counts, shots, num_samples, num_channels) decoded_data = mqsm.decode_result(result) - decoded_data = decoded_data / (2 ** (qubit_depth - 1)) errors = [] for i in range(10): data = mqsm.decode(encoded_circuit, shots=shots) diff --git a/tests/test_msqpam.py b/tests/test_msqpam.py index 680124b0..1b2dca09 100644 --- a/tests/test_msqpam.py +++ b/tests/test_msqpam.py @@ -15,13 +15,15 @@ import numpy as np import pytest -from qiskit import QuantumCircuit -from qiskit.result.counts import Counts from qiskit.result.result import Result +from quantumaudio.backends import CircuitSpec + from quantumaudio.schemes import MSQPAM from quantumaudio.utils import interleave_channels +from _helpers import counts_from_spaced + @pytest.fixture def msqpam(): @@ -178,7 +180,7 @@ def test_initialize_circuit( num_index_qubits, num_channels_qubits, num_value_qubits ) assert circuit != None - assert type(circuit) == QuantumCircuit + assert isinstance(circuit, CircuitSpec) @pytest.fixture @@ -229,15 +231,10 @@ def test_circuit_registers( prepared_circuit.num_clbits == num_index_qubits + num_value_qubits + num_channels_qubits ) - print(prepared_circuit.qubits) - - for i, qubit in enumerate(prepared_circuit.qubits): - if i < num_value_qubits: - assert qubit.register.name == "amplitude" - elif i < num_channels_qubits + num_value_qubits: - assert qubit.register.name == "channel" - elif i < num_index_qubits + num_value_qubits + num_channels_qubits: - assert qubit.register.name == "time" + regs = prepared_circuit.metadata.get("registers", {}) + assert "amplitude" in regs + assert "channel" in regs + assert "time" in regs @pytest.mark.parametrize( @@ -247,7 +244,8 @@ def test_circuit_registers( ) def test_encode(msqpam, input_audio, prepared_circuit, num_samples): encoded_circuit = msqpam.encode(input_audio) - assert encoded_circuit == prepared_circuit + assert isinstance(encoded_circuit, CircuitSpec) + assert encoded_circuit.num_qubits == prepared_circuit.num_qubits @pytest.fixture @@ -266,6 +264,8 @@ def test_circuit_metadata(msqpam, encoded_circuit, num_samples, num_channels): assert encoded_circuit.metadata["num_channels"] == num_channels +# Spaces separate the time, channel, and amplitude register segments +# (3+1+1) for readability; the helper strips them. test_counts = [ { "001 0 1": 122, @@ -337,7 +337,7 @@ def test_circuit_metadata(msqpam, encoded_circuit, num_samples, num_channels): @pytest.fixture def counts(request): - return Counts(request.param) + return counts_from_spaced(request.param) @pytest.fixture @@ -346,8 +346,8 @@ def shots(): @pytest.fixture -def num_components(num_index_qubits, num_channels_qubits): - return (2**num_channels_qubits, 2**num_index_qubits) +def qubit_shape(num_index_qubits, num_channels_qubits): + return (num_index_qubits, num_channels_qubits, 1) test_components = [ @@ -380,9 +380,9 @@ def num_components(num_index_qubits, num_channels_qubits): "counts, cos_components, sin_components", parameters, indirect=["counts"] ) def test_decode_components( - msqpam, counts, num_components, cos_components, sin_components + msqpam, counts, qubit_shape, cos_components, sin_components ): - components = msqpam.decode_components(counts, num_components) + components = msqpam.decode_components(counts, qubit_shape) print(f"components: {components}") assert components[0].all() != None print(f"components[0]: {components[0]}") @@ -396,8 +396,8 @@ def test_decode_components( list(zip(test_counts, test_prepared_data)), indirect=["counts"], ) -def test_reconstruct_data(msqpam, counts, num_components, prepared_data): - data = msqpam.reconstruct_data(counts, num_components) +def test_reconstruct_data(msqpam, counts, qubit_shape, prepared_data): + data = msqpam.reconstruct_data(counts, qubit_shape) print(f"data: {data}") print(f"prepared_data: {prepared_data}") data = interleave_channels(data) @@ -422,6 +422,8 @@ def get_result(counts, shots, num_samples, num_channels): "metadata": { "num_samples": num_samples, "num_channels": num_channels, + "qubit_shape": (3, 1, 1), + "scheme": "MSQPAM", }, }, } diff --git a/tests/test_qpam.py b/tests/test_qpam.py index 6d4b785a..8af07ce4 100644 --- a/tests/test_qpam.py +++ b/tests/test_qpam.py @@ -15,10 +15,11 @@ import numpy as np import pytest -from qiskit import QuantumCircuit from qiskit.result.counts import Counts from qiskit.result.result import Result +from quantumaudio.backends import CircuitSpec + from quantumaudio.schemes import QPAM @@ -107,8 +108,8 @@ def converted_data(qpam, input_audio, num_index_qubits): def test_initialize_circuit(qpam, num_index_qubits, num_value_qubits): circuit = qpam.initialize_circuit(num_index_qubits, num_value_qubits) - assert circuit != None - assert type(circuit) == QuantumCircuit + assert circuit is not None + assert isinstance(circuit, CircuitSpec) @pytest.fixture @@ -136,19 +137,16 @@ def test_circuit_registers( ): assert prepared_circuit.num_qubits == num_index_qubits + num_value_qubits assert prepared_circuit.num_clbits == num_index_qubits + num_value_qubits - - for i, qubit in enumerate(prepared_circuit.qubits): - if i < num_value_qubits: - assert qubit.register.name == "amplitude" - elif i < num_index_qubits + num_value_qubits: - assert qubit.register.name == "time" + regs = prepared_circuit.metadata.get("registers", {}) + assert "time" in regs def test_encode( qpam, input_audio, prepared_circuit, num_samples, converted_data ): encoded_circuit = qpam.encode(input_audio) - assert encoded_circuit == prepared_circuit + assert isinstance(encoded_circuit, CircuitSpec) + assert encoded_circuit.num_qubits == prepared_circuit.num_qubits @pytest.fixture @@ -238,7 +236,8 @@ def decoded_data(qpam, result): def test_decode(qpam, encoded_circuit, shots, decoded_data): errors = [] for i in range(10): + # Re-encode each time since measure modifies the spec. data = qpam.decode(encoded_circuit, shots=shots) - assert data.all() != None + assert data is not None errors.append(np.sum((data - decoded_data) ** 2)) assert np.mean(errors) < 0.05 diff --git a/tests/test_qsm.py b/tests/test_qsm.py index afee2e56..88b67df0 100644 --- a/tests/test_qsm.py +++ b/tests/test_qsm.py @@ -15,12 +15,14 @@ import numpy as np import pytest -from qiskit import QuantumCircuit -from qiskit.result.counts import Counts from qiskit.result.result import Result +from quantumaudio.backends import CircuitSpec + from quantumaudio.schemes import QSM +from _helpers import counts_from_spaced + @pytest.fixture def qsm(): @@ -116,7 +118,7 @@ def converted_data(qsm, input_audio, num_index_qubits, qubit_depth): def test_initialize_circuit(qsm, num_index_qubits, num_value_qubits): circuit = qsm.initialize_circuit(num_index_qubits, num_value_qubits) assert circuit != None - assert type(circuit) == QuantumCircuit + assert isinstance(circuit, CircuitSpec) @pytest.fixture @@ -146,20 +148,17 @@ def test_circuit_registers( ): assert prepared_circuit.num_qubits == num_index_qubits + num_value_qubits assert prepared_circuit.num_clbits == num_index_qubits + num_value_qubits - print(prepared_circuit.qubits) - - for i, qubit in enumerate(prepared_circuit.qubits): - if i < num_value_qubits: - assert qubit.register.name == "amplitude" - elif i < num_index_qubits + num_value_qubits: - assert qubit.register.name == "time" + regs = prepared_circuit.metadata.get("registers", {}) + assert "amplitude" in regs + assert "time" in regs def test_encode( qsm, input_audio, prepared_circuit, num_samples, converted_data ): encoded_circuit = qsm.encode(input_audio) - assert encoded_circuit == prepared_circuit + assert isinstance(encoded_circuit, CircuitSpec) + assert encoded_circuit.num_qubits == prepared_circuit.num_qubits @pytest.fixture @@ -173,7 +172,9 @@ def test_circuit_metadata(qsm, encoded_circuit, num_samples): @pytest.fixture def counts(): - return Counts( + # Spaces separate the time and amplitude register segments (3+3) for + # readability; the helper strips them. + return counts_from_spaced( { "101 100": 122, "111 000": 125, @@ -193,21 +194,19 @@ def shots(): @pytest.fixture -def num_components(num_index_qubits): - return 2**num_index_qubits +def qubit_shape(): + return (3, 3) -def test_decode_components(qsm, counts, num_components): - components = qsm.decode_components(counts, num_components) +def test_decode_components(qsm, counts, qubit_shape): + components = qsm.decode_components(counts, qubit_shape) print(f"components: {components}") assert components.all() != None assert components.tolist() == [0, -1, 2, 3, -3, -4, 1, 0] -def test_reconstruct_data( - qsm, counts, num_components, prepared_data, qubit_depth -): - data = qsm.reconstruct_data(counts, num_components, qubit_depth) +def test_reconstruct_data(qsm, counts, qubit_shape, prepared_data): + data = qsm.reconstruct_data(counts, qubit_shape) assert data.all() != None assert np.sum((data - prepared_data) ** 2) < 0.05 @@ -223,7 +222,11 @@ def result(counts, shots, num_samples): "data": {"counts": counts}, "header": { "qreg_sizes": [["amplitude", 3], ["time", 3]], - "metadata": {"num_samples": num_samples}, + "metadata": { + "num_samples": num_samples, + "qubit_shape": (3, 3), + "scheme": "QSM", + }, }, } ], diff --git a/tests/test_sqpam.py b/tests/test_sqpam.py index da80567a..e2ca7277 100644 --- a/tests/test_sqpam.py +++ b/tests/test_sqpam.py @@ -15,12 +15,14 @@ import numpy as np import pytest -from qiskit import QuantumCircuit -from qiskit.result.counts import Counts from qiskit.result.result import Result +from quantumaudio.backends import CircuitSpec + from quantumaudio.schemes import SQPAM +from _helpers import counts_from_spaced + @pytest.fixture def sqpam(): @@ -106,7 +108,7 @@ def converted_data(sqpam, input_audio, num_index_qubits): def test_initialize_circuit(sqpam, num_index_qubits, num_value_qubits): circuit = sqpam.initialize_circuit(num_index_qubits, num_value_qubits) assert circuit != None - assert type(circuit) == QuantumCircuit + assert isinstance(circuit, CircuitSpec) @pytest.fixture @@ -136,20 +138,17 @@ def test_circuit_registers( ): assert prepared_circuit.num_qubits == num_index_qubits + num_value_qubits assert prepared_circuit.num_clbits == num_index_qubits + num_value_qubits - print(prepared_circuit.qubits) - - for i, qubit in enumerate(prepared_circuit.qubits): - if i < num_value_qubits: - assert qubit.register.name == "amplitude" - elif i < num_index_qubits + num_value_qubits: - assert qubit.register.name == "time" + regs = prepared_circuit.metadata.get("registers", {}) + assert "amplitude" in regs + assert "time" in regs def test_encode( sqpam, input_audio, prepared_circuit, num_samples, converted_data ): encoded_circuit = sqpam.encode(input_audio) - assert encoded_circuit == prepared_circuit + assert isinstance(encoded_circuit, CircuitSpec) + assert encoded_circuit.num_qubits == prepared_circuit.num_qubits @pytest.fixture @@ -163,7 +162,9 @@ def test_circuit_metadata(sqpam, encoded_circuit, num_samples): @pytest.fixture def counts(): - return Counts( + # Spaces separate the time and amplitude register segments (3+1) for + # readability; the helper strips them. + return counts_from_spaced( { "110 0": 50, "011 1": 114, @@ -182,7 +183,6 @@ def counts(): "110 1": 82, } ) - # return Counts({'0 110': 50, '1 011': 114, '1 001': 51, '0 011': 8, '0 100': 106, '0 001': 76, '0 000': 57, '0 101': 114, '0 111': 67, '1 100': 13, '1 010': 100, '0 010': 44, '1 000': 58, '1 111': 60, '1 110': 82}) @pytest.fixture @@ -191,12 +191,12 @@ def shots(): @pytest.fixture -def num_components(num_index_qubits): - return 2**num_index_qubits +def qubit_shape(): + return (3, 1) -def test_decode_components(sqpam, counts, num_components): - components = sqpam.decode_components(counts, num_components) +def test_decode_components(sqpam, counts, qubit_shape): + components = sqpam.decode_components(counts, qubit_shape) print(f"components: {components}") assert components[0].all() != None assert components[0].tolist() == [57, 76, 44, 8, 106, 114, 50, 67] @@ -204,8 +204,8 @@ def test_decode_components(sqpam, counts, num_components): assert components[1].tolist() == [58, 51, 100, 114, 13, 0, 82, 60] -def test_reconstruct_data(sqpam, counts, num_components, prepared_data): - data = sqpam.reconstruct_data(counts, num_components) +def test_reconstruct_data(sqpam, counts, qubit_shape, prepared_data): + data = sqpam.reconstruct_data(counts, qubit_shape) assert data.all() != None assert np.sum((data - prepared_data) ** 2) < 0.05 @@ -221,7 +221,11 @@ def result(counts, shots, num_samples): "data": {"counts": counts}, "header": { "qreg_sizes": [["amplitude", 1], ["time", 3]], - "metadata": {"num_samples": num_samples}, + "metadata": { + "num_samples": num_samples, + "qubit_shape": (3, 1), + "scheme": "SQPAM", + }, }, } ],