Metadata-Version: 2.3
Name: xq-pulse
Version: 0.5.0
Summary: Pulse program implementation library
Author: Pit Hüne
Author-email: Pit Hüne <pit@xeedq.com>
Requires-Dist: attrs>=25.3.0
Requires-Dist: cattrs>=24.1.0
Requires-Dist: frozendict>=2.4.6
Requires-Dist: intervaltree>=3.1.0
Requires-Dist: pint>=0.24.4
Requires-Dist: plotly>=5.24.1
Requires-Dist: plotly-resampler>=0.11.0
Requires-Dist: python-constraint2>=2.4.0
Requires-Dist: sympy>=1.13.3
Requires-Python: >=3.10
Description-Content-Type: text/markdown

# 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:

```bash
uv sync
```

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

```bash
pip install xq-pulse
```

## Quick Start

```python
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:

```python
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():`.

```python
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

```python
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.

```python
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.

```python
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:

```python
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.

```python
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.

```python
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

```python
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

```python
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:

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

Plot an envelope and its FFT:

```python
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.

```python
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.

```python
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:

```python
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:

```python
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:

```python
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)
```
