Added partition function for stabilizer index kernel partitioning

src/qce_circuit/structure/acquisition_indexing/factory_stabilizer_index_kernel.py

@@ -0,0 +1,47 @@
+# -------------------------------------------
+# Module describing factory construction of index kernels.
+# -------------------------------------------
+from typing import List
+from qce_circuit.structure.acquisition_indexing.intrf_stabilizer_index_kernel import IStabilizerIndexingKernel
+from qce_circuit.structure.acquisition_indexing.intrf_index_strategy import FixedIndexStrategy
+
+
+class KernelPartitioner:
+    """
+    Behaviour class, for slicing/partitioning indexing kernels.
+    """
+
+    # region Static Class Methods
+    @staticmethod
+    def partition_in_equal_sections(index_kernel: IStabilizerIndexingKernel, sections: int) -> List[IStabilizerIndexingKernel]:
+        """
+        Creates a list of subset kernels based on a self.
+        :param index_kernel: The kernel describing the structure of a single subset (cycle).
+        :param sections: The number of subsets (partitions) to generate.
+        :return: List of independent kernel objects, each pointing to a unique subset of the data.
+        """
+        subsets: List[IStabilizerIndexingKernel] = []
+
+        # Calculate the cycle length of the template.
+        # We assume the template is 'floating', but we need its length.
+        # To ensure calculation is correct, we temporarily bind it to 0 to measure length.
+        offset_strategy = FixedIndexStrategy(index=index_kernel.start_index)
+        experiment_repetitions: int = index_kernel.experiment_repetitions
+        subset_repetitions: int = experiment_repetitions // sections
+
+        for i in range(sections):
+            # Calculate absolute start index for this subset
+            subset_start_index = offset_strategy.index + (i * subset_repetitions * index_kernel.kernel_cycle_length)
+
+            # Create strategy for this specific window
+            strategy = FixedIndexStrategy(index=subset_start_index)
+
+            # Create the subset kernel using the interface method
+            subset_kernel = index_kernel.with_index_strategy(
+                strategy=strategy,
+                experiment_repetitions=subset_repetitions,
+            )
+            subsets.append(subset_kernel)
+
+        return subsets
+    # endregion

src/qce_circuit/structure/acquisition_indexing/intrf_stabilizer_index_kernel.py

@@ -7,6 +7,7 @@
 from enum import Enum, unique
 from qce_circuit.utilities.custom_exceptions import InterfaceMethodException
 from qce_circuit.structure.acquisition_indexing.intrf_index_kernel import IIndexingKernel
+from qce_circuit.structure.acquisition_indexing.intrf_index_strategy import IIndexStrategy
 from qce_circuit.connectivity.intrf_channel_identifier import IQubitID
 
 
