Helpers

InstrumentResolver

InstrumentResolver()

Resolve instrument aliases and ids to their InstrumentInfo.

Maintains a cache of the instruments known to a client, keyed both by resource id and by short alias (the part of an alias after the unit prefix). Call refresh() to populate the cache from a client before resolving.

Source code in quelware-client/src/quelware_client/client/helpers/instrument_resolver/__init__.py

def __init__(self):
    self._id_to_inst_info: dict[ResourceId, InstrumentInfo] = {}
    self._short_alias_to_ids: dict[str, dict[str, ResourceId]] = {}

find_inst_info_by_alias

find_inst_info_by_alias(alias, unit=None)

Return the cached info for an instrument alias.

Parameters:

Name	Type	Description	Default
alias	str	Short alias of the instrument (without the unit prefix).	required
unit	str | None	Unit label to disambiguate when the alias exists on more than one unit.	None

Returns:

Type	Description
InstrumentInfo	The instrument's information.

Raises:

Type	Description
ValueError	If the alias is unknown, is missing on unit, or is ambiguous and no unit was given.

Source code in quelware-client/src/quelware_client/client/helpers/instrument_resolver/__init__.py

def find_inst_info_by_alias(
    self, alias: str, unit: str | None = None
) -> InstrumentInfo:
    """Return the cached info for an instrument alias.

    Args:
        alias: Short alias of the instrument (without the unit prefix).
        unit: Unit label to disambiguate when the alias exists on more
            than one unit.

    Returns:
        The instrument's information.

    Raises:
        ValueError: If the alias is unknown, is missing on ``unit``, or is
            ambiguous and no ``unit`` was given.
    """
    return self.find_inst_info_by_id(self._resolve_single(alias, unit))

find_inst_info_by_id

find_inst_info_by_id(instrument_id)

Return the cached info for an instrument id.

Parameters:

Name	Type	Description	Default
instrument_id	ResourceId	Full resource id of the instrument.	required

Returns:

Type	Description
InstrumentInfo	The instrument's information.

Raises:

Type	Description
ValueError	If the id is not in the cache.

Source code in quelware-client/src/quelware_client/client/helpers/instrument_resolver/__init__.py

def find_inst_info_by_id(self, instrument_id: ResourceId) -> InstrumentInfo:
    """Return the cached info for an instrument id.

    Args:
        instrument_id: Full resource id of the instrument.

    Returns:
        The instrument's information.

    Raises:
        ValueError: If the id is not in the cache.
    """
    if instrument_id not in self._id_to_inst_info:
        raise ValueError(f"Instrument with id '{instrument_id}' not found.")
    return self._id_to_inst_info[instrument_id]

refresh async

refresh(client)

Rebuild the instrument cache from a client.

Fetches information for every instrument resource the client exposes and indexes it by id and by short alias.

Parameters:

Name	Type	Description	Default
client	QuelwareClient	An initialized client to query.	required

Source code in quelware-client/src/quelware_client/client/helpers/instrument_resolver/__init__.py

async def refresh(self, client: QuelwareClient):
    """Rebuild the instrument cache from a client.

    Fetches information for every instrument resource the client exposes
    and indexes it by id and by short alias.

    Args:
        client: An initialized client to query.
    """
    resource_infos = list(
        rinfo
        for rinfo in await client.list_resource_infos()
        if rinfo.category is ResourceCategory.INSTRUMENT
    )

    coros = [client.get_instrument_info(rinfo.id) for rinfo in resource_infos]

    inst_infos: list[InstrumentInfo] = await asyncio.gather(*coros)

    new_id_to_inst_info: dict[ResourceId, InstrumentInfo] = {}
    new_short: dict[str, dict[str, ResourceId]] = {}
    for inst_info in inst_infos:
        full_alias = inst_info.definition.alias
        new_id_to_inst_info[inst_info.id] = inst_info
        unit, _, short = full_alias.partition(":")
        if short:
            new_short.setdefault(short, {})[unit] = inst_info.id
        else:
            new_short.setdefault(full_alias, {})[""] = inst_info.id

    self._id_to_inst_info = new_id_to_inst_info
    self._short_alias_to_ids = new_short
    logger.info(f"{len(self._id_to_inst_info)} instruments has been registered.")

resolve

resolve(aliases, unit=None)

Resolve several instrument aliases to their resource ids.

Parameters:

Name	Type	Description	Default
aliases	list[str]	Short aliases to resolve.	required
unit	str | None	Unit label to disambiguate aliases present on multiple units.	None

Returns:

Type	Description
list[ResourceId]	The resource ids, in the same order as aliases.

Raises:

Type	Description
ValueError	If any alias is unknown or ambiguous.

Source code in quelware-client/src/quelware_client/client/helpers/instrument_resolver/__init__.py

def resolve(self, aliases: list[str], unit: str | None = None) -> list[ResourceId]:
    """Resolve several instrument aliases to their resource ids.

    Args:
        aliases: Short aliases to resolve.
        unit: Unit label to disambiguate aliases present on multiple units.

    Returns:
        The resource ids, in the same order as ``aliases``.

    Raises:
        ValueError: If any alias is unknown or ambiguous.
    """
    return list(self._resolve_single(alias, unit) for alias in aliases)

Sequencer

Sequencer(
    default_sampling_period_ns, enforce_sample_grid=True
)

Build fixed timelines of waveform events and capture windows.

Waveforms are registered by name and then scheduled on instruments at nanosecond offsets, alongside capture windows. Each instrument alias is bound to its hardware sampling period so offsets and lengths can be validated against the sample grid and converted to samples. The assembled timeline for an instrument is exported as a SetFixedTimeline directive.

Create a sequencer.

Parameters:

Name	Type	Description	Default
default_sampling_period_ns	float	Sampling period, in nanoseconds, used for waveforms registered without an explicit period.	required
enforce_sample_grid	bool	When True, offsets and lengths that do not land on the sample grid raise ValueError; when False, they are rounded to the nearest sample with a warning.	True

Source code in quelware-client/src/quelware_client/client/helpers/sequencer/__init__.py

def __init__(
    self, default_sampling_period_ns: float, enforce_sample_grid: bool = True
):
    """Create a sequencer.

    Args:
        default_sampling_period_ns: Sampling period, in nanoseconds, used
            for waveforms registered without an explicit period.
        enforce_sample_grid: When True, offsets and lengths that do not
            land on the sample grid raise `ValueError`; when False, they
            are rounded to the nearest sample with a warning.
    """
    self._waveform_library: dict[str, _Waveform] = {}
    self._alias_to_events: dict[str, list[_SequencerEvent]] = defaultdict(list)
    self._alias_to_capwin: dict[str, list[_SequencerCaptureWindow]] = defaultdict(
        list
    )

    self._default_sampling_period_ns: float = default_sampling_period_ns
    self._iterations: int = 1

    self._bindings: dict[str, _AliasBinding] = {}
    self._enforce_sample_grid: bool = enforce_sample_grid
    self._length_ns: float = 0.0

aligned_length_fs property

aligned_length_fs

Timeline length in femtoseconds, aligned to bound step sizes.

The raw length is rounded up to a multiple of the least common multiple of each bound alias's sampling_period_fs * step_samples.

add_capture_window

add_capture_window(
    instrument_alias,
    window_name,
    start_offset_ns,
    length_ns,
)

Schedule a capture window on an instrument.

Parameters:

Name	Type	Description	Default
instrument_alias	str	Alias of the (bound) target instrument.	required
window_name	str	Name of the capture window.	required
start_offset_ns	float	Start time of the window, in nanoseconds.	required
length_ns	float	Length of the window, in nanoseconds.	required

Raises:

Type	Description
ValueError	If the offset or length is off the sample grid while grid enforcement is on.

Source code in quelware-client/src/quelware_client/client/helpers/sequencer/__init__.py

def add_capture_window(
    self,
    instrument_alias: str,
    window_name: str,
    start_offset_ns: float,
    length_ns: float,
):
    """Schedule a capture window on an instrument.

    Args:
        instrument_alias: Alias of the (bound) target instrument.
        window_name: Name of the capture window.
        start_offset_ns: Start time of the window, in nanoseconds.
        length_ns: Length of the window, in nanoseconds.

    Raises:
        ValueError: If the offset or length is off the sample grid while
            grid enforcement is on.
    """
    self._check_and_convert_to_samples(
        instrument_alias,
        start_offset_ns,
        f"Capture window '{window_name}' start_offset_ns",
    )
    self._check_and_convert_to_samples(
        instrument_alias, length_ns, f"Capture window '{window_name}' length_ns"
    )

    capwin = _SequencerCaptureWindow(
        name=window_name,
        start_offset_ns=start_offset_ns,
        length_ns=length_ns,
    )
    self._alias_to_capwin[instrument_alias].append(capwin)
    end_at_ns = start_offset_ns + length_ns
    self._length_ns = max(self._length_ns, end_at_ns)

