xq-pulse 0.5.0

Pulse program implementation library

xq-pulse

xq-pulse is a pulse-level abstraction for NV-center control that stays independent of the underlying hardware. You describe experiments as a typed, unit-safe pulse program; hardware-specific execution is handled later via channel mapping and setup backends.

Why this package exists

Most NV experiments share the same logical pulse structure (drive, laser, acquisition, delays, sweeps, loops), but different devices expose different channel APIs. xq-pulse separates those concerns:

• You build a hardware-agnostic pulse tree with a Python DSL.
• Parameters and units are explicit (pint) and validated early.
• You can use symbolic expressions for sweep-dependent values.
• You can unroll, inspect, visualize, serialize, and map to hardware channels.

Installation

Inside this monorepo:

uv sync

As a standalone dependency (if published in your environment):

pip install xq-pulse

Quick Start

from xq_pulse.pulse.dsl import (
    acquire,
    acquisition_target,
    delay,
    drive,
    laser,
    parallel,
    pulse_program,
)
from xq_pulse.util import unit

with pulse_program() as program:
    pl = acquisition_target("pl_counts", bins=200)

    drive(
        duration=200 * unit.ns,
        frequency=2.87 * unit.GHz,
        amplitude=100 * unit.uT,
        phase=0 * unit.rad,
    )
    with parallel():
        laser(duration=1 * unit.us, wavelength=532 * unit.nm, power=75 * unit.mW)
        acquire(duration=1 * unit.us, target=pl, bin=0)
    delay(duration=500 * unit.ns)

# Inspect or visualize
program_unrolled = program.unroll()
figure = program.plot(show=False)

Mental Model

xq-pulse programs are pulse trees built from mostly immutable node types (attrs @frozen pulse classes).

• Leaf pulses:
  • DelayPulse
  • DrivePulse
  • LaserPulse
  • AcquisitionPulse
  • ChannelMappedPulse (after mapping)
• Container pulses:
  • SequencePulse (serial composition)
  • ParallelPulse (time overlap)
  • ForLoopPulse (parametric repetition)

PulseProgram adds:

• root: pulse tree
• acquisition_targets: named acquisition buffers
• parameter_sweeps: global sweeps declared at program level

Unit System (pint)

All physical values are pint.Quantity values from:

from xq_pulse.util import unit

Examples:

• 200 * unit.ns
• 2.87 * unit.GHz
• 100 * unit.uT
• 90 * unit.deg or 0.5 * unit.rad

Important details:

• Bare int/float inputs to DSL sweep/loop helpers are treated as dimensionless.
• Unit compatibility is validated on pulse creation.
• Expressions are unit-aware and checked when constructed/validated.

DSL Overview

Main builders in xq_pulse.pulse.dsl:

• Program/context:
  • pulse_program()
  • sequence()
  • parallel()
  • parameter_sweep(start, stop, step)
  • for_idx(start, stop, step)
• Pulses:
  • drive(...)
  • laser(...)
  • delay(...)
  • acquire(...)
  • acquisition_target(name, bins)

1) Program context

All pulse-emitting functions must run inside with pulse_program():.

from xq_pulse.pulse.dsl import delay, pulse_program
from xq_pulse.util import unit

with pulse_program() as program:
    delay(duration=1 * unit.us)

2) Sequence and parallel composition

with pulse_program() as program:
    with sequence():
        drive(duration=200 * unit.ns, frequency=2.87 * unit.GHz, amplitude=100 * unit.uT)
        with parallel():
            laser(duration=1 * unit.us, wavelength=532 * unit.nm, power=75 * unit.mW)
            acquire(duration=1 * unit.us, target=acquisition_target("pl", bins=100), bin=0)

3) Parameter sweeps (experiment axes)

Use parameter_sweep for global experiment scans.

from xq_pulse.pulse.dsl import drive, parameter_sweep, pulse_program
from xq_pulse.util import unit

with pulse_program() as program:
    with parameter_sweep(
        start=2.84 * unit.GHz,
        stop=2.90 * unit.GHz,
        step=1 * unit.MHz,
    ) as (sweep_idx, frequency):
        drive(
            duration=200 * unit.ns,
            frequency=frequency,
            amplitude=100 * unit.uT,
        )

Notes:

• parameter_sweep yields (index_parameter, value_parameter).
• Sweeps are converted into ForLoopPulse nodes during unroll/plot.
• Sweep points are inclusive of stop (with a small numerical epsilon).
• There is a safety limit of < 10_000 elements per sweep.
• Sweeps must be declared before adding pulses to the program body.

4) For-loops inside the pulse tree

