labcore.protocols.base

Functions

select_platform(platform)	Select the hardware platform for subsequent protocol execution.
serialize_fit_params(params)

Classes

BranchBase([name])	A named sequence of operations that can be executed.
CheckResult(name, passed, description)	Result of a single named check in evaluate().
Condition(condition, true_branch, false_branch)	A conditional decision point that routes execution to different branches.
Correction()	Base class for stateful correction strategies.
CorrectionParameter(name, params, description)	Platform-aware correction strategy knob.
EvaluateResult(status, checks)	Return type for ProtocolOperation.evaluate() and correct().
OperationStatus(*values)	Return status for ProtocolOperation.evaluate()
ParamImprovement(old_value, new_value, param)
PlatformTypes(*values)
ProtocolBase([report_path])
ProtocolOperation()
ProtocolParameterBase(name, params, description)	Base class for protocol parameters with platform-specific getter/setter methods.
SuperOperationBase()	A composite operation that groups multiple operations together.

class labcore.protocols.base.BranchBase(name: str = 'Branch')

Bases: object

A named sequence of operations that can be executed.

Branches can contain: - Operations (ProtocolOperation instances) - Conditions (conditional routing to other branches)

Parameters:

name – Display name for this branch

Example

>>> main_branch = BranchBase("MainCalibration")
>>> main_branch.append(ResonatorSpectroscopy(params))
>>> main_branch.append(SaturationSpectroscopy(params))

append(item: ProtocolOperation | Condition) → BranchBase

Add an operation or condition to this branch

extend(items: list[ProtocolOperation | Condition]) → BranchBase

Add multiple operations/conditions to this branch

class labcore.protocols.base.CheckResult(name: str, passed: bool, description: str)

Bases: object

Result of a single named check in evaluate().

description: str

name: str

passed: bool

class labcore.protocols.base.Condition(condition: Callable[[], bool], true_branch: BranchBase, false_branch: BranchBase, name: str = 'Condition')

Bases: object

A conditional decision point that routes execution to different branches.

During protocol execution, the condition is evaluated and either true_branch or false_branch is executed.

Parameters:

• condition – Callable returning bool to determine which branch

• true_branch – Branch to execute if condition is True

• false_branch – Branch to execute if condition is False

• name – Optional name for this condition (for logging/reporting)

Example

>>> res_snr = params['resonator']['snr']
>>>
>>> high_snr_branch = BranchBase("HighSNR")
>>> high_snr_branch.append(PiSpectroscopy(params))
>>>
>>> low_snr_branch = BranchBase("LowSNR")
>>> low_snr_branch.append(PowerRabi(params))
>>> low_snr_branch.append(PiSpectroscopy(params))
>>>
>>> condition = Condition(
>>>     condition=lambda: res_snr() > 5.0,
>>>     true_branch=high_snr_branch,
>>>     false_branch=low_snr_branch,
>>>     name="SNR Check"
>>> )

condition_result: bool | None

evaluate() → BranchBase

Evaluate condition and return the branch to execute

report_output: list[str]

taken_branch: BranchBase | None

class labcore.protocols.base.Correction

Bases: object

Base class for stateful correction strategies.

Subclass this and attach an instance to the operation in __init__. The instance persists across retry attempts so that stateful strategies (e.g. stepping through a list of frequency windows) work correctly.

Example:

class FrequencySweepCorrection(Correction):
    name = "scan_next_frequency_window"
    description = "Step through candidate frequency windows"
    triggered_by = "peak_exists"

    def __init__(self, freq_param, windows: list[float]):
        self.freq_param = freq_param
        self.windows = windows
        self._idx = 0

    def can_apply(self) -> bool:
        return self._idx < len(self.windows)

    def apply(self) -> None:
        self.freq_param(self.windows[self._idx])
        self._idx += 1

apply() → None

Apply the correction. Called before the next retry attempt.

can_apply() → bool

Return False when this strategy is exhausted, forcing FAILURE.

description: str = ''

name: str = ''

report_output() → str

Return a string describing what apply() just changed. Optional.

triggered_by: str = ''

class labcore.protocols.base.CorrectionParameter(name: str, params: Any, description: str, platform_type: PlatformTypes | None = None)

Bases: ProtocolParameterBase

Platform-aware correction strategy knob.

Use this instead of ProtocolParameterBase for parameters that control correction strategies (e.g. window size, step count, noise tolerance). May have platform-specific units or scaling — implement the platform getter/setter methods for any unit conversions needed.

Subclass exactly like ProtocolParameterBase.

is_correction: ClassVar[bool] = True

class labcore.protocols.base.EvaluateResult(status: OperationStatus, checks: list[CheckResult] = <factory>)

Bases: object

Return type for ProtocolOperation.evaluate() and correct().

checks: list[CheckResult]

status: OperationStatus

class labcore.protocols.base.OperationStatus(*values)

Bases: Enum

Return status for ProtocolOperation.evaluate()

Indicates what the protocol executor should do next with this operation.

FAILURE = 'failure'

RETRY = 'retry'

SUCCESS = 'success'

class labcore.protocols.base.ParamImprovement(old_value: 'Any', new_value: 'Any', param: 'ProtocolParameterBase')

Bases: object

new_value: Any

old_value: Any

param: ProtocolParameterBase

class labcore.protocols.base.PlatformTypes(*values)

Bases: Enum

DUMMY = 3

OPX = 1

QICK = 2

class labcore.protocols.base.ProtocolBase(report_path: Path = PosixPath('.'))

Bases: object

execute() → None

Execute protocol by recursively executing branches

