diff --git a/src/qsub/data_classes.py b/src/qsub/data_classes.py new file mode 100644 index 0000000..af24326 --- /dev/null +++ b/src/qsub/data_classes.py @@ -0,0 +1,52 @@ + +from dataclasses import dataclass +from subroutine_model import SubroutineModel +@dataclass +class ObliviousAmplitudeAmplificationData: + input_state_squared_overlap:float = 0.5 + failure_tolerance: float = 0.001 +@dataclass +class StatePreparationOracleData: + failure_tolerance: float = 0.001 +@dataclass +class MarkedSubspaceOracleData: + failure_tolerance: float = 0.001 + +@dataclass +class LinearSystemBlockEncodingData: + failure_tolerance: float = 0.001 + sub_normaliation: float = 0.001 + condition_number: float = 0.001 + +@dataclass +class CarlemanBlockEncodingData: + failure_tolerance: float = 0.1 + kappa_P: float =0 + mu_P_A: float = 0 + A_stable: float = 0 + + +@dataclass +class LBMDragEstimationData: + failure_tolerance: float = 0 + estimation_error: float = 0 + estimated_drag_force: float = 0 + evolution_time: float = 0 + mu_P_A: float = 0 + kappa_P: float = 0 + norm_inhomogeneous_term_vector: float = 0 + norm_x_t: float = 0 + A_stable: bool = False + # Intialize subroutines as generic routines with task name + solve_quantum_ode: SubroutineModel = SubroutineModel("solve_quantum_ode") + mark_drag_vector: SubroutineModel = SubroutineModel("mark_drag_vector") + +@dataclass +class LBMDragReflectionData: + failure_tolerance: float = 0 + +@dataclass +class SphereBoundaryOracleData: + failure_tolerance: float = None, + radius: float = None, + grid_spacing: float = None, diff --git a/src/qsub/generic_block_encoding.py b/src/qsub/generic_block_encoding.py index 6fb260a..5dffc9e 100644 --- a/src/qsub/generic_block_encoding.py +++ b/src/qsub/generic_block_encoding.py @@ -2,7 +2,14 @@ from typing import Optional from .subroutine_model import SubroutineModel from sympy import symbols +from dataclasses import dataclass +@dataclass +class GenericLinearSystemBlockEncodingData: + failure_tolerance: float = 0.1 + kappa_P: float = 0.1, + mu_P_A: float = 0.1 + A_stable: float = 0.01 class GenericBlockEncoding(SubroutineModel): def __init__(self, task_name: str, requirements: Optional[dict] = None, **kwargs): @@ -38,29 +45,6 @@ def __init__(self, task_name: str, requirements: Optional[dict] = None, **kwargs if isinstance(value, SubroutineModel): setattr(self, attr, value) - def set_requirements( - self, - failure_tolerance: Optional[float] = None, - kappa_P: Optional[float] = None, - mu_P_A: Optional[float] = None, - A_stable: Optional[float] = None, - ): - args = locals() - # Clean up the args dictionary before setting requirements - args.pop("self") - args = { - k: v for k, v in args.items() if v is not None and not k.startswith("__") - } - # Initialize the requirements attribute if it doesn't exist - if not hasattr(self, "requirements"): - self.requirements = {} - - # Update the requirements with new values - self.requirements.update(args) - - # Call the parent class's set_requirements method with the updated requirements - super().set_requirements(**self.requirements) - def populate_requirements_for_subroutines(self): pass diff --git a/src/qsub/quantum_algorithms/differential_equation_solvers/linearization_methods.py b/src/qsub/quantum_algorithms/differential_equation_solvers/linearization_methods.py index dd3aa7c..7327c70 100644 --- a/src/qsub/quantum_algorithms/differential_equation_solvers/linearization_methods.py +++ b/src/qsub/quantum_algorithms/differential_equation_solvers/linearization_methods.py @@ -1,6 +1,6 @@ from qsub.subroutine_model import SubroutineModel from qsub.utils import consume_fraction_of_error_budget - +from dataclasses import dataclass from typing import Optional import warnings @@ -27,28 +27,6 @@ def __init__( "block_encode_quadratic_term" ) - def set_requirements( - self, - failure_tolerance: float = None, - kappa_P: float = None, - mu_P_A: float = None, - A_stable: bool = None, - ): - args = locals() - # Clean up the args dictionary before setting requirements - args.pop("self") - args = { - k: v for k, v in args.items() if v is not None and not k.startswith("__") - } - # Initialize the requirements attribute if it doesn't exist - if not hasattr(self, "requirements"): - self.requirements = {} - - # Update the requirements with new values - self.requirements.update(args) - - # Call the parent class's set_requirements method with the updated requirements - super().set_requirements(**self.requirements) def populate_requirements_for_subroutines(self): remaining_failure_tolerance = self.requirements["failure_tolerance"] diff --git a/src/qsub/quantum_algorithms/fluid_dynamics/lattice_boltzmann.py b/src/qsub/quantum_algorithms/fluid_dynamics/lattice_boltzmann.py index bd7e2d0..f36df17 100644 --- a/src/qsub/quantum_algorithms/fluid_dynamics/lattice_boltzmann.py +++ b/src/qsub/quantum_algorithms/fluid_dynamics/lattice_boltzmann.py @@ -2,6 +2,22 @@ from ...subroutine_model import SubroutineModel from qsub.utils import consume_fraction_of_error_budget import warnings +from dataclasses import dataclass + +@dataclass +class LBMDragEstimationData: + failure_tolerance: float = 0 + estimation_error: float = 0 + estimated_drag_force: float = 0 + evolution_time: float = 0 + mu_P_A: float = 0 + kappa_P: float = 0 + norm_inhomogeneous_term_vector: float = 0 + norm_x_t: float = 0 + A_stable: bool = False + # Intialize subroutines as generic routines with task name + solve_quantum_ode: SubroutineModel = SubroutineModel("solve_quantum_ode") + mark_drag_vector: SubroutineModel = SubroutineModel("mark_drag_vector") class LBMDragEstimation(SubroutineModel): @@ -18,47 +34,6 @@ def __init__( else: self.estimate_amplitude = SubroutineModel("estimate_amplitude") - # Initialize the sub-subtask requirements as generic subroutines with task names - self.requirements["solve_quantum_ode"] = SubroutineModel("solve_quantum_ode") - self.requirements["mark_drag_vector"] = SubroutineModel("mark_drag_vector") - - def set_requirements( - self, - failure_tolerance: float = None, - estimation_error: float = None, - estimated_drag_force: float = None, - evolution_time: float = None, - mu_P_A: float = None, - kappa_P: float = None, - norm_inhomogeneous_term_vector: float = None, - norm_x_t: float = None, - A_stable: bool = None, - solve_quantum_ode: Optional[SubroutineModel] = None, - mark_drag_vector: Optional[SubroutineModel] = None, - ): - args = locals() - # Clean up the args dictionary before setting requirements - args.pop("self") - args = { - k: v for k, v in args.items() if v is not None and not k.startswith("__") - } - - # Initialize the requirements attribute if it doesn't exist - if not hasattr(self, "requirements"): - self.requirements = {} - - for k, v in args.items(): - if k in self.requirements: - if not isinstance(self.requirements[k], SubroutineModel): - self.requirements[k] = v - else: - self.requirements[k] = v if v is not None else SubroutineModel(k) - - # Update the requirements with new values - self.requirements.update(args) - - # Call the parent class's set_requirements method with the updated requirements - super().set_requirements(**self.requirements) def populate_requirements_for_subroutines(self): # Note: This subroutine consumes no failure probability. @@ -114,7 +89,6 @@ def populate_requirements_for_subroutines(self): def count_qubits(self): return self.estimate_amplitude.count_qubits() - class LBMDragReflection(SubroutineModel): def __init__( self, @@ -129,25 +103,6 @@ def __init__( else: self.compute_boundary = SubroutineModel("compute_boundary") - def set_requirements( - self, - failure_tolerance: float = None, - ): - args = locals() - # Clean up the args dictionary before setting requirements - args.pop("self") - args = { - k: v for k, v in args.items() if v is not None and not k.startswith("__") - } - # Initialize the requirements attribute if it doesn't exist - if not hasattr(self, "requirements"): - self.requirements = {} - - # Update the requirements with new values - self.requirements.update(args) - - # Call the parent class's set_requirements method with the updated requirements - super().set_requirements(**self.requirements) def populate_requirements_for_subroutines(self): # Set number of calls to the quadratic term block encoding @@ -163,7 +118,6 @@ def get_normalization_factor(self): warnings.warn("This function is not fully implemented.", UserWarning) return 42 - class SphereBoundaryOracle(SubroutineModel): def __init__( self, @@ -190,28 +144,6 @@ def __init__( else: self.quantum_square = SubroutineModel("quantum_square") - def set_requirements( - self, - failure_tolerance: float = None, - radius: float = None, - grid_spacing: float = None, - ): - args = locals() - # Clean up the args dictionary before setting requirements - args.pop("self") - args = { - k: v for k, v in args.items() if v is not None and not k.startswith("__") - } - # Initialize the requirements attribute if it doesn't exist - if not hasattr(self, "requirements"): - self.requirements = {} - - # Update the requirements with new values - self.requirements.update(args) - - # Call the parent class's set_requirements method with the updated requirements - super().set_requirements(**self.requirements) - def populate_requirements_for_subroutines(self): remaining_failure_tolerance = self.requirements["failure_tolerance"] diff --git a/src/qsub/quantum_algorithms/general_quantum_algorithms/amplitude_amplification.py b/src/qsub/quantum_algorithms/general_quantum_algorithms/amplitude_amplification.py index c8a4976..5f6aac7 100644 --- a/src/qsub/quantum_algorithms/general_quantum_algorithms/amplitude_amplification.py +++ b/src/qsub/quantum_algorithms/general_quantum_algorithms/amplitude_amplification.py @@ -1,49 +1,22 @@ import numpy as np from typing import Optional from ...subroutine_model import SubroutineModel +from data_classes import StatePreparationOracleData, MarkedSubspaceOracleData class ObliviousAmplitudeAmplification(SubroutineModel): def __init__( self, + state_preparation_oracle: SubroutineModel, + mark_subspace: SubroutineModel, task_name="amplify_amplitude", - requirements=None, - state_preparation_oracle: Optional[SubroutineModel] = None, - mark_subspace: Optional[SubroutineModel] = None, ): - super().__init__(task_name, requirements) - - if state_preparation_oracle is not None: - self.state_preparation_oracle = state_preparation_oracle - else: - self.state_preparation_oracle = SubroutineModel("state_preparation_oracle") - - if mark_subspace is not None: - self.mark_subspace = mark_subspace - else: - self.mark_subspace = SubroutineModel("mark_subspace") - - def set_requirements( - self, - input_state_squared_overlap: float = None, - failure_tolerance: float = None, - ): - args = locals() - # Clean up the args dictionary before setting requirements - args.pop("self") - args = { - k: v for k, v in args.items() if v is not None and not k.startswith("__") - } - # Initialize the requirements attribute if it doesn't exist - if not hasattr(self, "requirements"): - self.requirements = {} - - # Update the requirements with new values - self.requirements.update(args) - - # Call the parent class's set_requirements method with the updated requirements - super().set_requirements(**self.requirements) - + super().__init__(task_name) + assert isinstance(state_preparation_oracle,SubroutineModel) + assert isinstance(mark_subspace, SubroutineModel) + self.state_preparation_oracle = state_preparation_oracle + self.mark_subspace = mark_subspace + def populate_requirements_for_subroutines(self): # Populate requirements for state_preparation_oracle and mark_subspace @@ -64,17 +37,16 @@ def populate_requirements_for_subroutines(self): # Set number of times called to number of Grover iterates self.state_preparation_oracle.number_of_times_called = number_of_grover_iterates - self.mark_subspace.number_of_times_called = number_of_grover_iterates - - self.state_preparation_oracle.set_requirements( - failure_tolerance=subroutine_error_budget_allocation[0] - * remaining_failure_tolerance, - ) - self.mark_subspace.set_requirements( - failure_tolerance=subroutine_error_budget_allocation[1] - * remaining_failure_tolerance, - ) + self.mark_subspace.number_of_times_called = number_of_grover_iterates + + StatePreparationOracleData.failure_tolerance = subroutine_error_budget_allocation[0] \ + * remaining_failure_tolerance + MarkedSubspaceOracleData.failure_tolerance = subroutine_error_budget_allocation[1] \ + * remaining_failure_tolerance + + self.state_preparation_oracle.set_requirements(StatePreparationOracleData) + self.mark_subspace.set_requirements(MarkedSubspaceOracleData) def count_qubits(self): return self.state_preparation_oracle.count_qubits() @@ -109,4 +81,3 @@ def test_oblivious_amplitude_amplification(): return print(obl_amp.count_subroutines()) -# test_oblivious_amplitude_amplification() diff --git a/src/qsub/quantum_algorithms/general_quantum_algorithms/amplitude_estimation.py b/src/qsub/quantum_algorithms/general_quantum_algorithms/amplitude_estimation.py index 7b7405e..cc575df 100644 --- a/src/qsub/quantum_algorithms/general_quantum_algorithms/amplitude_estimation.py +++ b/src/qsub/quantum_algorithms/general_quantum_algorithms/amplitude_estimation.py @@ -2,49 +2,21 @@ from typing import Optional from ...subroutine_model import SubroutineModel import warnings - - +from data_classes import StatePreparationOracleData, MarkedSubspaceOracleData class QuantumAmplitudeEstimation(SubroutineModel): def __init__( self, + state_preparation_oracle: SubroutineModel, + mark_subspace: SubroutineModel, task_name="estimate_amplitude", - requirements=None, - state_preparation_oracle: Optional[SubroutineModel] = None, - mark_subspace: Optional[SubroutineModel] = None, - ): - super().__init__(task_name, requirements) - - if state_preparation_oracle is not None: - self.state_preparation_oracle = state_preparation_oracle - else: - self.state_preparation_oracle = SubroutineModel("state_preparation_oracle") - - if mark_subspace is not None: - self.mark_subspace = mark_subspace - else: - self.mark_subspace = SubroutineModel("mark_subspace") - def set_requirements( - self, - estimation_error: float = None, - failure_tolerance: float = None, - amplitude: float = None, ): - args = locals() - # Clean up the args dictionary before setting requirements - args.pop("self") - args = { - k: v for k, v in args.items() if v is not None and not k.startswith("__") - } - # Initialize the requirements attribute if it doesn't exist - if not hasattr(self, "requirements"): - self.requirements = {} - - # Update the requirements with new values - self.requirements.update(args) - - # Call the parent class's set_requirements method with the updated requirements - super().set_requirements(**self.requirements) + super().__init__(task_name) + assert isinstance(state_preparation_oracle,SubroutineModel) + assert isinstance(mark_subspace, SubroutineModel) + self.state_preparation_oracle = state_preparation_oracle + self.mark_subspace = mark_subspace + def populate_requirements_for_subroutines(self): # Populate requirements for state_preparation_oracle and mark_subspace @@ -72,17 +44,14 @@ def populate_requirements_for_subroutines(self): ) self.mark_subspace.number_of_times_called = 2 * number_of_grover_iterates - self.state_preparation_oracle.set_requirements( - failure_tolerance=subroutine_error_budget_allocation[0] - / self.state_preparation_oracle.number_of_times_called - * remaining_failure_tolerance, - ) + StatePreparationOracleData.failure_tolerance= (subroutine_error_budget_allocation[0]/ + self.state_preparation_oracle.number_of_times_called)* remaining_failure_tolerance + MarkedSubspaceOracleData.failure_tolerance = (subroutine_error_budget_allocation[1] + / self.mark_subspace.number_of_times_called) * remaining_failure_tolerance - self.mark_subspace.set_requirements( - failure_tolerance=subroutine_error_budget_allocation[1] - / self.mark_subspace.number_of_times_called - * remaining_failure_tolerance, - ) + self.state_preparation_oracle.set_requirements(StatePreparationOracleData) + + self.mark_subspace.set_requirements(MarkedSubspaceOracleData) def count_qubits(self): return self.state_preparation_oracle.count_qubits() @@ -110,43 +79,14 @@ class IterativeQuantumAmplitudeEstimation(SubroutineModel): def __init__( self, + state_preparation_oracle: SubroutineModel = None, + mark_subspace: SubroutineModel = None, task_name="estimate_amplitude", - requirements=None, - state_preparation_oracle: Optional[SubroutineModel] = None, - mark_subspace: Optional[SubroutineModel] = None, - ): - super().__init__(task_name, requirements) - - if state_preparation_oracle is not None: - self.state_preparation_oracle = state_preparation_oracle - else: - self.state_preparation_oracle = SubroutineModel("state_preparation_oracle") - - if mark_subspace is not None: - self.mark_subspace = mark_subspace - else: - self.mark_subspace = SubroutineModel("mark_subspace") - def set_requirements( - self, - estimation_error: float = None, - failure_tolerance: float = None, ): - args = locals() - # Clean up the args dictionary before setting requirements - args.pop("self") - args = { - k: v for k, v in args.items() if v is not None and not k.startswith("__") - } - # Initialize the requirements attribute if it doesn't exist - if not hasattr(self, "requirements"): - self.requirements = {} - - # Update the requirements with new values - self.requirements.update(args) - - # Call the parent class's set_requirements method with the updated requirements - super().set_requirements(**self.requirements) + super().__init__(task_name) + self.state_preparation_oracle = state_preparation_oracle + self.mark_subspace = mark_subspace def populate_requirements_for_subroutines(self): # Populate requirements for state_preparation_oracle and mark_subspace @@ -172,15 +112,14 @@ def populate_requirements_for_subroutines(self): self.state_preparation_oracle.number_of_times_called = number_of_grover_iterates self.mark_subspace.number_of_times_called = number_of_grover_iterates - self.state_preparation_oracle.set_requirements( - failure_tolerance=subroutine_error_budget_allocation[0] - * remaining_failure_tolerance, - ) + StatePreparationOracleData.failure_tolerance=(subroutine_error_budget_allocation[0] + / self.state_preparation_oracle.number_of_times_called)* remaining_failure_tolerance + MarkedSubspaceOracleData.failure_tolerance = (subroutine_error_budget_allocation[1] + / self.mark_subspace.number_of_times_called)* remaining_failure_tolerance - self.mark_subspace.set_requirements( - failure_tolerance=subroutine_error_budget_allocation[1] - * remaining_failure_tolerance, - ) + self.state_preparation_oracle.set_requirements(StatePreparationOracleData) + + self.mark_subspace.set_requirements(MarkedSubspaceOracleData) def count_qubits(self): return self.state_preparation_oracle.count_qubits() diff --git a/src/qsub/quantum_algorithms/general_quantum_algorithms/linear_systems.py b/src/qsub/quantum_algorithms/general_quantum_algorithms/linear_systems.py index d29831b..ab7c5b9 100644 --- a/src/qsub/quantum_algorithms/general_quantum_algorithms/linear_systems.py +++ b/src/qsub/quantum_algorithms/general_quantum_algorithms/linear_systems.py @@ -6,50 +6,19 @@ from typing import Optional import math from qsub.utils import consume_fraction_of_error_budget +from data_classes import LinearSystemBlockEncodingData, StatePreparationOracleData class TaylorQLSA(SubroutineModel): def __init__( self, + linear_system_block_encoding: SubroutineModel, + prepare_b_vector: SubroutineModel, task_name="solve_quantum_linear_system", - requirements=None, - linear_system_block_encoding: Optional[SubroutineModel] = None, - prepare_b_vector: Optional[SubroutineModel] = None, ): - super().__init__(task_name, requirements) - - if linear_system_block_encoding is not None: - self.linear_system_block_encoding = linear_system_block_encoding - else: - self.linear_system_block_encoding = SubroutineModel( - "linear_system_block_encoding" - ) - - if prepare_b_vector is not None: - self.prepare_b_vector = prepare_b_vector - else: - self.prepare_b_vector = SubroutineModel("prepare_b_vector") - - def set_requirements( - self, - failure_tolerance: float = None, - # condition_number: float = None, - ): - args = locals() - # Clean up the args dictionary before setting requirements - args.pop("self") - args = { - k: v for k, v in args.items() if v is not None and not k.startswith("__") - } - # Initialize the requirements attribute if it doesn't exist - if not hasattr(self, "requirements"): - self.requirements = {} - - # Update the requirements with new values - self.requirements.update(args) - - # Call the parent class's set_requirements method with the updated requirements - super().set_requirements(**self.requirements) + super().__init__(task_name) + self.linear_system_block_encoding = linear_system_block_encoding + self.prepare_b_vector = prepare_b_vector def populate_requirements_for_subroutines(self): # Allocate failure tolerance @@ -80,14 +49,10 @@ def populate_requirements_for_subroutines(self): self.prepare_b_vector.number_of_times_called = n_calls_to_b # Set block encoding requirements - self.linear_system_block_encoding.set_requirements( - failure_tolerance=linear_system_block_encoding_failure_tolerance, - ) - - # Set block encoding requirements - self.prepare_b_vector.set_requirements( - failure_tolerance=prepare_b_vector_failure_tolerance, - ) + LinearSystemBlockEncodingData.failure_tolerance = linear_system_block_encoding_failure_tolerance + StatePreparationOracleData.failure_tolerance=prepare_b_vector_failure_tolerance + self.linear_system_block_encoding.set_requirements(LinearSystemBlockEncodingData) + self.prepare_b_vector.set_requirements(StatePreparationOracleData) def count_qubits(self): # From Theorem 1 in https://arxiv.org/abs/2305.11352 diff --git a/src/qsub/subroutine_model.py b/src/qsub/subroutine_model.py index 3ce44c9..52c42db 100644 --- a/src/qsub/subroutine_model.py +++ b/src/qsub/subroutine_model.py @@ -2,26 +2,26 @@ from graphviz import Digraph from anytree import Node, RenderTree -from anytree.exporter import DotExporter import plotly.graph_objects as go from sympy import symbols +from abc import ABC, abstractmethod +from dataclasses import dataclass, asdict, is_dataclass -class SubroutineModel: - def __init__(self, task_name: str, requirements: Optional[dict] = None, **kwargs): +class SubroutineModel(ABC): + def __init__(self, task_name: str, **kwargs): self.task_name = task_name - self.requirements = requirements or {} + self.requirements = {} + self.inner_subroutine = None self.number_of_times_called: Optional[Union[float, int]] = None for attr, value in kwargs.items(): if isinstance(value, SubroutineModel): setattr(self, attr, value) - - def set_requirements(self, *args, **kwargs): - if args: - raise TypeError( - "The set_requirements method expects keyword arguments of the form argument=value." - ) - self.requirements = kwargs + + def set_requirements(self,requirements:dataclass): + assert is_dataclass(requirements) + self.requirements = asdict(requirements) + assert "failure_tolerance" in self.requirements def populate_requirements_for_subroutines(self): pass @@ -251,3 +251,7 @@ def plot_graph(self): ) fig.show() + + def delegate(self, inner_subroutine): + self.inner_subroutine = inner_subroutine + return self diff --git a/src/qsub/utils.py b/src/qsub/utils.py index 5f756ce..de0e6f0 100644 --- a/src/qsub/utils.py +++ b/src/qsub/utils.py @@ -1,6 +1,3 @@ -# Module of helper functions - - def consume_fraction_of_error_budget(consumed_fraction, current_error_budget): # Takes in the current error budget and the amount to be consumed and outputs the amount consumed and the remaining error budget consumed_error_budget = consumed_fraction * current_error_budget diff --git a/tests/qsub/quantum_algorithms/general_quantum_algorithms/test_amplitude_amplification.py b/tests/qsub/quantum_algorithms/general_quantum_algorithms/test_amplitude_amplification.py index 5879be1..4e309c8 100644 --- a/tests/qsub/quantum_algorithms/general_quantum_algorithms/test_amplitude_amplification.py +++ b/tests/qsub/quantum_algorithms/general_quantum_algorithms/test_amplitude_amplification.py @@ -1,10 +1,40 @@ import pytest -from src.qsub.quantum_algorithms.general_quantum_algorithms.amplitude_amplification import ( +from qsub.quantum_algorithms.general_quantum_algorithms.amplitude_amplification import ( ObliviousAmplitudeAmplification, compute_number_of_grover_iterates_for_obl_amp, SubroutineModel, ) - +from typing import Optional +from dataclasses import dataclass + + +@pytest.fixture() +def state_preparation_oracle(): + class MockSubroutineModel(SubroutineModel): + def __init__(self, task_name: str, **kwargs): + super().__init__(task_name, **kwargs) + def populate_requirements_for_subroutines(self): + return super().populate_requirements_for_subroutines() + + return MockSubroutineModel(task_name="state_prep") + +@pytest.fixture() +def mark_subspace(): + class MockSubroutineModel(SubroutineModel): + def __init__(self, task_name: str, **kwargs): + super().__init__(task_name, **kwargs) + def populate_requirements_for_subroutines(self): + return super().populate_requirements_for_subroutines() + + return MockSubroutineModel(task_name="mark_space") + +@pytest.fixture() +def mock_data_class(): + @dataclass + class SubroutineModelData: + input_state_squared_overlap: float = 0.5 + failure_tolerance: float = 0.03 + return SubroutineModelData() def test_compute_number_of_grover_iterates_for_obl_amp(): # Test with typical parameters @@ -12,36 +42,11 @@ def test_compute_number_of_grover_iterates_for_obl_amp(): assert isinstance(result, float) -def test_obl_amp_initialization(): - # Test initialization without optional parameters - obl_amp = ObliviousAmplitudeAmplification() - assert isinstance(obl_amp.state_preparation_oracle, SubroutineModel) - assert isinstance(obl_amp.mark_subspace, SubroutineModel) - - -def test_obl_amp_initialization_with_parameters(): - # Test initialization with optional parameters - state_prep_oracle = SubroutineModel("state_prep") - mark_subspace = SubroutineModel("mark_space") - obl_amp = ObliviousAmplitudeAmplification( - state_preparation_oracle=state_prep_oracle, mark_subspace=mark_subspace - ) - assert obl_amp.state_preparation_oracle.task_name == "state_prep" - assert obl_amp.mark_subspace.task_name == "mark_space" - - -def test_obl_amp_set_requirements(): - # Test set_requirements method - obl_amp = ObliviousAmplitudeAmplification() - obl_amp.set_requirements(input_state_squared_overlap=0.2, failure_tolerance=0.01) - assert obl_amp.requirements["input_state_squared_overlap"] == 0.2 - assert obl_amp.requirements["failure_tolerance"] == 0.01 - +def test_obl_amp_populate_requirements_for_subroutines(mock_data_class, state_preparation_oracle, mark_subspace): -def test_obl_amp_populate_requirements_for_subroutines(): # Test populate_requirements_for_subroutines method - obl_amp = ObliviousAmplitudeAmplification() - obl_amp.set_requirements(input_state_squared_overlap=0.2, failure_tolerance=0.01) + obl_amp = ObliviousAmplitudeAmplification(state_preparation_oracle, mark_subspace) + obl_amp.set_requirements(mock_data_class) obl_amp.populate_requirements_for_subroutines() # Assertions to check the correct population of requirements assert obl_amp.state_preparation_oracle.number_of_times_called > 0 diff --git a/tests/qsub/quantum_algorithms/general_quantum_algorithms/test_amplitude_estimation.py b/tests/qsub/quantum_algorithms/general_quantum_algorithms/test_amplitude_estimation.py index d904f6f..f58d97f 100644 --- a/tests/qsub/quantum_algorithms/general_quantum_algorithms/test_amplitude_estimation.py +++ b/tests/qsub/quantum_algorithms/general_quantum_algorithms/test_amplitude_estimation.py @@ -1,51 +1,64 @@ import numpy as np -from src.qsub.quantum_algorithms.general_quantum_algorithms.amplitude_estimation import ( +from qsub.quantum_algorithms.general_quantum_algorithms.amplitude_estimation import ( QuantumAmplitudeEstimation, SubroutineModel, + compute_number_of_grover_iterates_for_quantum_amp_est, + compute_number_of_grover_iterates_for_iterative_amp_est ) +import pytest +from dataclasses import dataclass +@pytest.fixture() +def state_preparation_oracle(): + class MockSubroutineModel(SubroutineModel): + def __init__(self, task_name: str, **kwargs): + super().__init__(task_name, **kwargs) + def populate_requirements_for_subroutines(self): + return super().populate_requirements_for_subroutines() + + return MockSubroutineModel(task_name="state_prep") + +@pytest.fixture() +def mark_subspace(): + class MockSubroutineModel(SubroutineModel): + def __init__(self, task_name: str, **kwargs): + super().__init__(task_name, **kwargs) + def populate_requirements_for_subroutines(self): + return super().populate_requirements_for_subroutines() + + return MockSubroutineModel(task_name="mark_space") + +@pytest.fixture() +def mock_data_class(): + @dataclass + class SubroutineModelData: + estimation_error: float = 0.01 + failure_tolerance: float = 0.1 + return SubroutineModelData() # Test the function for computing the number of Grover iterates # TODO: This test uses a function compute_number_of_grover_iterates_for_amp_est which # is not defined anywhere in the codebase. Maybe it was removed or renamed? -def test_compute_number_of_grover_iterates_for_amp_est(): - failure_tolerance = 0.1 - estimation_error = 0.01 - expected_iterates = np.log(1 / failure_tolerance) / estimation_error - actual_iterates = compute_number_of_grover_iterates_for_amp_est( - failure_tolerance, estimation_error - ) - assert actual_iterates == expected_iterates - - -# Test the initialization of QuantumAmplitudeEstimation -def test_quantum_amplitude_estimation_initialization(): - qae = QuantumAmplitudeEstimation() - assert qae.task_name == "estimate_amplitude" - assert isinstance(qae.state_preparation_oracle, SubroutineModel) - assert isinstance(qae.mark_subspace, SubroutineModel) - - -# Test setting requirements in QuantumAmplitudeEstimation -def test_quantum_amplitude_estimation_set_requirements(): - qae = QuantumAmplitudeEstimation() - estimation_error = 0.01 - failure_tolerance = 0.1 - qae.set_requirements( - estimation_error=estimation_error, failure_tolerance=failure_tolerance - ) - assert qae.requirements["estimation_error"] == estimation_error - assert qae.requirements["failure_tolerance"] == failure_tolerance +# Amara: changed it to the one imported TODO: make sure this is the right +# substitution +# def test_compute_number_of_grover_iterates_for_amp_est(): +# failure_tolerance = 0.1 +# estimation_error = 0.01 +# expected_iterates = np.log(1 / failure_tolerance) / estimation_error +# actual_iterates = compute_number_of_grover_iterates_for_iterative_amp_est( +# failure_tolerance, estimation_error +# ) +# assert actual_iterates == expected_iterates # Test populating requirements for subroutines in QuantumAmplitudeEstimation -def test_quantum_amplitude_estimation_populate_requirements_for_subroutines(): - qae = QuantumAmplitudeEstimation() - estimation_error = 0.01 - failure_tolerance = 0.1 - qae.set_requirements( - estimation_error=estimation_error, failure_tolerance=failure_tolerance - ) +def test_quantum_amplitude_estimation_populate_requirements_for_subroutines(mock_data_class, + state_preparation_oracle, + mark_subspace + ): + qae = QuantumAmplitudeEstimation(state_preparation_oracle, mark_subspace) + qae.set_requirements(mock_data_class) + print(qae.requirements) qae.populate_requirements_for_subroutines() assert qae.state_preparation_oracle.requirements["failure_tolerance"] is not None assert qae.mark_subspace.requirements["failure_tolerance"] is not None diff --git a/tests/qsub/quantum_algorithms/general_quantum_algorithms/test_linear_systems.py b/tests/qsub/quantum_algorithms/general_quantum_algorithms/test_linear_systems.py index 364e747..1254311 100644 --- a/tests/qsub/quantum_algorithms/general_quantum_algorithms/test_linear_systems.py +++ b/tests/qsub/quantum_algorithms/general_quantum_algorithms/test_linear_systems.py @@ -1,14 +1,45 @@ import pytest -from src.qsub.subroutine_model import SubroutineModel -from src.qsub.quantum_algorithms.general_quantum_algorithms.linear_systems import ( +from qsub.subroutine_model import SubroutineModel +from qsub.generic_block_encoding import GenericLinearSystemBlockEncoding +from qsub.quantum_algorithms.general_quantum_algorithms.linear_systems import ( TaylorQLSA, get_taylor_qlsa_num_block_encoding_calls, ) +from dataclasses import dataclass +@pytest.fixture() +def linear_system_block_encoding(): + class MockSubroutineModel(GenericLinearSystemBlockEncoding): + def __init__(self, task_name: str, **kwargs): + super().__init__(task_name, **kwargs) + def populate_requirements_for_subroutines(self): + return super().populate_requirements_for_subroutines() + + return MockSubroutineModel(task_name="linear_system_block_encoding") + +@pytest.fixture() +def prepare_b_vector(): + class MockSubroutineModel(SubroutineModel): + def __init__(self, task_name: str, **kwargs): + super().__init__(task_name, **kwargs) + def populate_requirements_for_subroutines(self): + return super().populate_requirements_for_subroutines() + + return MockSubroutineModel(task_name="test") + +@pytest.fixture() +def mock_data_class(): + @dataclass + class SubroutineModelData: + failure_tolerance: float = 0.1 + subnormalization: float = 1 + condition_number: float = 4 + return SubroutineModelData() def test_get_taylor_qlsa_num_block_encoding_calls_normal(): # Test with normal parameters result = get_taylor_qlsa_num_block_encoding_calls(0.1, 1.0, 4.0) + print(result) assert isinstance(result, float) @@ -24,41 +55,19 @@ def test_get_taylor_qlsa_num_block_encoding_calls_invalid_condition_number(): get_taylor_qlsa_num_block_encoding_calls(0.1, 1.0, 2.0) -def test_taylor_qlsa_init_with_block_encoding(): +def test_taylor_qlsa_init_with_block_encoding(linear_system_block_encoding, prepare_b_vector): # Test initialization with linear_system_block_encoding - block_encoding = SubroutineModel("test") - taylor_qlsa = TaylorQLSA(linear_system_block_encoding=block_encoding) - assert taylor_qlsa.linear_system_block_encoding.task_name == "test" - - -def test_taylor_qlsa_init_without_block_encoding(): - # Test initialization without linear_system_block_encoding - taylor_qlsa = TaylorQLSA() - assert ( - taylor_qlsa.linear_system_block_encoding.task_name - == "linear_system_block_encoding" - ) - - -def test_taylor_qlsa_set_requirements(): - # Test set_requirements method - taylor_qlsa = TaylorQLSA() - taylor_qlsa.set_requirements( - failure_tolerance=0.1, subnormalization=1.0, condition_number=4.0 - ) - assert taylor_qlsa.requirements == { - "failure_tolerance": 0.1, - "subnormalization": 1.0, - "condition_number": 4.0, - } + taylor_qlsa = TaylorQLSA(linear_system_block_encoding, prepare_b_vector) + assert taylor_qlsa.linear_system_block_encoding.task_name == "linear_system_block_encoding" -def test_taylor_qlsa_populate_requirements_for_subroutines(): +def test_taylor_qlsa_populate_requirements_for_subroutines(mock_data_class, + linear_system_block_encoding, + prepare_b_vector + ): # Test populate_requirements_for_subroutines method - taylor_qlsa = TaylorQLSA() - taylor_qlsa.set_requirements( - failure_tolerance=0.1, subnormalization=1.0, condition_number=4.0 - ) + taylor_qlsa = TaylorQLSA(linear_system_block_encoding, prepare_b_vector) + taylor_qlsa.set_requirements(mock_data_class) taylor_qlsa.populate_requirements_for_subroutines() # Assertions to check the correct population of requirements assert taylor_qlsa.linear_system_block_encoding.number_of_times_called > 0 diff --git a/tests/qsub/test_subroutine_model.py b/tests/qsub/test_subroutine_model.py index abb977f..593b23a 100644 --- a/tests/qsub/test_subroutine_model.py +++ b/tests/qsub/test_subroutine_model.py @@ -1,41 +1,72 @@ import pytest - +from dataclasses import dataclass from qsub.subroutine_model import SubroutineModel +from typing import Optional -def test_init(): - sub = SubroutineModel(task_name="TestTask") - assert sub.task_name == "TestTask" - assert sub.requirements == {} - assert sub.number_of_times_called is None - - sub_with_reqs = SubroutineModel(task_name="TestTask", requirements={"req1": "val1"}) - assert sub_with_reqs.requirements == {"req1": "val1"} +@pytest.fixture() +def subroutinemodel(): + class MockSubroutineModel(SubroutineModel): + def __init__(self, task_name: str, **kwargs): + super().__init__(task_name, **kwargs) + def populate_requirements_for_subroutines(self): + return super().populate_requirements_for_subroutines() + return MockSubroutineModel(task_name="TestTask") -def test_set_requirements(): - sub = SubroutineModel(task_name="TestTask") - sub.set_requirements(req1="val1") - assert sub.requirements == {"req1": "val1"} +@pytest.fixture() +def child_subroutine_1(): + class MockSubroutineModel(SubroutineModel): + def __init__(self, task_name: str, requirements: Optional[dict] = None, **kwargs): + super().__init__(task_name, requirements, **kwargs) + def populate_requirements_for_subroutines(self): + return super().populate_requirements_for_subroutines() + return MockSubroutineModel(task_name="Task1") - with pytest.raises(TypeError): - sub.set_requirements("invalid_arg") +@pytest.fixture() +def child_subroutine_2(): + class MockSubroutineModel(SubroutineModel): + def __init__(self, task_name: str, requirements: Optional[dict] = None, **kwargs): + super().__init__(task_name, requirements, **kwargs) + def populate_requirements_for_subroutines(self): + return super().populate_requirements_for_subroutines() + return MockSubroutineModel(task_name="Task2") +@pytest.fixture() +def mock_data_class(): + @dataclass + class SubroutineModelData: + field_1: float = 0 + field_2: float = 1 + failure_tolerance: float = 0.01 + return SubroutineModelData() + +def test_initializing_subroutine(subroutinemodel, mock_data_class): + sub = subroutinemodel + sub.set_requirements(mock_data_class) + assert sub.task_name == "TestTask" + assert sub.requirements != {} + assert sub.number_of_times_called is None + assert 'field_1' in sub.requirements.keys() + assert 'field_2' in sub.requirements.keys() -def test_count_subroutines(): - sub = SubroutineModel(task_name="Main") - sub.sub1 = SubroutineModel(task_name="Sub1") - sub.sub2 = SubroutineModel(task_name="Sub2") - +def test_count_subroutines(subroutinemodel, child_subroutine_1 , child_subroutine_2): + sub = subroutinemodel + sub.sub1 = child_subroutine_1 + sub.sub2 = child_subroutine_2 counts = sub.count_subroutines() - assert counts == {"Main": 1, "Sub1": 1, "Sub2": 1} + print("number of counts: ", counts) + assert len(counts) == len({"TestTask": 1, " Task1": 1, " Task2": 1}) + assert counts == {'TestTask': 1, 'Task1': 1, 'Task2': 1} -def test_print_profile(capsys): - sub = SubroutineModel(task_name="Main", requirements={"req1": "val1"}) +def test_print_profile(capsys,subroutinemodel, mock_data_class): + sub = subroutinemodel + sub.set_requirements(mock_data_class) sub.print_profile() captured = capsys.readouterr() - assert "Subroutine: SubroutineModel (Task: Main)" in captured.out + print(captured) + assert "Subroutine: MockSubroutineModel (Task: TestTask)" in captured.out assert "Requirements:" in captured.out - assert "req1: val1" in captured.out + assert "field_1: 0" in captured.out