diff --git a/src/qce_circuit/structure/acquisition_indexing/factory_stabilizer_index_kernel.py b/src/qce_circuit/structure/acquisition_indexing/factory_stabilizer_index_kernel.py new file mode 100644 index 0000000..1930c6a --- /dev/null +++ b/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 diff --git a/src/qce_circuit/structure/acquisition_indexing/intrf_stabilizer_index_kernel.py b/src/qce_circuit/structure/acquisition_indexing/intrf_stabilizer_index_kernel.py index 492b8c6..e93b010 100644 --- a/src/qce_circuit/structure/acquisition_indexing/intrf_stabilizer_index_kernel.py +++ b/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 diff --git a/src/qce_circuit/structure/acquisition_indexing/kernel_repetition_code.py b/src/qce_circuit/structure/acquisition_indexing/kernel_repetition_code.py index b90507e..4d563b6 100644 --- a/src/qce_circuit/structure/acquisition_indexing/kernel_repetition_code.py +++ b/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.