Code design changes

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,

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
 

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"]

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"]
 

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()