SVD data processing

qiskit_experiments/data_processing/__init__.py

@@ -23,6 +23,7 @@
 
     DataProcessor
     DataAction
+    TrainableDataAction
 
 
 Data Processing Nodes
@@ -33,13 +34,17 @@
     Probability
     ToImag
     ToReal
+    SVD
+    AverageData
 """
 
-from .data_action import DataAction
+from .data_action import DataAction, TrainableDataAction
 from .nodes import (
     Probability,
     ToImag,
     ToReal,
+    SVD,
+    AverageData,
 )
 
 from .data_processor import DataProcessor

qiskit_experiments/data_processing/data_action.py

@@ -13,7 +13,7 @@
 """Defines the steps that can be used to analyse data."""
 
 from abc import ABCMeta, abstractmethod
-from typing import Any
+from typing import Any, List, Optional, Tuple
 
 
 class DataAction(metaclass=ABCMeta):
@@ -30,46 +30,75 @@ def __init__(self, validate: bool = True):
         self._validate = validate
 
     @abstractmethod
-    def _process(self, datum: Any) -> Any:
+    def _process(self, datum: Any, error: Optional[Any] = None) -> Tuple[Any, Any]:
         """
         Applies the data processing step to the datum.
 
         Args:
             datum: A single item of data which will be processed.
+            error: An optional error estimation on the datum that can be further propagated.
 
         Returns:
-            processed data: The data that has been processed.
+            processed data: The data that has been processed along with the propagated error.
         """
 
     @abstractmethod
-    def _format_data(self, datum: Any) -> Any:
-        """
-        Check that the given data has the correct structure. This method may
+    def _format_data(self, datum: Any, error: Optional[Any] = None) -> Tuple[Any, Any]:
+        """Format and validate the input.
+
+        Check that the given data and error has the correct structure. This method may
         additionally change the data type, e.g. converting a list to a numpy array.
 
         Args:
             datum: The data instance to check and format.
+            error: An optional error estimation on the datum to check and format.
 
         Returns:
-            datum: The data that was checked.
+            datum, error: The formatted datum and its optional error.
 
         Raises:
-            DataProcessorError: If the data does not have the proper format.
+            DataProcessorError: If either the data or the error do not have the proper format.
         """
 
-    def __call__(self, data: Any) -> Any:
-        """
-        Call the data action of this node on the data.
+    def __call__(self, data: Any, error: Optional[Any] = None) -> Tuple[Any, Any]:
+        """Call the data action of this node on the data and propagate the error.
 
         Args:
             data: The data to process. The action nodes in the data processor will
                 raise errors if the data does not have the appropriate format.
+            error: An optional error estimation on the datum that can be further processed.
 
         Returns:
-            processed data: The data processed by self.
+            processed data: The data processed by self as a tuple of processed datum and
+                optionally the propagated error estimate.
         """
-        return self._process(self._format_data(data))
+        return self._process(*self._format_data(data, error))
 
     def __repr__(self):
         """String representation of the node."""
         return f"{self.__class__.__name__}(validate={self._validate})"
+
+
+class TrainableDataAction(DataAction):
+    """A base class for data actions that need training."""
+
+    @property
+    @abstractmethod
+    def is_trained(self) -> bool:
+        """Return False if the DataAction needs to be trained.
+
+        Subclasses must implement this property to communicate if they have been trained.
+
+        Return:
+            True if the data action has been trained.
+        """
+
+    @abstractmethod
+    def train(self, data: List[Any]):
+        """Train a DataAction.
+
+        Certain data processing nodes, such as a SVD, require data to first train.
+
+        Args:
+            data: A list of datum. Each datum is a point used to train the node.
+        """

qiskit_experiments/data_processing/data_processor.py

@@ -14,7 +14,7 @@
 
 from typing import Any, Dict, List, Set, Tuple, Union
 
-from qiskit_experiments.data_processing.data_action import DataAction
+from qiskit_experiments.data_processing.data_action import DataAction, TrainableDataAction
 from qiskit_experiments.data_processing.exceptions import DataProcessorError
 
 
@@ -54,23 +54,34 @@ def append(self, node: DataAction):
         """
         self._nodes.append(node)
 
