Implementation of QGAN and some minor updates in QML module

.settings/requirements_full.txt

@@ -12,8 +12,8 @@ amazon-braket-sdk==1.35.1
 scipy==1.11.1
 botocore==1.25.7
 boto3==1.22.7
-pennylane==0.28.0
-pennylane-lightning==0.28.0
+pennylane==0.36.0
+pennylane-lightning==0.36.0
 amazon-braket-pennylane-plugin==1.5.2
 dwave-samplers==1.0.0
 nnf==0.4.1
@@ -25,11 +25,15 @@ pyqubo==1.4.0
 dwave_networkx==0.8.13
 qiskit==0.45.0
 pulp==2.7.0
-pandas==2.1.0
+pandas==1.5.2
 openpyxl==3.1.2
 pulser==0.16.0
-qiskit-ibmq-provider==0.19.2
 cma==3.3.0
+matplotlib==3.7.5
 tensorboard==2.13.0
 tensorboardX==2.6.2
+torch==2.0.1
+jax==0.4.30
+jaxlib==0.4.30
 qiskit_aer==0.11.2
+qiskit-ibmq-provider==0.19.2

src/modules/applications/QML/generative_modeling/GenerativeModeling.py

@@ -42,7 +42,7 @@ def get_requirements() -> list[dict]:
         """
         Returns requirements of this module
 
-        :return: list of dict with requirements of this module
+        :return: list of dicts with requirements of this module
         :rtype: list[dict]
         """
         return []
@@ -81,53 +81,51 @@ def get_parameter_options(self) -> dict:
             }
         }
 
-    def generate_problem(self, config) -> dict:
+    def generate_problem(self, config: dict) -> dict:
 
         """
         The number of qubits is chosen for this problem.
 
-        :param config:
+        :param config: dictionary including the number of qubits
         :type config: dict
-        :return: n_qubits
+        :return: dictionary with the number of qubits
         :rtype: dict
         """
 
         application_config = {"n_qubits": config["n_qubits"]}
         return application_config
 
-    def preprocess(self, input_data: dict, config: dict, **kwargs):
+    def preprocess(self, input_data: dict, config: dict, **kwargs: dict) -> tuple[dict, float]:
         """
         Generate the actual problem instance in the preprocess function.
         :param input_data: Usually not used for this method.
         :type input_data: dict
-
         :param config: config for the problem creation.
-        :param config: 
-
+        :type config: dict
         :param kwargs: Optional additional arguments
         :type kwargs: dict
-
         :param kwargs: optional additional arguments.
 
-        :return:
+        :return: tuple containing qubit number and the function's computation time
+        :rtype: tuple[dict, float]
         """
         start = start_time_measurement()
         output = self.generate_problem(config)
         output["store_dir_iter"] = f"{kwargs['store_dir']}/rep_{kwargs['rep_count']}"
         return output, end_time_measurement(start)
 
-    def postprocess(self, input_data: dict, config: dict, **kwargs):
+    def postprocess(self, input_data: dict, config: dict, **kwargs: dict) -> tuple[dict, float]:
         """
-        Process the solution here, then validate and evluate it.
+        Process the solution here, then validate and evaluate it.
 
-        :param input_data: A representation of the quntum machine learning model that will be trained
+        :param input_data: A representation of the quantum machine learning model that will be trained
         :type input_data: dict
         :param config: Config specifying the parameters of the training
         :type config: dict
         :param kwargs: optional keyword arguments
         :type kwargs: dict
-        :return: tuple with same dictionary like input_data and the time it
-        :rtype: (dict, float)
+        :return: tuple with input_data and the function's computation time
+        :rtype: tuple[dict, float]
         """
 
         start = start_time_measurement()

src/modules/applications/QML/generative_modeling/data/data_handler/ContinuousData.py

@@ -81,6 +81,11 @@ def get_parameter_options(self) -> dict:
                             "data_set": {
                                 "values": ["X_2D", "O_2D", "MG_2D", "Stocks_2D"],
                                 "description": "Which dataset do you want to use?"
+                            },
+
+                            "train_size": {
+                                "values": [0.1, 0.3, 0.5, 0.7, 1.0],
+                                "description": "What percentage of the dataset do you want to use for training?"
                             }
                         }
 
