Executor

`keysight.qcs.executor.CallbackHandler`	An abstract class that provides a callback method for live data.
`keysight.qcs.executor.Executor`	A class to manipulate and alter programs.
`keysight.qcs.executor.HclBackend`	A pass for running a `Program` on hardware.
`keysight.qcs.executor.LinkerPass`	A pass for converting operations in a `Program`.
`keysight.qcs.executor.LiveDataSubscriber`	A class for subscribing to live data published to a nats server.
`keysight.qcs.executor.LiveIqData`	A class containing live IQ data.
`keysight.qcs.executor.LivePlottingConfig`	Defines options for live data plotting.
`keysight.qcs.executor.Pass`	An abstract base class for all passes that manipulate and return an altered program.

CallbackHandler

class keysight.qcs.executor.CallbackHandler

An abstract class that provides a callback method for live data.

Requires subclasses to override the handle_data.

abstract handle_data(data: LiveIqData)

A callback method for live data. This method gets called each time new data is received.

Executor

class keysight.qcs.executor.Executor(*passes: Pass)

A class to manipulate and alter programs.

An instance of this class stores a list of Passes that define rules for how to alter (e.g., convert gate operations to waveforms) or execute programs.

Parameters:

*passes – The passes for this executor.

property passes: tuple[keysight.qcs.executor.base.Pass]

A list of passes applied by this executor.

execute(program: Program) → Program

Returns the executed program by sequentially applying each pass to the program.

Parameters:

program – The program to execute.

HclBackend

class keysight.qcs.executor.HclBackend(channel_mapper: ChannelMapper, hw_demod: bool = False, init_time: float = 0.0001, hvi_debug: bool = False, blocking: bool = True, auto_delays: bool = True, suppress_rounding_warnings: bool = False, keep_progress_bar: bool = True, live_plotting_config: Optional[Union[LivePlottingConfig, Channels]] = None, reset_phase_every_shot: bool = False)

A pass for running a Program on hardware.

This pass supports hardware sweeping of variables. This is enabled automatically when a program’s repetitions() contain a Sweep as its innermost repetition. This can be accomplished by calling sweep() before n_shots().

Note

This feature is currently only supported with hw_demod=True and will only sweep the innermost sweep and repeat on hardware.

Parameters:

channel_mapper – A ChannelMapper.

hw_demod – Whether to demodulate and classify trace data directly on the hardware. Doing this loses the trace data and only IQ and classified will be accessible.

init_time – The time delay between the end of one program iteration and the start of the next iteration, in seconds.

hvi_debug – Whether to log HVI debugging information.

blocking – Whether to block the Python kernel while the program is executing. When blocking is False, the program is returned immediately with its loader set to access the results once it finishes executing. This disables the progress bar.

auto_delays – Whether to automatically add delays based on FPGA firmware. This option has been deprecated and now must remain True.

suppress_rounding_warnings – Whether to suppress warnings about rounding durations.

keep_progress_bar – If True, keeps the progress bar upon completion of Program.

live_plotting_config – The configuration or channels for live result plotting, or None to disable live plotting.

reset_phase_every_shot – Whether to performa a phase reset across all channels before every program iteration.

Raises:

ValueError – If the current platform is not Windows and no remote address was provided in the channel mapper.

property blocking: bool

Whether to block the kernel during execution of programs.

property channel_mapper: ChannelMapper

The map from virtual to physical channels.

property hw_demod: bool

Whether to demodulate trace data directly on the hardware by default.

abort_program(accession_id: int) → bool

Abort a program with a specified accession_id.

Parameters:

accession_id – The accession id of the program to abort.

activate_user() → None

Activate user who has been deactivated. Only users with the ‘Admin’ privileges can activate a user.

apply(program: Program, **kwargs) → Program

Runs the program on hardware.

Parameters:

program – The program to run.

Raises:

ValueError – If the system is not ready to run programs.

ValueError – If performing a hardware sweep without hardware demodulation.

ValueError – If performing a nested hardware sweep.

assign_user_role() → None

Assign a role for user. Only users with the ‘Admin’ privileges can assign role to themselves or other users.

change_my_password() → None

Change current password. Users can change their current password.

change_password() → None

Change current password. Only user with ‘Admin’ privileges can change team members password.

create_user() → None

Create new users. Only users with ‘Admin’ privileges can create new users.

deactivate_user() → None

Deactivate an active user. Only users with the ‘Admin’ privileges can deactivate a user.

get_messages(accession_id: int, skip_messages: int = 0) → list[str]

Gets the messages associated with a program submission.

Parameters:

accession_id – The accession ID of the program.

skip_messages – The number of messages to skip.

get_program(accession_id: int, timeout: int = 300) → Program

Gets the program associated with an accession_id from the database.

Parameters:

accession_id – The accession ID of the program.

timeout – The timeout in seconds to wait.

get_program_state(accession_id: int) → str

Returns a string representation of a program’s execution state

Parameters:

accession_id – The accession id of the program to check the status.

get_program_execution_history(count: int = 10) → list

Returns historical list of executed program accession id’s.

Parameters:

count – Total number of executed program accession id’s to return, default value is 10.

get_pending_programs_info(state: Optional[Literal['Queued', 'Running', 'Compiling']] = None) → list[dict]

Returns the list of information about pending programs from the queue.