-    def __call__(self, datum: Dict[str, Any]) -> Any:
+    @property
+    def is_trained(self) -> bool:
+        """Return True if all nodes of the data processor have been trained."""
+        for node in self._nodes:
+            if isinstance(node, TrainableDataAction):
+                if not node.is_trained:
+                    return False
+
+        return True
+
+    def __call__(self, datum: Dict[str, Any], **options) -> Tuple[Any, Any]:
         """
         Call self on the given datum. This method sequentially calls the stored data actions
         on the datum.
 
         Args:
             datum: A single item of data, typically from an ExperimentData instance, that needs
                 to be processed. This dict also contains the metadata of each experiment.
+            options: Run-time options given as keyword arguments that will be passed to the nodes.
 
         Returns:
             processed data: The data processed by the data processor.
         """
-        return self._call_internal(datum, False)
+        return self._call_internal(datum, **options)
 
     def call_with_history(
         self, datum: Dict[str, Any], history_nodes: Set = None
-    ) -> Tuple[Any, List]:
+    ) -> Tuple[Any, Any, List]:
         """
         Call self on the given datum. This method sequentially calls the stored data actions
         on the datum and also returns the history of the processed data.
@@ -89,10 +100,13 @@ def call_with_history(
         return self._call_internal(datum, True, history_nodes)
 
     def _call_internal(
-        self, datum: Dict[str, Any], with_history: bool, history_nodes: Set = None
-    ) -> Union[Any, Tuple[Any, List]]:
-        """
-        Internal function to process the data with or with storing the history of the computation.
+        self,
+        datum: Dict[str, Any],
+        with_history: bool = False,
+        history_nodes: Set = None,
+        call_up_to_node: int = None,
+    ) -> Union[Tuple[Any, Any], Tuple[Any, Any, List]]:
+        """Process the data with or without storing the history of the computation.
 
         Args:
             datum: A single item of data, typically from an ExperimentData instance, that
@@ -101,31 +115,56 @@ def _call_internal(
             history_nodes: The nodes, specified by index in the data processing chain, to
                 include in the history. If None is given then all nodes will be included
                 in the history.
+            call_up_to_node: The data processor will use each node in the processing chain
+                up to the node indexed by call_up_to_node. If this variable is not specified
+                then all nodes in the data processing chain will be called.
 
         Returns:
             datum_ and history if with_history is True or datum_ if with_history is False.
 
         Raises:
             DataProcessorError: If the input key of the data processor is not contained in datum.
         """
+        if call_up_to_node is None:
+            call_up_to_node = len(self._nodes)
 
         if self._input_key not in datum:
             raise DataProcessorError(
                 f"The input key {self._input_key} was not found in the input datum."
             )
 
         datum_ = datum[self._input_key]
+        error_ = None
 
         history = []
         for index, node in enumerate(self._nodes):
-            datum_ = node(datum_)
 
-            if with_history and (
-                history_nodes is None or (history_nodes and index in history_nodes)
-            ):
-                history.append((node.__class__.__name__, datum_, index))
+            if index < call_up_to_node:
+                datum_, error_ = node(datum_, error_)
+
+                if with_history and (
+                    history_nodes is None or (history_nodes and index in history_nodes)
+                ):
+                    history.append((node.__class__.__name__, datum_, error_, index))
 
         if with_history:
-            return datum_, history
+            return datum_, error_, history
         else:
-            return datum_
+            return datum_, error_
+
+    def train(self, data: List[Dict[str, Any]]):
+        """Train the nodes of the data processor.
+
+        Args:
+            data: The data to use to train the data processor.
+        """
+
+        for index, node in enumerate(self._nodes):
+            if isinstance(node, TrainableDataAction):
+                if not node.is_trained:
+                    # Process the data up to the untrained node.
+                    train_data = []
+                    for datum in data:
+                        train_data.append(self._call_internal(datum, call_up_to_node=index)[0])
+
+                    node.train(train_data)