@@ -89,7 +94,13 @@ def get_parameter_options(self) -> dict:
             "data_set": {
                 "values": ["X_2D", "O_2D", "MG_2D", "Stocks_2D"],
                 "description": "Which dataset do you want to use?"
+            },
+
+            "train_size": {
+                "values": [0.1, 0.3, 0.5, 0.7, 1.0],
+                "description": "What percentage of the dataset do you want to use for training?"
             }
+
         }
 
     class Config(TypedDict):
@@ -103,6 +114,7 @@ class Config(TypedDict):
         """
 
         data_set: int
+        train_size: float
 
     def data_load(self, gen_mod: dict, config: Config) -> dict:
 
@@ -111,10 +123,10 @@ def data_load(self, gen_mod: dict, config: Config) -> dict:
 
         :param gen_mod: Dictionary with collected information of the previous modules
         :type gen_mod: dict
-        :param config: Config specifying the paramters of the data handler
-        :type config: dict
-        :return: Must always return the mapped problem and the time it took to create the mapping
-        :rtype: tuple(any, float)
+        :param config: Config specifying the parameters of the data handler
+        :type config: Config
+        :return: dictionary including the mapped problem
+        :rtype: dict
         """
         self.dataset_name = config["data_set"]
         self.n_qubits = gen_mod["n_qubits"]
@@ -127,19 +139,20 @@ def data_load(self, gen_mod: dict, config: Config) -> dict:
             "dataset_name": self.dataset_name,
             "n_qubits": self.n_qubits,
             "dataset": self.dataset,
+            "train_size": config["train_size"],
             "store_dir_iter": gen_mod["store_dir_iter"]}
 
         return application_config
 
-    def evaluate(self, solution: list, **kwargs) -> (float, float):
+    def evaluate(self, solution: dict) -> tuple[float, float]:
         """
         Calculate KL in original space.
 
-        :param solution: A dictionary-like object containing the solution data, including histogram_generated_original
-                         and histogram_train_original.
-        :type solution: list
-        :return: KL for the generated samples and the time it took to calculate it.
-        :rtype: tuple(float, float)
+        :param solution: a dictionary containing the solution data, including histogram_generated_original
+                         and histogram_train_original
+        :type solution: dict
+        :return: Kullback-Leibler (KL) divergence for the generated samples and the time it took to calculate it
+        :rtype: tuple[float, float]
         """
         start = start_time_measurement()
 
@@ -149,16 +162,23 @@ def evaluate(self, solution: list, **kwargs) -> (float, float):
         histogram_train_original = solution["histogram_train_original"]
         histogram_train_original[histogram_train_original == 0] = 1e-8
 
-        # Flatten the arrays for efficient computation
         target = histogram_train_original.ravel()
         generated = generated.ravel()
-
-        # Compute KL divergence using NumPy vectorized operations
         kl_divergence = self.kl_divergence(target, generated)
 
         logging.info(f"KL original space: {kl_divergence}")
 
         return kl_divergence, end_time_measurement(start)
 
-    def kl_divergence(self, target, q):
+    def kl_divergence(self, target: np.ndarray, q: np.ndarray) -> float:
+        """
+        Function to calculate KL divergence
+
+        :param target: Probability mass function of the target distribution
+        :type target: np.ndarray
+        :param q: Probability mass function generated by the quantum circuit
+        :type q: np.ndarray
+        :return: Kullback-Leibler divergence
+        :rtype: float
+        """
         return np.sum(target * np.log(target / q))

src/modules/applications/QML/generative_modeling/data/data_handler/DataHandler.py

@@ -61,7 +61,7 @@ def get_requirements() -> list[dict]:
             }
         ]
 
-    def preprocess(self, input_data: dict, config: dict, **kwargs):
+    def preprocess(self, input_data: dict, config: dict, **kwargs) -> tuple[dict, float]:
         """
         In this module, the preprocessing step is transforming the data to the correct target format.
 
@@ -72,7 +72,7 @@ def preprocess(self, input_data: dict, config: dict, **kwargs):
         :param kwargs: optional additional settings
         :type kwargs: dict
         :return: tuple with transformed problem and the time it took to map it
-        :rtype: (dict, float)
+        :rtype: tuple[dict, float]
         """
         start = start_time_measurement()
         output = self.data_load(input_data, config)