Parameters:

state – Optional parameter to filter the pending programs,

it can be one of the states - Queued, Compiling, Running. If None, gets all pending programs of all of the above states

is_program_completed(accession_id: int) → bool

Whether a program is completed.

Parameters:

accession_id – The accession id of the program to check the status.

is_system_ready() → bool

Whether the system is ready to run programs.

login(username=None, password=None, cloud_user=False) → None

Authenticates and starts a user session.

When no credentials are passed as arguments, the username and password is obtained via user inputs.

When credentials are passed as arguments, the password should be encoded in base64 in order to obfuscate. Visit. https://www.base64encode.org/ for a simple online encoder.

Parameters:

username – The username.

password – The password, in base64.

cloud_user – True if cloud user is trying to login.

revoke_user_role() → None

Revoke a role for user. Only users with the ‘Admin’ privileges can revoke role to themselves or other users.

revoke_token() → None

Revoke an authentication token.

set_access_token(tokenId: str) → None

Set access token for starting a user session.

Parameters:

tokenId – The Access token to set.

get_log_messages(max_messages: int = 100, only_errors: bool = False) → list[tuple[str, str, int]]

Returns a list of logs containing timestamp, message, and level information based on the provided parameters. Log Level information corresponds to 1 = Debug, 2 = Info, 3 = Warning, 4 = Error

MaxMessages:

Maximum number of logs to be retrieved, default 100.

OnlyErrors:

Boolean flag for logs with errors, default false.

run_module_self_test() → list[tuple[str, str]]

Returns a collection of the self test results, one entry for each module. This is accessible only to users with ‘Admin’ privileges

get_module_temperatures() → list[tuple[str, float]]

Returns a collection of module temperatures, one entry for each module. This is accessible only to users with ‘Admin’ privileges

Linker Pass

class keysight.qcs.executor.LinkerPass(*linkers: BaseLinker, duration_map: dict[BaseOperation, float | Variable[float]] | None = None)

A pass for converting operations in a Program.

Note

When using linkers that contain operations without specified durations, their durations should be specified using the duration_map argument.

Note

When using this pass with conditional operations, rather than using a single linker for the entire operation, the pass will instead link each of the operations individually to create a new conditional operation.

Note

Linkers with conditional operations in their program are not supported by this pass.

Parameters:

*linkers – A BaseLinker to apply.

duration_map – A dictionary from base operations to their durations.

apply(program: Program) → Program

Applies the pass.

Parameters:

program – The program to apply the pass to.

LiveDataSubscriber

class keysight.qcs.executor.LiveDataSubscriber(host: str, shape: tuple[int], channels: Channels)

A class for subscribing to live data published to a nats server.

This class has a main method called subscribe, which connects to the specified nats server and begins listening for data. When done streaming, the caller must call stop in order to end the subscription thread.

Parameters:

host – The IP address to subscribe to.

shape – The shape of the live data.

channels – The channels to subscribe to.

accession_id – The accession_id associated with the program the data is generated from.

property channels: str

The channels to subscribe to.

property live_data: LiveIqData

The live data.

property nats_address: str

The nats server address to listen to data on.

property subject: str

The nats subject to listen for messages on.

subscribe(callbackHandler: CallbackHandler, accession_id: int) → None

Subscribes to data coming from a nats server. When valid data is received, the handle_data method on callbackHandler is called.

This method starts a new thread. It is the responsibility of the caller to close the thread by calling stop().

Parameters:

callbackHandler – A class that provides a handle_data method. This method gets called when valid data is received.

accession_id – The program accession id to subscribe to.

stop() → None

Stops subscribing to data and joins the subscription thread.

LiveIqData

class keysight.qcs.executor.LiveIqData(channels: Channels, shape: tuple[int])

A class containing live IQ data.

Parameters:

channels – The channels to filter for.

shape – The shape of the live data.

property data: dict[keysight.qcs.channels.channels.Channels, numpy.ndarray[complex]]

Returns the data as a dictionary from single element Channels to a numpy array of IQ data.

Returns:

the data as a dictionary from single element Channels

to a numpy array of IQ data.

add_new_chunk(chunk: dict) → bool

Adds a new chunk to this.

Parameters:

chunk – a decoded json message from the nats server.

Returns:

True if new data was added to this, False otherwise.

LivePlottingConfig

class keysight.qcs.executor.LivePlottingConfig(channels: Channels, iq_mode: Literal['I', 'Q', 'Magnitude', 'Phase'] = 'Magnitude', skip_plotting: bool = False, callback: Optional[CallbackHandler] = None)

Defines options for live data plotting.

Parameters:

channels – The digitizer virtual channels to plot.

iq_mode – The default view of the IQ plot. Must be one of “I”, “Q”, “Magnitude”, or “Phase”.

skip_plotting – When True this will enable the data streaming from the hardware while skipping live plotting. This is recomended for using cloud dashboard to display the live data.

callback – When not None, this skips live plotting in favor of a custom callback handler.

Pass

class keysight.qcs.executor.Pass

An abstract base class for all passes that manipulate and return an altered program.

Every pass has an apply method that accepts a program and returns a (potentially new) instance of the program. Pass-specific options can be provided when initializing the pass.

abstract apply(program: Program) → Program

Applies the pass.

Parameters:

program – The program to apply the pass to.

On this page