Use for_idx for loop structure in the pulse tree itself.

with pulse_program() as program:
    with for_idx(start=0, stop=3, step=1) as (i, _loop_value):
        acquire(duration=1 * unit.us, target=acquisition_target("pl", bins=4), bin=i)

for_idx also supports physical units:

with for_idx(start=2.84 * unit.GHz, stop=2.90 * unit.GHz, step=2 * unit.MHz) as (_, freq):
    drive(duration=100 * unit.ns, frequency=freq, amplitude=80 * unit.uT)

Reusable Sequence Functions

A common pattern is writing Python functions that emit DSL operations, then calling them inside a context.

from xq_pulse.pulse.dsl import delay, drive, pulse_program, sequence
from xq_pulse.util import Quantity, unit

def x_pulse(frequency: Quantity, amplitude: Quantity, rabi_period: Quantity) -> None:
    drive(duration=rabi_period / 2, frequency=frequency, amplitude=amplitude)

def y_pulse(frequency: Quantity, amplitude: Quantity, rabi_period: Quantity) -> None:
    drive(
        duration=rabi_period / 2,
        frequency=frequency,
        amplitude=amplitude,
        phase=90 * unit.deg,
    )

def xy_block(tau: Quantity, frequency: Quantity, amplitude: Quantity, rabi_period: Quantity) -> None:
    with sequence():
        x_pulse(frequency, amplitude, rabi_period)
        delay(duration=tau - rabi_period / 2)
        y_pulse(frequency, amplitude, rabi_period)

with pulse_program() as program:
    xy_block(
        tau=1 * unit.us,
        frequency=2.87 * unit.GHz,
        amplitude=100 * unit.uT,
        rabi_period=80 * unit.ns,
    )

This is how the built-in experiment helpers are implemented.

Symbolic Expressions and Parameters

Pulse fields can be expressions, not only literals.

import numpy as np
from xq_pulse.pulse.dsl import drive, parameter_sweep, pulse_program
from xq_pulse.util import unit

with pulse_program() as program:
    with parameter_sweep(start=1000 * unit.ns, stop=2000 * unit.ns, step=200 * unit.ns) as (_, tau):
        phase_shift = 2 * np.pi * unit.rad * (1.08 * unit.MHz) * tau
        drive(
            duration=1 * unit.us,
            frequency=2.86 * unit.GHz,
            amplitude=75 * unit.uT,
            phase=phase_shift,
        )

Supported expression operations include:

• Sum / difference
• Product / quotient
• Maximum (xq_pulse.pulse.expression.max(...))
• Binding via bind(...)
• Evaluation via eval_expression(...) once fully bound

The package patches pint.Quantity arithmetic so expressions like 700 * unit.ns - sweep_param work naturally.

Envelope Shaping

Drive pulses accept an Envelope (SquareEnvelope by default).

Available envelope types:

• SquareEnvelope(level=...)
• PiecewiseEnvelope(segments=(EnvelopeSegment(...), ...))
• SymbolicEnvelope(expression="...")
• Convenience helpers:
  • gaussian_envelope(...)
  • sin_envelope()
  • symbolic_envelope(expression=...)

Piecewise example

from xq_pulse.pulse.envelope import EnvelopeSegment, PiecewiseEnvelope

env = PiecewiseEnvelope(
    segments=(
        EnvelopeSegment(end=0.3, value=0.0),
        EnvelopeSegment(end=1.0, value=1.0),
    )
)

Symbolic example

from xq_pulse.pulse.envelope import SymbolicEnvelope

env = SymbolicEnvelope(expression="exp(-((tau - 0.5) / 0.15)**2)")

Symbolic envelope rules:

• Variable must be tau (normalized time in [0, 1]).
• Only supported math functions are allowed (exp, sin, cos, tan, sqrt, log, Abs, Piecewise, Max, Min).
• Envelope values are dimensionless.

Apply to a drive pulse:

drive(
    duration=100 * unit.ns,
    frequency=2.87 * unit.GHz,
    amplitude=150 * unit.uT,
    envelope=env,
)

Plot an envelope and its FFT:

figure = env.plot(show=False)

Program Introspection and Transformation

PulseProgram utilities:

• program.simplify(): simplify expression and container structure.
• program.unroll(): expand loops/sweeps to a timeline (UnrolledPulse).
• program.plot(show=False): pulse timeline visualization.
• program.append(other, gap=...): concatenate two programs (second must have no sweeps).

UnrolledPulse utilities:

• pulse_starts, pulse_ends, unrolled_pulses
• source_by_unrolled and unrolled_by_source mappings
• append(...) and merge(...)