@@ -82,7 +82,7 @@ def preprocess(self, input_data: dict, config: dict, **kwargs):
 
         return output, end_time_measurement(start)
 
-    def postprocess(self, input_data: dict, config: dict, **kwargs):
+    def postprocess(self, input_data: dict, config: dict, **kwargs) -> tuple[dict, float]:
         """
         In this module, the postprocessing step is transforming the data to the correct target format.
 
@@ -93,7 +93,7 @@ def postprocess(self, input_data: dict, config: dict, **kwargs):
         :param kwargs: optional additional settings
         :type kwargs: dict
         :return: tuple with an output_dictionary and the time it took
-        :rtype: (dict, float)
+        :rtype: tuple[dict, float]
         """
         start = start_time_measurement()
         store_dir_iter = input_data["store_dir_iter"]
@@ -120,7 +120,7 @@ def postprocess(self, input_data: dict, config: dict, **kwargs):
 
         # Save metrics per iteration
         if "inference" not in input_data.keys():
-            DataHandler.tb_to_pd(logdir=store_dir_iter, rep=kwargs['rep_count'])
+            DataHandler.tb_to_pd(logdir=store_dir_iter, rep=str(kwargs['rep_count']))
             self.metrics.add_metric_batch(
                 {"metrics_pandas": os.path.relpath(f"{store_dir_iter}/data.pkl", current_directory)})
             if self.generalization_mark is not None:
@@ -148,6 +148,7 @@ def postprocess(self, input_data: dict, config: dict, **kwargs):
                 f"{store_dir_iter}/best_parameters_{kwargs['rep_count']}.npy", current_directory)})
 
             # Save training results
+            input_data.pop("circuit_transpiled")
             with open(f"{store_dir_iter}/training_results-{kwargs['rep_count']}.pkl", 'wb') as f:
                 pickle.dump(input_data, f)
 
@@ -167,7 +168,7 @@ def postprocess(self, input_data: dict, config: dict, **kwargs):
         return input_data, end_time_measurement(start)
 
     @abstractmethod
-    def data_load(self, gen_mod: dict, config: dict) -> dict:
+    def data_load(self, gen_mod: dict, config: dict) -> tuple[any, float]:
         """
         Helps to ensure that the model can effectively learn the underlying
         patterns and structure of the data, and produce high-quality outputs.
@@ -177,18 +178,16 @@ def data_load(self, gen_mod: dict, config: dict) -> dict:
         :param config: config specifying the parameters of the data handler
         :type config: dict
         :return: mapped problem and the time it took to create the mapping
-        :rtype: tuple(any, float)
+        :rtype: tuple[any, float]
         """
         pass
 
-    def generalisation(self) -> (dict, float):
+    def generalisation(self) -> tuple[dict, float]:
         """
         Compute generalisation metrics
 
-        :param solution:
-        :type solution: any
         :return: Evaluation and the time it took to create it
-        :rtype: tuple(any, float)
+        :rtype: tuple[dict, float]
 
         """
         # Compute your metrics here
@@ -197,17 +196,17 @@ def generalisation(self) -> (dict, float):
         return metrics, time_taken
 
     @abstractmethod
-    def evaluate(self, solution: any) -> (dict, float):
+    def evaluate(self, solution: any) -> tuple[any, float]:
         """
-        Compute best loss values.
+        Compute the best loss values.
 
-        :param solution:
+        :param solution: solution data
         :type solution: any
-        :return: bool and the time it took to create it
-        :rtype: tuple(bool, float)
+        :return: evaluation data and the time it took to create it
+        :rtype: tuple[any, float]
 
         """
-        pass
+        return None, 0.0
 
     @staticmethod
     def tb_to_pd(logdir: str, rep: str) -> None:
@@ -217,11 +216,13 @@ def tb_to_pd(logdir: str, rep: str) -> None:
 
         :param logdir: path to the log directory containing TensorBoard event files
         :type logdir: str
-
+        :param rep: repetition counter
+        :type rep: str
         """
         event_acc = EventAccumulator(logdir)
         event_acc.Reload()
         tags = event_acc.Tags()
+        data = []
         tag_data = {}
         for tag in tags['scalars']:
             data = event_acc.Scalars(tag)