@@ -100,4 +101,16 @@ def get_projected_calibration_acquisition_indices(self, qubit_id: IQubitID, stat
         :return: Tensor of indices pointing at all projection acquisition within calibration points.
         """
         raise InterfaceMethodException
+
+    @abstractmethod
+    def with_index_strategy(self, strategy: 'IIndexStrategy', experiment_repetitions: int) -> 'IStabilizerIndexingKernel':
+        """
+        Constructs a copy of self, but with a modified index strategy.
+        Useful for creating subset kernels (windowing) or creating repetitions.
+
+        :param strategy: The new strategy determining the start_index.
+        :param experiment_repetitions: Number of experimental repetitions to point to.
+        :return: A new instance of IIndexingKernel.
+        """
+        raise InterfaceMethodException
     # endregion

src/qce_circuit/structure/acquisition_indexing/kernel_repetition_code.py

@@ -3,7 +3,7 @@
 # Specializes repetition-code kernel implementations.
 # -------------------------------------------
 from dataclasses import dataclass, field
-from typing import List
+from typing import List, Optional
 import warnings
 import numpy as np
 from numpy.typing import NDArray
@@ -174,7 +174,7 @@ def indexing_kernels(self) -> List[IIndexingKernel]:
     # endregion
 
     # region Class Constructor
-    def __init__(self, rounds: List[int], heralded_initialization: bool, qutrit_calibration_points: bool, involved_data_qubit_ids: List[IQubitID], involved_ancilla_qubit_ids: List[IQubitID], experiment_repetitions: int):
+    def __init__(self, rounds: List[int], heralded_initialization: bool, qutrit_calibration_points: bool, involved_data_qubit_ids: List[IQubitID], involved_ancilla_qubit_ids: List[IQubitID], experiment_repetitions: int, index_offset_strategy: Optional[IIndexStrategy] = None):
         self._rounds: List[int] = rounds
         """Array-like of integers corresponding to number of repetitions. Each integer element represents a separate RepetitionIndexKernel."""
         self._heralded_initialization: bool = heralded_initialization
@@ -186,11 +186,14 @@ def __init__(self, rounds: List[int], heralded_initialization: bool, qutrit_cali
         """Array-like of involved data- and ancilla-qubit-ID's."""
         self._repetitions: int = experiment_repetitions
         """Total length of data indices, including all experiment repetitions. Size of dataset."""
+        self._index_offset_strategy: IIndexStrategy = index_offset_strategy if index_offset_strategy is not None else FixedIndexStrategy(index=0)
+
         self._repetition_kernels: List[RepetitionIndexKernel] = []
-        for nr_round in self._rounds:
-            # Uses this index as excluded start to count forward.
-            offset_strategy: IIndexStrategy = FixedIndexStrategy(index=0)
-            if self._repetition_kernels:  # Not empty
+        for i, nr_round in enumerate(self._rounds):
+            # Default to fixed strategy for the first kernel if none is provided, or chain relative to the previous
+            if i == 0:
+                offset_strategy: IIndexStrategy = self._index_offset_strategy
+            else:  # if self._repetition_kernels:  # Not empty
                 offset_strategy = RelativeIndexStrategy(reference_index_kernel=self._repetition_kernels[-1])
             # Append indexing kernel
             kernel: RepetitionIndexKernel = RepetitionIndexKernel(
@@ -201,9 +204,14 @@ def __init__(self, rounds: List[int], heralded_initialization: bool, qutrit_cali
                 involved_ancilla_qubit_ids=self._involved_ancilla_ids,
             )
             self._repetition_kernels.append(kernel)
+
+        # Calibration kernel follows the last repetition kernel
+        reference_kernel = self._repetition_kernels[-1] if self._repetition_kernels else None
+        calib_strategy = RelativeIndexStrategy(reference_index_kernel=reference_kernel) if reference_kernel else self._index_offset_strategy
+
         self._calibration_kernel: QutritCalibrationIndexKernel = QutritCalibrationIndexKernel(
             heralded_initialization=self._heralded_initialization,
-            index_offset_strategy=RelativeIndexStrategy(reference_index_kernel=self._repetition_kernels[-1]),
+            index_offset_strategy=calib_strategy,
             involved_qubit_ids=self._involved_data_ids + self._involved_ancilla_ids,
         )
     # endregion
@@ -213,6 +221,25 @@ def contains(self, element: IQubitID) -> List[int]:
         """:return: Array-like of measurement indices corresponding to element within this indexing kernel."""
         raise NotImplemented
 
+    def with_index_strategy(self, strategy: IIndexStrategy, experiment_repetitions: int) -> 'RepetitionExperimentKernel':
+        """
+        Constructs a copy of self, but with a modified index strategy.
+        Useful for creating subset kernels (windowing) or creating repetitions.
+
+        :param strategy: The new strategy determining the start_index.
+        :param experiment_repetitions: Number of experimental repetitions to point to.
+        :return: A new instance of IIndexingKernel.
+        """
+        return RepetitionExperimentKernel(
+            rounds=self._rounds,
+            heralded_initialization=self._heralded_initialization,
+            qutrit_calibration_points=self._qutrit_calibration_points,
+            involved_data_qubit_ids=self._involved_data_ids,
+            involved_ancilla_qubit_ids=self._involved_ancilla_ids,
+            experiment_repetitions=experiment_repetitions,
+            index_offset_strategy=strategy
+        )
+
     def get_projected_calibration_acquisition_indices(self, qubit_id: IQubitID, state: StateKey) -> NDArray[np.int_]:
         """
         :param qubit_id: Identifier to which these acquisition indices correspond to.
@@ -392,6 +419,7 @@ def __init__(
             involved_data_qubit_ids: List[IQubitID],
             involved_ancilla_qubit_ids: List[IQubitID],
             experiment_repetitions: int,
+            index_offset_strategy: Optional[IIndexStrategy] = None,
     ):
         self._rounds: List[int] = rounds
         """Array-like of integers corresponding to number of repetitions. Each integer element represents a separate RepetitionIndexKernel."""
@@ -400,11 +428,14 @@ def __init__(
         """Array-like of involved data- and ancilla-qubit-ID's."""
         self._repetitions: int = experiment_repetitions
         """Total length of data indices, including all experiment repetitions. Size of dataset."""
+        self._index_offset_strategy: IIndexStrategy = index_offset_strategy if index_offset_strategy is not None else FixedIndexStrategy(index=0)
+
         self._repetition_kernels: List[RepetitionIndexKernel] = []
-        for nr_round in self._rounds:
-            # Uses this index as excluded start to count forward.
-            offset_strategy: IIndexStrategy = FixedIndexStrategy(index=0)
-            if self._repetition_kernels:  # Not empty
+        for i, nr_round in enumerate(self._rounds):
+            # Default to fixed strategy for the first kernel if none is provided, or chain relative to the previous
+            if i == 0:
+                offset_strategy: IIndexStrategy = self._index_offset_strategy
+            else:  # if self._repetition_kernels:  # Not empty
                 offset_strategy = RelativeIndexStrategy(reference_index_kernel=self._repetition_kernels[-1])
             # Append indexing kernel
             kernel: RepetitionIndexKernel = RepetitionIndexKernel(
@@ -423,6 +454,23 @@ def contains(self, element: IQubitID) -> List[int]:
         """:return: Array-like of measurement indices corresponding to element within this indexing kernel."""
         raise NotImplemented
 
+    def with_index_strategy(self, strategy: IIndexStrategy, experiment_repetitions: int) -> 'SimulatedRepetitionExperimentKernel':
+        """
+        Constructs a copy of self, but with a modified index strategy.
+        Useful for creating subset kernels (windowing) or creating repetitions.
+
+        :param strategy: The new strategy determining the start_index.
+        :param experiment_repetitions: Number of experimental repetitions to point to.
+        :return: A new instance of IIndexingKernel.
+        """
+        return SimulatedRepetitionExperimentKernel(
+            rounds=self._rounds,
+            involved_data_qubit_ids=self._involved_data_ids,
+            involved_ancilla_qubit_ids=self._involved_ancilla_ids,
+            experiment_repetitions=experiment_repetitions,
+            index_offset_strategy=strategy
+        )
+
     def get_projected_calibration_acquisition_indices(self, qubit_id: IQubitID, state: StateKey) -> NDArray[np.int_]:
         """
         :param qubit_id: Identifier to which these acquisition indices correspond to.