add_event

add_event(
    instrument_alias,
    waveform_name,
    start_offset_ns,
    gain=1.0,
    phase_offset_deg=0.0,
)

Schedule a registered waveform on an instrument.

Parameters:

Name	Type	Description	Default
instrument_alias	str	Alias of the (bound) target instrument.	required
waveform_name	str	Name of a registered waveform.	required
start_offset_ns	float	Start time of the event, in nanoseconds.	required
gain	float	Linear gain applied to the waveform.	1.0
phase_offset_deg	float	Phase offset applied to the waveform, in degrees.	0.0

Raises:

Type	Description
ValueError	If the waveform is not registered, or the offset is off the sample grid while grid enforcement is on.

Source code in quelware-client/src/quelware_client/client/helpers/sequencer/__init__.py

def add_event(
    self,
    instrument_alias: str,
    waveform_name: str,
    start_offset_ns: float,
    gain: float = 1.0,
    phase_offset_deg: float = 0.0,
):
    """Schedule a registered waveform on an instrument.

    Args:
        instrument_alias: Alias of the (bound) target instrument.
        waveform_name: Name of a registered waveform.
        start_offset_ns: Start time of the event, in nanoseconds.
        gain: Linear gain applied to the waveform.
        phase_offset_deg: Phase offset applied to the waveform, in degrees.

    Raises:
        ValueError: If the waveform is not registered, or the offset is off
            the sample grid while grid enforcement is on.
    """
    if waveform_name not in self._waveform_library:
        raise ValueError(f"waveform '{waveform_name}' is not registered.")

    self._check_and_convert_to_samples(
        instrument_alias, start_offset_ns, "Event start_offset_ns"
    )

    event = _SequencerEvent(
        waveform_name=waveform_name,
        start_offset_ns=start_offset_ns,
        gain=gain,
        phase_offset_deg=phase_offset_deg,
    )
    self._alias_to_events[instrument_alias].append(event)

    waveform = self._waveform_library[waveform_name]
    end_at_ns = (
        start_offset_ns + len(waveform.iq_array) * waveform.sampling_period_ns
    )
    self._length_ns = max(self._length_ns, end_at_ns)

bind

bind(alias, sampling_period_fs, step_samples)

Bind an instrument alias to its hardware timing.

Must be called before adding events or capture windows for the alias.

Parameters:

Name	Type	Description	Default
alias	str	Instrument alias to bind.	required
sampling_period_fs	int	Sampling period of the instrument, in femtoseconds.	required
step_samples	int	Granularity, in samples, that the timeline length is aligned to for this alias.	required

Source code in quelware-client/src/quelware_client/client/helpers/sequencer/__init__.py

def bind(self, alias: str, sampling_period_fs: int, step_samples: int):
    """Bind an instrument alias to its hardware timing.

    Must be called before adding events or capture windows for the alias.

    Args:
        alias: Instrument alias to bind.
        sampling_period_fs: Sampling period of the instrument, in
            femtoseconds.
        step_samples: Granularity, in samples, that the timeline length is
            aligned to for this alias.
    """
    self._bindings[alias] = _AliasBinding(
        sampling_period_fs=sampling_period_fs,
        step_samples=step_samples,
    )

export_set_fixed_timeline_directive

export_set_fixed_timeline_directive(instrument_alias)

Build the timeline directive for one instrument.

Collects the events and capture windows scheduled for the alias, converts their nanosecond offsets to samples, and packages them with the aligned length and iteration count.

Parameters:

Name	Type	Description	Default
instrument_alias	str	Alias of the (bound) instrument to export.	required

Returns:

Type	Description
SetFixedTimeline	The assembled SetFixedTimeline directive.

Raises:

Type	Description
ValueError	If the alias is not bound.

Source code in quelware-client/src/quelware_client/client/helpers/sequencer/__init__.py