root_branch: BranchBase | None

success: bool | None

verify_all_parameters() → bool

Verify parameters in all operations across all branches

class labcore.protocols.base.ProtocolOperation

Bases: object

DEFAULT_MAX_ATTEMPTS = 100

analyze() → None

condition: str

correct(result: EvaluateResult) → EvaluateResult

Apply parameter changes based on the evaluation result.

This is the only place where parameter values should be modified.

Default implementation: - Appends a check summary table to the report. - On RETRY: applies registered corrections for each failed check.

If any correction’s can_apply() returns False, escalates to FAILURE.

• On SUCCESS / FAILURE: no-op (override to apply found values on SUCCESS).

Override to apply found values on SUCCESS or implement complex correction logic. Call super().correct(result) first to get check reporting and default RETRY correction handling, then handle SUCCESS.

Parameters:

result – The EvaluateResult returned by evaluate().

Returns:

Possibly modified EvaluateResult (e.g. RETRY escalated to FAILURE when a correction is exhausted).

correction_params: dict[str, CorrectionParameter]

current_attempt: int

data_loc: Path | None

dependents: dict[str, ArrayLike]

evaluate() → EvaluateResult

Assess operation results and return named check outcomes.

Default implementation runs all checks registered via _register_check(). Returns SUCCESS if all checks pass, RETRY if any fail.

Override for complex logic (conditional checks, custom status rules). Overrides should still return an EvaluateResult with CheckResult objects so that reports and the default correct() work correctly.

Returns:

EvaluateResult containing the status and individual check outcomes.

execute() → EvaluateResult

Execute the full operation workflow: measure -> load_data -> analyze -> evaluate -> correct.

Returns:

EvaluateResult from correct(), which may differ from evaluate()’s result (e.g. RETRY escalated to FAILURE when a correction is exhausted).

figure_paths: list[Path]

improvements: list[ParamImprovement]

independents: dict[str, ArrayLike]

input_params: dict[str, ProtocolParameterBase]

load_data() → bool

max_attempts: int

measure() → Path

output_params: dict[str, ProtocolParameterBase]

platform_type: PlatformTypes | None

report_output: list[str | Path]

total_attempts_made: int

class labcore.protocols.base.ProtocolParameterBase(name: str, params: Any, description: str, platform_type: PlatformTypes | None = None)

Bases: object

Base class for protocol parameters with platform-specific getter/setter methods.

Subclasses must implement the platform-specific getter/setter methods: - _qick_getter() / _qick_setter(value) for QICK platform - _opx_getter() / _opx_setter(value) for OPX platform - _dummy_getter() / _dummy_setter(value) for DUMMY platform

These methods should directly call the QCoDeS-style parameter with: - Get: my_param() - Set: my_param(value)

Example

>>> @dataclass
>>> class QubitFrequency(ProtocolParameterBase):
...     name: str = field(default="Qubit IF", init=False)
...     description: str = field(default="Qubit intermediate frequency", init=False)
...
...     def _qick_getter(self):
...         return self.params.qubit.f_ge()
...
...     def _qick_setter(self, value):
...         return self.params.qubit.f_ge(value)
...
...     def _opx_getter(self):
...         return self.params.qubit.frequency()
...
...     def _opx_setter(self, value):
...         return self.params.qubit.frequency(value)

description: str

name: str

params: Any

platform_type: PlatformTypes | None = None

class labcore.protocols.base.SuperOperationBase

Bases: ProtocolOperation

A composite operation that groups multiple operations together.

SuperOperations execute a sequence of operations as a single unit, sharing the same retry mechanism. If any sub-operation fails, the entire SuperOperation can be retried.

Key features: - All sub-operations execute in sequence - Retry logic applies to the entire group - Reports are aggregated under the SuperOperation section - Branching (Conditions) is NOT permitted in SuperOperations - Protocol treats it the same as a regular operation

Subclasses must: 1. Call super().__init__() in their __init__ 2. Set self.operations to a list of ProtocolOperation instances 3. NOT include any Condition instances in self.operations

Example

>>> class CalibrationSuite(SuperOperationBase):
...     def __init__(self, params):
...         super().__init__()
...         self.operations = [
...             ResonatorSpectroscopy(params),
...             PowerRabi(params),
...             PiSpectroscopy(params)
...         ]
...
...     def evaluate(self) -> OperationStatus:
...         # All operations succeeded, check aggregate quality
...         if all_calibrations_good():
...             return OperationStatus.SUCCESS
...         else:
...             return OperationStatus.RETRY

analyze() → None

Not used in SuperOperation - operations handle their own analysis

correct(result: EvaluateResult) → EvaluateResult

Apply super-operation level corrections. Override to implement super-level parameter changes. Default: no-op (sub-operations handle their own corrections inside their own execute() calls).

execute() → EvaluateResult

Execute all sub-operations in sequence and aggregate their reports.

This method overrides ProtocolOperation.execute() to iterate through all sub-operations instead of calling measure/load_data/analyze.

Returns:

EvaluateResult from correct() after all sub-operations complete.

load_data() → bool

Not used in SuperOperation - operations handle their own data loading

measure() → Path

Not used in SuperOperation - operations handle their own measurement

operations: list[ProtocolOperation]

labcore.protocols.base.select_platform(platform: PlatformTypes | str) → None

Select the hardware platform for subsequent protocol execution.

Must be called once before instantiating any ProtocolBase subclass. Accepts either a PlatformTypes member or its name as a string (case-insensitive).

labcore.protocols.base.serialize_fit_params(params: Any) → dict[str, dict[str, float | None]]