ActivePulseIntervals can split unrolled timelines into contiguous regions with stable active-pulse sets.

Hardware Abstraction: Channels and Setup

Execution is abstracted through Setup and Channel.

• Define channel classes with can_generate(pulse).
• Define a Setup with available channels and a run(program) method.
• Use map_channels(program, setup) to assign abstract pulses to concrete channels.

from attrs import frozen
from xq_pulse.pulse.channel import Channel, ChannelType
from xq_pulse.pulse.channel_mapping import map_channels
from xq_pulse.pulse.pulse import DrivePulse
from xq_pulse.pulse.setup import Setup
from xq_pulse.util import unit

@frozen
class MWChannel(Channel):
    type: ChannelType = ChannelType.DRIVE

    def can_generate(self, pulse):
        return isinstance(pulse, DrivePulse) and (2.2 * unit.GHz <= pulse.frequency <= 3.0 * unit.GHz)

@frozen
class MySetup(Setup):
    channels: frozenset[Channel] = frozenset({MWChannel(name="mw1")})

    def run(self, program):
        mapped = map_channels(program, self)
        # Translate mapped program into hardware calls here.

Mapping constraints:

• Programs cannot have free parameters outside declared sweeps.
• Parallel branches require non-conflicting channel assignments.

Serialization

Use xq_pulse.pulse.serialization.create_converter() for cattrs-based serialization/deserialization.

import json
from xq_pulse.pulse.program import PulseProgram
from xq_pulse.pulse.serialization import create_converter

converter = create_converter()
payload = converter.unstructure(program)
json.dumps(payload)  # transport-safe
program2 = converter.structure(payload, PulseProgram)

Serialization notes:

• Envelopes and expressions are tagged unions.
• ChannelMappedPulse is intentionally not serializable; serialize unmapped programs.

Built-In Experiment Helpers (xq_pulse.experiments)

Top-level helpers:

from xq_pulse.experiments import (
    deer,
    deer_sequence,
    hahn_echo,
    hahn_echo_sequence,
    pulsed_odmr,
    rabi,
    ur8,
    ur8_sequence,
    ur16,
    ur16_sequence,
    xy8,
    xy8_sequence,
)

What they provide:

• rabi(...): duration sweep of a drive pulse.
• pulsed_odmr(...): frequency sweep for ODMR.
• hahn_echo(...) / hahn_echo_sequence(...): spin-echo block and swept sequence.
• xy8(...) / xy8_sequence(...): XY8 dynamical decoupling with order and tau sweep.
• ur8/ur16 variants: universally robust DD sequences.
• deer(...) / deer_sequence(...): DEER-style coupled sequence.

Additional advanced helper available in submodule:

from xq_pulse.experiments.axy import axy8, axy8_sequence

Validation Ranges and Assumptions

Built-in pulse validators enforce practical bounds:

• duration: 1 ns to 1000 us
• DrivePulse.amplitude: (0, 1 mT)
• DrivePulse.frequency: (0, 20 GHz)
• DrivePulse.phase: radians-compatible
• LaserPulse.wavelength: (400, 1000) nm
• LaserPulse.power: (0, 500) mW
• AcquisitionPulse.bin: dimensionless and >= 0

Common pitfalls:

• Calling DSL pulse functions outside a context raises an error.
• acquisition_target(...) must be called inside pulse_program.
• Unit mismatches fail fast.
• Negative/zero durations or out-of-range physical values are rejected.

Minimal End-to-End Example

This combines sweeps, reusable blocks, envelope shaping, and serialization:

from xq_pulse.pulse.dsl import (
    acquire,
    acquisition_target,
    delay,
    drive,
    parameter_sweep,
    pulse_program,
    sequence,
)
from xq_pulse.pulse.envelope import gaussian_envelope
from xq_pulse.pulse.serialization import create_converter
from xq_pulse.util import unit

def readout_block(target) -> None:
    with sequence():
        delay(duration=200 * unit.ns)
        acquire(duration=1 * unit.us, target=target, bin=0)

with pulse_program() as program:
    target = acquisition_target("pl", bins=200)
    env = gaussian_envelope(amplitude=1.0, center=0.5, sigma=0.15)
    with parameter_sweep(start=2.84 * unit.GHz, stop=2.90 * unit.GHz, step=2 * unit.MHz) as (_, freq):
        drive(
            duration=120 * unit.ns,
            frequency=freq,
            amplitude=120 * unit.uT,
            envelope=env,
        )
        readout_block(target)

figure = program.plot(show=False)
payload = create_converter().unstructure(program)