def export_set_fixed_timeline_directive(
    self, instrument_alias: str
) -> SetFixedTimeline:
    """Build the timeline directive for one instrument.

    Collects the events and capture windows scheduled for the alias,
    converts their nanosecond offsets to samples, and packages them with
    the aligned length and iteration count.

    Args:
        instrument_alias: Alias of the (bound) instrument to export.

    Returns:
        The assembled `SetFixedTimeline` directive.

    Raises:
        ValueError: If the alias is not bound.
    """
    if instrument_alias not in self._bindings:
        raise ValueError(f"Alias '{instrument_alias}' is not bound.")

    sampling_period_fs = self._bindings[instrument_alias].sampling_period_fs

    name_to_index: dict[str, int] = {}
    counter = 0
    local_library: list[IqWaveform] = []
    local_events: list[WaveformEvent] = []

    for event in self._alias_to_events[instrument_alias]:
        if event.waveform_name in name_to_index:
            index = name_to_index[event.waveform_name]
        else:
            waveform = self._waveform_library[event.waveform_name]
            local_library.append(
                IqWaveform(
                    sampling_period_fs=round(waveform.sampling_period_ns * 1e6),
                    iq_array=waveform.iq_array,
                )
            )
            index = counter
            name_to_index[event.waveform_name] = index
            counter += 1

        start_offset_samples = self._check_and_convert_to_samples(
            instrument_alias, event.start_offset_ns, "Export event"
        )
        local_events.append(
            WaveformEvent(
                waveform_index=index,
                start_offset_samples=start_offset_samples,
                gain=event.gain,
                phase_offset_deg=event.phase_offset_deg,
            )
        )

    local_capwins: list[CaptureWindow] = []
    for capwin in self._alias_to_capwin[instrument_alias]:
        local_capwins.append(
            CaptureWindow(
                name=capwin.name,
                start_offset_samples=self._check_and_convert_to_samples(
                    instrument_alias, capwin.start_offset_ns, "Export capwin start"
                ),
                length_samples=self._check_and_convert_to_samples(
                    instrument_alias, capwin.length_ns, "Export capwin length"
                ),
            )
        )

    length_sample = (
        self.aligned_length_fs + sampling_period_fs - 1
    ) // sampling_period_fs

    return SetFixedTimeline(
        waveform_library=local_library,
        events=local_events,
        capture_windows=local_capwins,
        length=length_sample,
        iterations=self._iterations,
    )

extend_length_ns

extend_length_ns(additional_ns)

Extend the overall timeline by additional_ns nanoseconds.

Source code in quelware-client/src/quelware_client/client/helpers/sequencer/__init__.py

def extend_length_ns(self, additional_ns: float):
    """Extend the overall timeline by ``additional_ns`` nanoseconds."""
    self._length_ns += additional_ns

register_waveform

register_waveform(name, waveform, sampling_period_ns=None)

Register a named IQ waveform.

Parameters:

Name	Type	Description	Default
name	str	Name used to reference the waveform in add_event().	required
waveform	ArrayLike	Complex IQ samples; every amplitude must lie within [-1, 1].	required
sampling_period_ns	float | None	Sampling period of the waveform, in nanoseconds. Defaults to the sequencer's default period.	None

Raises:

Type	Description
ValueError	If any sample amplitude exceeds 1 in magnitude.

Source code in quelware-client/src/quelware_client/client/helpers/sequencer/__init__.py

def register_waveform(
    self,
    name: str,
    waveform: npt.ArrayLike,
    sampling_period_ns: float | None = None,
):
    """Register a named IQ waveform.

    Args:
        name: Name used to reference the waveform in `add_event()`.
        waveform: Complex IQ samples; every amplitude must lie within
            ``[-1, 1]``.
        sampling_period_ns: Sampling period of the waveform, in
            nanoseconds. Defaults to the sequencer's default period.

    Raises:
        ValueError: If any sample amplitude exceeds 1 in magnitude.
    """
    if sampling_period_ns is None:
        sampling_period_ns = self._default_sampling_period_ns
    if np.any(np.abs(waveform) > 1):
        raise ValueError("The amplitude must be in the range -1 to 1.")
    self._waveform_library[name] = _Waveform(
        sampling_period_ns=sampling_period_ns,
        iq_array=np.array(waveform, dtype=complex),
    )

set_iterations

set_iterations(iterations)

Set how many times the exported timeline repeats.

Source code in quelware-client/src/quelware_client/client/helpers/sequencer/__init__.py

def set_iterations(self, iterations: int):
    """Set how many times the exported timeline repeats."""
    self._iterations = iterations