RsInstrument package

Subpackages

Submodules

RsInstrument.RsInstrument module

Root class for remote-controlling instrument with SCPI commands.

class RsInstrument(resource_name: str, id_query: bool = True, reset: bool = False, options: str | None = None, direct_session: object | None = None)[source]

Bases: object

Root class for remote-controlling instrument with SCPI commands.

Initializes new RsInstrument session.

Parameters:
  • resource_name – VISA resource name, e.g. ‘TCPIP::192.168.2.1::INSTR’

  • id_query – if True, the instrument’s model name is verified against the models supported by the driver and eventually throws an exception

  • reset – Resets the instrument (sends *RST) command and clears its status syb-system

  • direct_session – Another driver object or pyVisa object to reuse the session instead of opening a new session

  • options – string tokens alternating the driver settings. More tokens are separated by comma.

Parameter options tokens examples:
  • Simulate=True - starts the session in simulation mode. Default: False

  • SelectVisa=socketio - uses no VISA implementation for socket connections - you do not need any VISA-C installation

  • SelectVisa=rs - prefers usage of RohdeSchwarz VISA

  • SelectVisa=ni - prefers usage of National Instruments VISA

  • SelectVisa=pyvisa-py - prefers usage of python VISA backend pyvisa-py

  • Profile = HM8123 - setting profile fitting the specific non-standard instruments. Available values: HM8123, CMQ, ATS, Minimal. Default: none

  • OpenTimeout=5000 - sets timeout used at the session opening. This timeout is only used in waiting for a locked session to be freed. Default: 2000ms

  • ExclusiveLock=True - opens the session with exclusive lock on the VISA level. Default: False

  • QueryInstrumentStatus = False - same as driver.utilities.instrument_status_checking = False. Default: True

  • WriteDelay = 20, ReadDelay = 5 - introduces delay of 20ms before each write and 5ms before each read. Default: 0ms for both

  • TerminationCharacter = "\r" - sets the termination character for reading. Default: \n (LineFeed or LF)

  • AssureWriteWithTermChar = True - makes sure each command/query is terminated with termination character. Default: Interface dependent

  • AddTermCharToWriteBinBlock = True - adds one additional LF to the end of the binary data (some instruments require that). Default: False

  • DataChunkSize = 10E3 - maximum size of one write/read segment. If transferred data is bigger, it is split to more segments. Default: 1E7 bytes

  • OpcTimeout = 10000 - same as driver.utilities.opc_timeout = 10000. Default: 30000ms

  • VisaTimeout = 5000 - same as driver.utilities.visa_timeout = 5000. Default: 10000ms

  • ViClearExeMode = Disabled - viClear() execution mode. Default: execute_on_all

  • OpcQueryAfterWrite = True - same as driver.utilities.opc_query_after_write = True. Default: False

  • OpcWaitMode = OpcQuery - mode for all the opc-synchronized write/reads. Other modes: StbPolling, StbPollingSlow, StbPollingSuperSlow. Default: StbPolling

  • StbInErrorCheck = False - if true, the driver checks errors with *STB? If false, it uses SYST:ERR?. Default: True

  • SkipStatusSystemSettings = False - some instruments do not support full status system commands. In such case, set this value to True. Default: False

  • SkipClearStatus = True - set to True for instruments that do not support *CLS command. Default: False

  • DisableOpcQuery = True - set to True for instruments that do not support *OPC? query. Default: False

  • EachCmdAsQuery = True, set to True, for instruments that always return answer. Default: false

  • CmdIdn = ID? - defines which SCPI command to use for identification query. Use ‘<none>’ string to skip identification query at the init. Default: *IDN?

  • CmdReset = RT - defines which SCPI command to use for reset. Default: *RST

  • VxiCapable = false - you can force a session to a VXI-incapable. Default: <interface-dependent>

  • Encoding = utf-8 - setting of encoding for strings into bytes and vice versa. Default: charmap

  • OpcSyncQueryMechanism = AlsoCheckMav - setting of mechanism for OPC-synchronized queries. Default: OnlyCheckMavErrQueue

  • FirstCmds = *CLS - first command(s) to sent after init. Separated more commands/queries with ‘;;’. Default: (empty)

  • EachCmdPrefix = lf - this prefix is added to the beginning of each command sent to the instrument. Default: (empty)

  • EachCmdSuffix = cr - this suffix is added to the end of each command sent to the instrument. Default: (empty)

  • OptimizeOpcExecute = True - Only takes effect in the RsInstrument.execute() method. By default (False), the suffix ‘;*OPC’ triggers the _with_opc execution. If set to true, also the ‘;*OPC?’ suffix behaves the same way.

  • StripStringTrailingWhitespaces = True - use it to strip white spaces from string query responses. Default: False

  • LoggingMode = On - sets the logging status right from the start. Possible values: On | Off | Error. Default: Off

  • LoggingName = 'MyDevice' - sets the name to represent the session in the log entries. Default: <resource_name>

  • LoggingFormat = 'PAD_LEFT12(%START_TIME%) PAD_LEFT25(%DEVICE_NAME%) PAD_LEFT12(%DURATION%) %SCPI_COMMAND%' - sets the format of the log entries. Default: PAD_LEFT12(%START_TIME%) PAD_LEFT25(%DEVICE_NAME%) PAD_LEFT12(%DURATION%)  %LOG_STRING_INFO%: %LOG_STRING%

  • LogToGlobalTarget = True - sets the logging target to the class-property previously set with RsInstrument.set_global_logging_target() Default: False

  • LoggingToConsole = True - immediately starts logging to the console. Default: False

  • LoggingToUdp = True - immediately starts logging to the UDP port. Default: False

  • LoggingUdpPort = 49200 - UDP port to log to. Default: 49200

  • LoggingRelativeTimeOfFirstEntry = True - Logging starts with relative time set to the first log entry, which causes the first start time to be ‘00:00:00.000’. Default: False

add_instr_option(option: str) None[source]

Adds new option if not already existing.

static assert_minimum_version(min_version: str) None[source]

Asserts that the driver version fulfills the minimum required version you have entered. This way you make sure your installed driver is of the entered version or newer.

assign_lock(lock: RLock) None[source]

Assigns the provided thread lock.

property bin_float_numbers_format: BinFloatFormat

Sets / returns format of float numbers when transferred as binary data

property bin_int_numbers_format: BinIntFormat

Sets / returns format of integer numbers when transferred as binary data

check_status() None[source]

Throws InstrumentStatusException in case of an error in the instrument’s error queue. The status checking is performed always, independent of the property ‘instrument_status_checking’. Also, the property ScpiLogger.log_status_check_ok is ignored, and the Status check is always logged.

classmethod clear_global_logging_relative_timestamp() None[source]

Clears the global relative timestamp. After this, all the instances using the global relative timestamp continue logging with the absolute timestamps.

clear_lock() None[source]

Clears the existing thread lock, making the current session thread-independent from others that might share the current thread lock.

clear_status() None[source]

Clears instrument’s status system, the session’s I/O buffers and the instrument’s error queue

close() None[source]

Closes the active RsInstrument session

property data_chunk_size: int

Returns max chunk size of one data block.

static discovery(expression: str = '?*::INSTR', visa_select: str | None = None, identify: bool = False, identify_timeout_ms: int = 3000) DiscoverySnapshot[source]

Discover instruments visible to VISA.

Opens a fresh ResourceManager, calls list_resources, optionally identifies each instrument via *IDN? in parallel worker threads.

Parameters:
  • expression – VISA resource expression filter.

  • visa_select – VISA implementation selector (e.g. ‘rs’, ‘ni’, @py’).

  • identify – If True, open each resource briefly and query *IDN?.

  • identify_timeout_ms – Timeout per instrument for *IDN? query.

Returns:

Immutable DiscoverySnapshot with all found instruments.

property driver_version: str

Returns the RsInstrument package version

property encoding: str

Returns string<=>bytes encoding of the session.

property events: Events

Interface for event handlers, see here

execute(command: str) str | None[source]

Executes a single SCPI command string, auto-dispatching to the matching I/O method.

The command is stripped of surrounding whitespace and routed to one of write / query / write_with_opc / query_with_opc based on:

  • whether the (sent) command contains ? -> query, otherwise -> write, and

  • an optional trailing OPC synchronization suffix, interpreted according to the optimize_opc_execute property.

Handling of the trailing OPC suffix:

  • optimize_opc_execute = False (default): a trailing ;*OPC requests OPC-synchronized execution (write_with_opc / query_with_opc) and is

    stripped from the command. A trailing ;*OPC? is kept and the whole string is sent as an ordinary query.

  • optimize_opc_execute = True: a trailing ;*OPC? or ;*OPC requests OPC-synchronized execution instead; it is stripped and the base command is run OPC-synchronized.

Examples with the default optimize_opc_execute = False:

  • execute('*IDN?') -> query('*IDN?') -> response string

  • execute('*RST') -> write('*RST') -> None

  • execute('INIT;*OPC') -> write_with_opc('INIT') -> None

  • execute('READ?;*OPC') -> query_with_opc('READ?') -> response string

  • execute('READ?;*OPC?') -> query('READ?;*OPC?') -> response string

Parameters:

command – SCPI command string, optionally ending with an OPC synchronization suffix (;*OPC or ;*OPC?, see above).

Returns:

Query response as str when the command to send contains ?; None for write commands.

file_exists(instr_file: str) bool[source]

Returns true, if the instrument file exist.

classmethod from_existing_session(session: object, options: str | None = None) RsInstrument[source]

Creates a new RsInstrument object with the entered ‘session’ reused. :param session: can be another driver or a direct pyvisa session. :param options: string tokens alternating the driver settings. More tokens are separated by comma.

property full_instrument_model_name: str

Returns the current instrument’s full name e.g. ‘FSW26’

classmethod get_driver_version() str[source]

Returns the RsInstrument package version

get_file_size(instr_file: str) int | None[source]

Return size of the instrument file, or None if the file does not exist.

classmethod get_global_logging_relative_timestamp() datetime | None[source]

Returns global common relative timestamp for log entries.

classmethod get_global_logging_target()[source]

Returns global common target stream.

get_last_sent_cmd() str[source]

Returns the last commands sent to the instrument. Only works in simulation mode.

get_lock() RLock[source]

Returns the thread lock for the current session.

By default,

  • If you create a new RsInstrument instance with new VISA session, the session gets a new thread lock. You can assign it to another RsInstrument sessions in order to share one physical instrument with a multi-thread access.

  • If you create a new RsInstrument from an existing session, the thread lock is shared automatically making both instances multi-thread safe.

You can always assign new thread lock by calling driver.utilities.assign_lock()

get_session_handle()[source]

Returns the underlying pyvisa session

get_total_execution_time() timedelta[source]

Returns total time spent by the library on communicating with the instrument. This time is always shorter than get_total_time(), since it does not include gaps between the communication. You can reset this counter with reset_time_statistics().

get_total_time() timedelta[source]

Returns delta time spent by the library between the get_total_time_startpoint() and now. This time is always longer than get_total_execution_time(), since it also includes all other activities besides the communication. You can set the total time startpoint to now with reset_time_statistics().

get_total_time_startpoint() datetime[source]

Returns time from which the execution started. This is the value that the get_total_time() calculates as its reference. Calling the reset_time_statistics() sets this time to now.

go_to_local(mixed_mode: bool = True) None[source]

Puts the instrument into local state. By default, the method uses a mechanism to keep the instrument in a mixed mode: remote and local. That means, you can remote-control your instrument, and at the same time it still allows manual control. Set the mixed_mode to False, if you want your instrument to go to remote mode as soon as it receives the first remote command.

go_to_remote() None[source]

Puts the instrument into remote state.

has_instr_option(options: str | List[str]) bool[source]

Returns true, if the entered options (case-insensitive) matches at least one of the installed options (or-logic). You can enter either a string with one option, or more options ‘/’-separated, or more options as a list of strings. If K0 is present, all the K-options are reported as present. B-options are not affected by K0. Example 1: options=’k23’ returns true, if the instrument has the option ‘K23’. Example 2: options=’k23 / K23e’ returns true, if the instrument has either the option ‘K23’ or the option ‘K23E’. Example 3: options=[‘k11’,’K22’] returns true, if the instrument has either the option ‘K11’ or the option ‘K22’.

has_instr_option_k0() bool[source]

Returns true, if the instrument has K0 installed.

has_instr_option_regex(re_options: str | List[str]) bool[source]

Returns true, if the entered regex string (case-insensitive) matches at least one of the installed options. The match must be complete, not just partial (search). You can enter either a string with one option, or more options ‘/’-separated, or more options as a list of strings. Example 1: re_options=’k10.’ returns true, if the instrument contains any option ‘K100’ … up to ‘K109’ . Example 2: re_options=’k10. / k20.*’ returns true, if the instrument contains any of the options ‘K10x’ or ‘K20xxx’. Example 3: re_options=[‘k10.’, ‘k20.*’] returns true, if the instrument contains any options ‘K10x’ or ‘K20xxx’.

property idn_string: str

Returns instrument’s identification string - the response on the SCPI command *IDN?

instr_err_suppressor(visa_tout_ms: int = 0, suppress_only_codes: int | List[int] | None = None) InstrErrorSuppressor[source]

Returns Context Manager that suppresses the instrument errors. Other exceptions types are still raised. On entering the context, this class clears all the instrument status errors. :param visa_tout_ms: VISA Timeout in milliseconds, that is set for this context. Afterward, it is changed back. Default value: do-not-change. :param suppress_only_codes: You can enter a code or list of codes for errors to be suppressed. Other errors will be reported. Example: If you enter -113 here, only the ‘Undefined Header’ error will be suppressed. Default value: suppress-all-errors.

property instrument_firmware_version: str

Returns instrument’s firmware version

property instrument_model_name: str

Returns the current instrument’s family name e.g. ‘FSW’

property instrument_options: List[str]

Returns all the instrument options. The options are sorted in the ascending order starting with K-options and continuing with B-options

property instrument_serial_number: str

Returns instrument’s serial_number

property instrument_status_checking: bool

Sets / returns Instrument Status Checking. When True (default is True), all the driver methods and properties are sending “SYSTem:ERRor?” at the end to immediately react on error that might have occurred. We recommend keeping the state checking ON all the time. Switch it OFF only in rare cases when you require maximum speed. The default state after initializing the session is ON.

is_connection_active() bool[source]

Returns true, if the VISA connection is active and the communication with the instrument still works. WARNING!!! this method queries the session’s VISA Timeout and additionally, queries the *IDN? from the instrument, hence affects the performance of your application when used regularly.

property ivi_direct_io: IviDirectIo

Interface for IVI standard Direct IO, see here

property ivi_utility: IviUtility

Interface for IVI standard Utility, see here

static list_resources(expression: str = '?*::INSTR', visa_select: str | None = None) List[str][source]

Finds all the resources defined by the expression.

  • ‘?*’ - matches all the available instruments

  • ‘USB::?*’ - matches all the USB instruments

  • ‘TCPIP::192?*’ - matches all the LAN instruments with the IP address starting with 192

Parameters:
  • expression – see the examples in the function

  • visa_select – optional parameter selecting a specific VISA. Examples: @ivi’, @rs

lock_resource(timeout: int, requested_key: str | bytes | None = None) bytes | str | None[source]

Locks the instrument to prevent it from communicating with other clients.

property logger: ScpiLogger

SCPI Logger interface, see here

property manufacturer: str

Returns manufacturer of the instrument

property opc_query_after_write: bool

Sets / returns Instrument *OPC? query sending after each command write. When True, (default is False) the driver sends *OPC? every time a write command is performed. Use this if you want to make sure your sequence is performed command-after-command.

property opc_sync_query_mechanism: OpcSyncQueryMechanism

Returns the current setting of the OPC-Sync query mechanism.

property opc_timeout: int

Sets / returns timeout in milliseconds for all the operations that use OPC synchronization.

property optimize_opc_execute: bool

Sets / returns how the execute() method interprets a trailing OPC suffix. This flag only takes effect in the execute() method. - False (default): only a trailing ;*OPC triggers OPC-synchronized execution (write_with_opc / query_with_opc); a trailing ;*OPC? is sent as an ordinary query. - True: in addition to the ;*OPC, a trailing ;*OPC? triggers OPC-synchronized execution as well. Same as the OptimizeOpcExecute init option token. The default after initializing the session is False.

process_all_commands() None[source]

SCPI command: *WAI Stops further commands processing until all commands sent before *WAI have been executed.

query(query: str) str[source]

Sends the string query to the instrument and returns the response as string. The response is trimmed of any trailing LF characters and has no length limit. This method is an alias to the query_str() method.

query_all_errors() List[str] | None[source]

Queries and clears all the errors from the instrument’s error queue. The method returns list of strings as error messages. If no error is detected, the return value is None. The process is: querying ‘SYSTem:ERRor?’ in a loop until the error queue is empty. If you want to include the error codes, call the query_all_errors_with_codes()

Note: If status-error suppression rules are configured (see status_error_suppression_add_rule()), errors matching those rules are filtered out and will not appear in the returned list.

query_all_errors_with_codes() List[Tuple[int, str]] | None[source]

Queries and clears all the errors from the instrument’s error queue. The method returns list of tuples (code: int, message: str). If no error is detected, the return value is None. The process is: querying ‘SYSTem:ERRor?’ in a loop until the error queue is empty.

Note: If status-error suppression rules are configured (see status_error_suppression_add_rule()), errors matching those rules are filtered out and will not appear in the returned list.

query_bin_block(query: str) bytes[source]

Queries binary data block to bytes. Throws an exception if the returned data was not a binary data. Returns data:bytes

query_bin_block_to_file(query: str, file_path: str, append: bool = False) None[source]

Queries binary data block to the provided file. If append is False, any existing file content is discarded. If append is True, the new content is added to the end of the existing file, or if the file does not exit, it is created. Throws an exception if the returned data was not a binary data. Example for transferring a file from Instrument -> PC: query = f”MMEM:DATA? ‘{INSTR_FILE_PATH}’”.

Alternatively, use the dedicated methods for this purpose:

  • send_file_from_pc_to_instrument()

  • read_file_from_instrument_to_pc()

query_bin_block_to_file_with_opc(query: str, file_path: str, append: bool = False, timeout: int | None = None) None[source]

Sends a OPC-synced query and writes the returned data to the provided file. If append is False, any existing file content is discarded. If append is True, the new content is added to the end of the existing file, or if the file does not exit, it is created. Throws an exception if the returned data was not a binary data.

query_bin_block_with_opc(query: str, timeout: int | None = None) bytes[source]

Sends a OPC-synced query and returns binary data block to bytes. If you do not provide timeout, the method uses current opc_timeout.

query_bin_or_ascii_float_list(query: str) List[float][source]

Queries a list of floating-point numbers that can be returned as ASCII or binary format.

  • For ASCII format, the list numbers are decoded as comma-separated values.

  • For Binary Format, the numbers are decoded based on the property BinFloatFormat, usually float 32-bit (FORM REAL,32).

query_bin_or_ascii_float_list_with_opc(query: str, timeout: int | None = None) List[float][source]

Sends a OPC-synced query and reads a list of floating-point numbers that can be returned as ASCII or binary format.

  • For ASCII format, the list numbers are decoded as comma-separated values.

  • For Binary Format, the numbers are decoded based on the property BinFloatFormat, usually float 32-bit (FORM REAL,32).

If you do not provide timeout, the method uses current opc_timeout.

query_bin_or_ascii_int_list(query: str) List[int][source]

Queries a list of floating-point numbers that can be returned as ASCII or binary format.

  • For ASCII format, the list numbers are decoded as comma-separated values.

  • For Binary Format, the numbers are decoded based on the property BinFloatFormat, usually float 32-bit (FORM REAL,32).

query_bin_or_ascii_int_list_with_opc(query: str, timeout: int | None = None) List[int][source]

Sends a OPC-synced query and reads a list of floating-point numbers that can be returned as ASCII or binary format.

  • For ASCII format, the list numbers are decoded as comma-separated values.

  • For Binary Format, the numbers are decoded based on the property BinFloatFormat, usually float 32-bit (FORM REAL,32).

If you do not provide timeout, the method uses current opc_timeout.

query_bool(query: str) bool[source]

Sends the query to the instrument and returns the response as boolean.

query_bool_list(query: str) List[bool][source]

Sends the string query to the instrument and returns the response as List of booleans, where the delimiter is comma (‘,’). Blank or empty response is returned as an empty list.

query_bool_list_with_opc(query: str, timeout: int | None = None) List[bool][source]

Sends a OPC-synced query and reads response from the instrument as csv-list of booleans. If you do not provide timeout, the method uses current opc_timeout. Blank or empty response is returned as an empty list.

query_bool_with_opc(query: str, timeout: int | None = None) bool[source]

Sends the opc-synced query to the instrument and returns the response as boolean. If you do not provide timeout, the method uses current opc_timeout.

query_float(query: str) float[source]

Sends the query to the instrument and returns the response as float.

query_float_with_opc(query: str, timeout: int | None = None) float[source]

Sends the opc-synced query to the instrument and returns the response as float. If you do not provide timeout, the method uses current opc_timeout.

query_int(query: str) int[source]

Sends the query to the instrument and returns the response as integer.

query_int_with_opc(query: str, timeout: int | None = None) int[source]

Sends the opc-synced query to the instrument and returns the response as integer. If you do not provide timeout, the method uses current opc_timeout.

query_opc(timeout: int = 0) int[source]

SCPI command: *OPC? Queries the instrument’s OPC bit and hence it waits until the instrument reports operation complete. If you define timeout > 0, the VISA timeout is set to that value just for this method call.

query_str(query: str) str[source]

Sends the string query to the instrument and returns the response as string. The response is trimmed of any trailing LF characters and has no length limit. This method is an alias to the query() method.

query_str_list(query: str, remove_blank_response: bool = False) List[str][source]

Sends the string query to the instrument and returns the response as List of strings, where the delimiter is comma (‘,’). Each element of the list is trimmed for leading and trailing quotes.

Meaning of the ‘remove_blank_response’:
  • False(default): whitespaces-only response is returned as a list with one empty element [‘’].

  • True: whitespaces-only response is returned as an empty list [].

query_str_list_with_opc(query: str, timeout: int | None = None, remove_blank_response: bool = False) List[str][source]
Sends a OPC-synced query and reads response from the instrument as csv-list.

If you do not provide timeout, the method uses current opc_timeout.

Meaning of the ‘remove_blank_response’:
  • False(default): whitespaces-only response is returned as a list with one empty element [‘’].

  • True: whitespaces-only response is returned as an empty list [].

query_str_stripped(query: str) str[source]

Sends the string query to the instrument and returns the response as string stripped of the trailing LF and leading/trailing single/double quotes. The stripping of the leading/trailing quotes is blocked, if the string contains the quotes in the middle. This method is an alias to the query_stripped() method.

query_str_with_opc(query: str, timeout: int | None = None) str[source]

Sends the opc-synced query to the instrument and returns the response as string. The response is trimmed of any trailing LF characters and has no length limit. If you do not provide timeout, the method uses current opc_timeout.

query_stripped(query: str) str[source]

Sends the string query to the instrument and returns the response as string stripped of the trailing LF and leading/trailing single/double quotes. The stripping of the leading/trailing quotes is blocked, if the string contains the quotes in the middle.

query_with_opc(query: str, timeout: int | None = None) str[source]

This method is an alias to the write_str_with_opc(). Sends the opc-synced query to the instrument and returns the response as string. The response is trimmed of any trailing LF characters and has no length limit. If you do not provide timeout, the method uses current opc_timeout.

read_file_from_instrument_to_pc(source_instr_file: str, target_pc_file: str, append_to_pc_file: bool = False) None[source]

SCPI Command: MMEM:DATA?

Reads file from instrument to the PC.

Set the append_to_pc_file to True if you want to append the read content to the end of the existing PC file.

reconnect(force_close: bool = False) bool[source]

If the connection is not active, the method tries to reconnect to the device. If the connection is active, and force_close is False, the method does nothing. If the connection is active, and force_close is True, the method closes, and opens the session again. Returns True, if the reconnection has been performed.

remove_instr_option(option: str) None[source]

Removes the option if exists.

reset(timeout: int = 0) None[source]

SCPI command: *RST Sends *RST command + calls the clear_status(). If you define timeout > 0, the VISA timeout is set to that value just for this method call.

reset_time_statistics() None[source]

Resets all execution and total time counters. Affects the results of get_total_time(), get_total_execution_time() and get_total_time_startpoint()

property resource_name: str

Returns the resource name used in the constructor.

self_test(timeout: int | None = None) Tuple[int, str][source]

SCPI command: *TST? Performs instrument’s self-test. Returns tuple (code:int, message: str). Code 0 means the self-test passed. You can define the custom timeout in milliseconds. If you do not define it, the method uses default self-test timeout (usually 60 secs).

send_file_from_pc_to_instrument(source_pc_file: str, target_instr_file: str) None[source]

SCPI Command: MMEM:DATA

Sends file from PC to the instrument.

classmethod set_global_logging_relative_time_of_first_entry() None[source]

This method sets the global flag, that takes the very first log entry start time of any global instance as a global relative timestamp. This means, after this call, the first instance that logs an entry set the relative timestamp for all global target instances, and begins with the start timestamp 0:00:00.000

classmethod set_global_logging_relative_timestamp(timestamp: datetime) None[source]

Sets global common relative timestamp for log entries. To use it, call the following: io.logger.set_relative_timestamp_global()

classmethod set_global_logging_relative_timestamp_now() None[source]

Sets global common relative timestamp for log entries to this moment. To use it, call the following: io.logger.set_relative_timestamp_global().

classmethod set_global_logging_target(target) None[source]

Sets global common target stream that each instance can use. To use it, call the following: io.logger.set_logging_target_global(). If an instance uses global logging target, it automatically uses the global relative timestamp (if set). You can set the target to None in order to invalidate it.

status_error_suppression_add_rule(rule: Tuple[int, str | Pattern[str]] | str | Pattern[str] | int) InstrumentStatusErrorRule[source]

Adds one suppression rule and returns the created rule object.

The rule argument can be one of the following:

  • (code, pattern) tuple: suppress only when the error code equals code AND the message matches pattern (logical AND).

  • pattern (str or re.Pattern): suppress by message only, for any error code.

  • code (int): suppress by error code only, for any message.

  • an already-built InstrumentStatusErrorRule: stored as-is.

A pattern is either a regex source string or an already-compiled re.Pattern. Message matching uses re.search(), so the pattern only has to be found somewhere in the message, not match it in full.

Rules are de-duplicated by value: adding a rule that is equal to one already in the list (same code and/or same pattern source string and flags) does not create a second entry, and the existing/normalized rule is returned.

Parameters:

rule – The rule definition, in any of the forms listed above.

Returns:

The normalized InstrumentStatusErrorRule that was added (or the equal one already present). Keep this reference to remove the rule later with status_error_suppression_remove_rule().

Raises:
  • TypeError – If rule (or a tuple’s code/pattern) has an unsupported type.

  • ValueError – If a tuple rule does not have exactly two elements.

status_error_suppression_add_rules(rules: Iterable[Tuple[int, str | Pattern[str]] | str | Pattern[str] | int]) None[source]

Same as the status_error_suppression_add_rule(), but the parameter is a collection of rules.

status_error_suppression_clear_all_rules() None[source]

Clears the status-error suppression rules - all instrument errors are reported.

status_error_suppression_get_rules() list[InstrumentStatusErrorRule][source]

Returns the status-error suppression rules used by automatic status checks, or query_all_errors / query_all_errors_with_codes.

status_error_suppression_remove_rule(rule: InstrumentStatusErrorRule) bool[source]

Removes the entered rule from the error suppression list. Returns true, if the rule existed in the list.

property supported_models: List[str]

Returns a list of the instrument models supported by this instrument driver

unlock_resource() None[source]

Unlocks the instrument to other clients.

property visa_manufacturer: str

Returns the manufacturer of the current VISA session.

property visa_timeout: int

Sets / returns visa IO timeout in milliseconds.

visa_tout_suppressor(visa_tout_ms: int = 0) VisaTimeoutSuppressor[source]

Returns Context Manager that suppresses the VISA timeout error. Careful!!!: Only the very first VISA Timeout exception is suppressed, and afterward the context ends. Therefore, use only one command per context manager, if you do not want to skip the following ones. :param visa_tout_ms: VISA Timeout in milliseconds, that is set for this context. Afterward, it is changed back. Default value: do-not-change.

write(cmd: str) None[source]

Writes the command to the instrument as string. This method is an alias to the write_str() method.

write_bin_block(cmd: str, payload: bytes) None[source]

Writes all the payload as binary data block to the instrument. The binary data header is added at the beginning of the transmission automatically, do not include it in the payload!!!

write_bin_block_from_file(cmd: str, file_path: str) None[source]

Writes data from the file as binary data block to the instrument using the provided command. Example for transferring a file from PC -> Instrument: cmd = f”MMEM:DATA ‘{INSTR_FILE_PATH}’,”.

Alternatively, use the dedicated methods for this purpose:

  • send_file_from_pc_to_instrument()

  • read_file_from_instrument_to_pc()

write_bool(cmd: str, param: bool) None[source]

Writes the command to the instrument followed by the boolean parameter: e.g.: cmd = ‘OUTPUT’ param = ‘True’, result command = ‘OUTPUT ON’

write_bool_with_opc(cmd: str, param: bool, timeout: int | None = None) None[source]

Writes the command with OPC to the instrument followed by the boolean parameter: e.g.: cmd = ‘OUTPUT’ param = ‘True’, result command = ‘OUTPUT ON’ If you do not provide timeout, the method uses current opc_timeout.

write_float(cmd: str, param: float) None[source]

Writes the command to the instrument followed by the boolean parameter: e.g.: cmd = ‘CENTER:FREQ’ param = ‘10E6’, result command = ‘CENTER:FREQ 10E6’

write_float_with_opc(cmd: str, param: float, timeout: int | None = None) None[source]

Writes the command with OPC to the instrument followed by the boolean parameter: e.g.: cmd = ‘CENTER:FREQ’ param = ‘10E6’, result command = ‘CENTER:FREQ 10E6’ If you do not provide timeout, the method uses current opc_timeout.

write_int(cmd: str, param: int) None[source]

Writes the command to the instrument followed by the integer parameter: e.g.: cmd = ‘SELECT:INPUT’ param = ‘2’, result command = ‘SELECT:INPUT 2’

write_int_with_opc(cmd: str, param: int, timeout: int | None = None) None[source]

Writes the command with OPC to the instrument followed by the integer parameter: e.g.: cmd = ‘SELECT:INPUT’ param = ‘2’, result command = ‘SELECT:INPUT 2’ If you do not provide timeout, the method uses current opc_timeout.

write_str(cmd: str) None[source]

Writes the command to the instrument as string. This method is an alias to write() method.

write_str_with_opc(cmd: str, timeout: int | None = None) None[source]

Writes the opc-synced command to the instrument. If you do not provide timeout, the method uses current opc_timeout.

write_with_opc(cmd: str, timeout: int | None = None) None[source]

This method is an alias to the write_str_with_opc(). Writes the opc-synced command to the instrument. If you do not provide timeout, the method uses current opc_timeout.

Module contents

VISA communication interface for SCPI-based instrument remote control. :version: 1.131.0 :copyright: 2025 by Rohde & Schwarz GMBH & Co. KG :license: MIT, see LICENSE for more details.

class BinFloatFormat(value)[source]

Bases: Enum

Binary format of a float number.

Double_8bytes = 3
Double_8bytes_swapped = 4
Single_4bytes = 1
Single_4bytes_swapped = 2
class BinIntFormat(value)[source]

Bases: Enum

Binary format of an integer number.

Integer16_2bytes = 3
Integer16_2bytes_swapped = 4
Integer32_4bytes = 1
Integer32_4bytes_swapped = 2
class DiscoveredInstrument(resource: str, identified: bool = False, manufacturer: str = '', model: str = '', serial: str = '', firmware: str = '', idn_raw: str = '', ip: str = '', hostname: str = '', services: tuple[str, ...] = (), discovery_source: str = '', notes: tuple[str, ...] = ())[source]

Bases: object

One discovered instrument.

discovery_source: str = ''
firmware: str = ''
hostname: str = ''
identified: bool = False
idn_raw: str = ''
ip: str = ''
manufacturer: str = ''
model: str = ''
notes: tuple[str, ...] = ()
resource: str
serial: str = ''
services: tuple[str, ...] = ()
class Discovery(expression: str = '?*::INSTR', visa_select: str | None = None, identify: bool = False, identify_timeout_ms: int = 3000, identify_workers: int = 8, ttl_s: float = 30.0, cache: DiscoveryCacheStore | None = None, *, enrich_lxi: bool = False, enrich_dns: bool = False, mdns_timeout_s: float = 0.0, lxi_timeout_s: float = 2.0, dns_timeout_s: float = 2.0, enrich_workers: int = 8)[source]

Bases: object

Pull-based instrument discovery with TTL cache.

No background threads — cache is refreshed on demand via get(). Caches DiscoverySnapshot results only; never holds RsInstrument sessions.

Configure discovery parameters, TTL, enrichment, and cache backend.

clear_cache() None[source]

Invalidate the store entry without scanning.

get(*, use_cache: bool = True) DiscoveryResult[source]

Return cached snapshot if fresh; else live-scan, save, return.

refresh() DiscoveryResult[source]

Force live scan and update cache.

property snapshot: DiscoverySnapshot | None

Last stored snapshot if present (ignores TTL). None if never scanned.

class DiscoveryCacheStore(*args, **kwargs)[source]

Bases: Protocol

Pluggable cache backend for Discovery.

Thread-safety contract: load / save / delete are safe for concurrent calls from multiple threads in the same process.

delete(key: str) None[source]

Remove the entry for key. No-op if not present.

load(key: str) tuple[DiscoverySnapshot, float] | None[source]

Return (snapshot, age_seconds) or None if not cached.

save(key: str, snapshot: DiscoverySnapshot) None[source]

Persist snapshot under key. Store records its own timestamp.

class DiscoveryResult(snapshot: DiscoverySnapshot, source: Literal['cache', 'live'])[source]

Bases: object

Return type of Discovery.get() and Discovery.refresh().

snapshot: DiscoverySnapshot
source: Literal['cache', 'live']
class DiscoverySnapshot(instruments: tuple[~RsInstrument.Internal.Discovery.DiscoveredInstrument, ...]=(), timestamp: datetime = <factory>, expression: str = '?*::INSTR', visa_select: str | None = None, identify: bool = False, method_notes: dict[str, str]=<factory>)[source]

Bases: object

Immutable point-in-time discovery result.

expression: str = '?*::INSTR'
filter(model: str = '', manufacturer: str = '') tuple[DiscoveredInstrument, ...][source]

Filter instruments by model/manufacturer substring (case-insensitive).

Returns an empty tuple when called on a snapshot where identify=False was used, because model/manufacturer fields are empty strings. This is intentional — callers should use identify=True before filtering.

identify: bool = False
instruments: tuple[DiscoveredInstrument, ...] = ()
method_notes: dict[str, str]
timestamp: datetime
visa_select: str | None = None
exception DriverValueError(rsrc_name: str, message: str)[source]

Bases: RsInstrException

Exception for different driver value settings e.g. RepCap values or Enum values.

class FileDiscoveryCache(path: str | Path)[source]

Bases: object

JSON file cache using UTC wall clock for cross-process age tracking.

One file = one cache entry. No merge — each save fully replaces the file.

Create a file-backed cache at path (one entry per file).

delete(key: str) None[source]

Delete the cache file. No-op if missing.

load(key: str) tuple[DiscoverySnapshot, float] | None[source]

Return (snapshot, age_seconds) or None if missing/invalid.

save(key: str, snapshot: DiscoverySnapshot) None[source]

Atomically replace the cache file with snapshot for key.

class InMemoryDiscoveryCache[source]

Bases: object

Process-local cache using time.monotonic() for age tracking.

Create an empty in-memory cache.

delete(key: str) None[source]

Remove the entry for key. No-op if not present.

load(key: str) tuple[DiscoverySnapshot, float] | None[source]

Return (snapshot, age_seconds) or None if not cached.

save(key: str, snapshot: DiscoverySnapshot) None[source]

Persist snapshot under key using monotonic time.

class IoTransferEventArgs(reading: bool, opc_sync: bool, total_size: int | None, context: str)[source]

Bases: object

Contains event data for driver read or write operations.

Initializes new instance of IoTransferEventArgs :param reading: True: reading operation, False: writing operation :param opc_sync: defines if the command is OPC-synchronized :param total_size: total size of the data received :param context: SCPI query. It is truncated to maximum of 100 characters

binary: bool | None

string data

Type:

True

Type:

Binary data, False

chunk_ix: int | None

0-based index of the chunk.

chunk_size: int | None

Size of one chunk of data. This number does not change during the transfer.

data: str | bytes | None

If the feature of transferring data over R/W event is switched ON, this field contains the whole data.

id_generator = count(100)
classmethod read_chunk(opc_sync: bool, context: str) IoTransferEventArgs[source]

Creates new IoTransferEventArgs of read string

Parameters:
  • opc_sync – defines if the command is OPC-synchronised

  • context – SCPI query. It is truncated to maximum of 100 characters.

Returns:

IoTransferEventArgs object of a read string operation.

resource_name: str | None

Visa Resource Name of the instrument that generated the data.

set_end_of_transfer()[source]

Sets fields to signal end of transfer.

total_chunks: int | None

Expected number of chunks.

property transfer_id: int

Unique number for each transfer for this Instrument. If the transfer is performed in more chunks, the transfer_id stays the same during the whole transfer.

classmethod write_bin(context: str) IoTransferEventArgs[source]

Creates new IoTransferEventArgs of read binary data

Parameters:

context – SCPI command. It is truncated to maximum of 100 characters.

Returns:

IoTransferEventArgs object of a write-binary-data operation.

classmethod write_str(opc_sync: bool, total_size: int, context: str) IoTransferEventArgs[source]

Creates new IoTransferEventArgs of write string

Parameters:
  • opc_sync – defines if the command is OPC-synchronized

  • total_size – size of the data to write

  • context – SCPI command write. It is truncated to maximum of 100 characters

Returns:

IoTransferEventArgs object of a write-string operation.

class LoggingMode(value)[source]

Bases: Enum

Determines the format of the logging message.

Default = 3
Errors = 2
Off = 0
On = 1
class OpcSyncQueryMechanism(value)[source]

Bases: Enum

Mechanism to use when querying with OPC.

also_check_mav = 1
cls_only_check_mav_err_queue = 2
only_check_mav_err_queue = 3
standard = 0
exception ResourceError(rsrc_name: str, message: str)[source]

Bases: RsInstrException

Exception for resource name - e.g. resource not found.

exception RsInstrException(message: str)[source]

Bases: Exception

Exception base class for all the RsInstrument exceptions.

class RsInstrument(resource_name: str, id_query: bool = True, reset: bool = False, options: str | None = None, direct_session: object | None = None)[source]

Bases: object

Root class for remote-controlling instrument with SCPI commands.

Initializes new RsInstrument session.

Parameters:
  • resource_name – VISA resource name, e.g. ‘TCPIP::192.168.2.1::INSTR’

  • id_query – if True, the instrument’s model name is verified against the models supported by the driver and eventually throws an exception

  • reset – Resets the instrument (sends *RST) command and clears its status syb-system

  • direct_session – Another driver object or pyVisa object to reuse the session instead of opening a new session

  • options – string tokens alternating the driver settings. More tokens are separated by comma.

Parameter options tokens examples:
  • Simulate=True - starts the session in simulation mode. Default: False

  • SelectVisa=socketio - uses no VISA implementation for socket connections - you do not need any VISA-C installation

  • SelectVisa=rs - prefers usage of RohdeSchwarz VISA

  • SelectVisa=ni - prefers usage of National Instruments VISA

  • SelectVisa=pyvisa-py - prefers usage of python VISA backend pyvisa-py

  • Profile = HM8123 - setting profile fitting the specific non-standard instruments. Available values: HM8123, CMQ, ATS, Minimal. Default: none

  • OpenTimeout=5000 - sets timeout used at the session opening. This timeout is only used in waiting for a locked session to be freed. Default: 2000ms

  • ExclusiveLock=True - opens the session with exclusive lock on the VISA level. Default: False

  • QueryInstrumentStatus = False - same as driver.utilities.instrument_status_checking = False. Default: True

  • WriteDelay = 20, ReadDelay = 5 - introduces delay of 20ms before each write and 5ms before each read. Default: 0ms for both

  • TerminationCharacter = "\r" - sets the termination character for reading. Default: \n (LineFeed or LF)

  • AssureWriteWithTermChar = True - makes sure each command/query is terminated with termination character. Default: Interface dependent

  • AddTermCharToWriteBinBlock = True - adds one additional LF to the end of the binary data (some instruments require that). Default: False

  • DataChunkSize = 10E3 - maximum size of one write/read segment. If transferred data is bigger, it is split to more segments. Default: 1E7 bytes

  • OpcTimeout = 10000 - same as driver.utilities.opc_timeout = 10000. Default: 30000ms

  • VisaTimeout = 5000 - same as driver.utilities.visa_timeout = 5000. Default: 10000ms

  • ViClearExeMode = Disabled - viClear() execution mode. Default: execute_on_all

  • OpcQueryAfterWrite = True - same as driver.utilities.opc_query_after_write = True. Default: False

  • OpcWaitMode = OpcQuery - mode for all the opc-synchronized write/reads. Other modes: StbPolling, StbPollingSlow, StbPollingSuperSlow. Default: StbPolling

  • StbInErrorCheck = False - if true, the driver checks errors with *STB? If false, it uses SYST:ERR?. Default: True

  • SkipStatusSystemSettings = False - some instruments do not support full status system commands. In such case, set this value to True. Default: False

  • SkipClearStatus = True - set to True for instruments that do not support *CLS command. Default: False

  • DisableOpcQuery = True - set to True for instruments that do not support *OPC? query. Default: False

  • EachCmdAsQuery = True, set to True, for instruments that always return answer. Default: false

  • CmdIdn = ID? - defines which SCPI command to use for identification query. Use ‘<none>’ string to skip identification query at the init. Default: *IDN?

  • CmdReset = RT - defines which SCPI command to use for reset. Default: *RST

  • VxiCapable = false - you can force a session to a VXI-incapable. Default: <interface-dependent>

  • Encoding = utf-8 - setting of encoding for strings into bytes and vice versa. Default: charmap

  • OpcSyncQueryMechanism = AlsoCheckMav - setting of mechanism for OPC-synchronized queries. Default: OnlyCheckMavErrQueue

  • FirstCmds = *CLS - first command(s) to sent after init. Separated more commands/queries with ‘;;’. Default: (empty)

  • EachCmdPrefix = lf - this prefix is added to the beginning of each command sent to the instrument. Default: (empty)

  • EachCmdSuffix = cr - this suffix is added to the end of each command sent to the instrument. Default: (empty)

  • OptimizeOpcExecute = True - Only takes effect in the RsInstrument.execute() method. By default (False), the suffix ‘;*OPC’ triggers the _with_opc execution. If set to true, also the ‘;*OPC?’ suffix behaves the same way.

  • StripStringTrailingWhitespaces = True - use it to strip white spaces from string query responses. Default: False

  • LoggingMode = On - sets the logging status right from the start. Possible values: On | Off | Error. Default: Off

  • LoggingName = 'MyDevice' - sets the name to represent the session in the log entries. Default: <resource_name>

  • LoggingFormat = 'PAD_LEFT12(%START_TIME%) PAD_LEFT25(%DEVICE_NAME%) PAD_LEFT12(%DURATION%) %SCPI_COMMAND%' - sets the format of the log entries. Default: PAD_LEFT12(%START_TIME%) PAD_LEFT25(%DEVICE_NAME%) PAD_LEFT12(%DURATION%)  %LOG_STRING_INFO%: %LOG_STRING%

  • LogToGlobalTarget = True - sets the logging target to the class-property previously set with RsInstrument.set_global_logging_target() Default: False

  • LoggingToConsole = True - immediately starts logging to the console. Default: False

  • LoggingToUdp = True - immediately starts logging to the UDP port. Default: False

  • LoggingUdpPort = 49200 - UDP port to log to. Default: 49200

  • LoggingRelativeTimeOfFirstEntry = True - Logging starts with relative time set to the first log entry, which causes the first start time to be ‘00:00:00.000’. Default: False

add_instr_option(option: str) None[source]

Adds new option if not already existing.

static assert_minimum_version(min_version: str) None[source]

Asserts that the driver version fulfills the minimum required version you have entered. This way you make sure your installed driver is of the entered version or newer.

assign_lock(lock: RLock) None[source]

Assigns the provided thread lock.

property bin_float_numbers_format: BinFloatFormat

Sets / returns format of float numbers when transferred as binary data

property bin_int_numbers_format: BinIntFormat

Sets / returns format of integer numbers when transferred as binary data

check_status() None[source]

Throws InstrumentStatusException in case of an error in the instrument’s error queue. The status checking is performed always, independent of the property ‘instrument_status_checking’. Also, the property ScpiLogger.log_status_check_ok is ignored, and the Status check is always logged.

classmethod clear_global_logging_relative_timestamp() None[source]

Clears the global relative timestamp. After this, all the instances using the global relative timestamp continue logging with the absolute timestamps.

clear_lock() None[source]

Clears the existing thread lock, making the current session thread-independent from others that might share the current thread lock.

clear_status() None[source]

Clears instrument’s status system, the session’s I/O buffers and the instrument’s error queue

close() None[source]

Closes the active RsInstrument session

property data_chunk_size: int

Returns max chunk size of one data block.

static discovery(expression: str = '?*::INSTR', visa_select: str | None = None, identify: bool = False, identify_timeout_ms: int = 3000) DiscoverySnapshot[source]

Discover instruments visible to VISA.

Opens a fresh ResourceManager, calls list_resources, optionally identifies each instrument via *IDN? in parallel worker threads.

Parameters:
  • expression – VISA resource expression filter.

  • visa_select – VISA implementation selector (e.g. ‘rs’, ‘ni’, @py’).

  • identify – If True, open each resource briefly and query *IDN?.

  • identify_timeout_ms – Timeout per instrument for *IDN? query.

Returns:

Immutable DiscoverySnapshot with all found instruments.

property driver_version: str

Returns the RsInstrument package version

property encoding: str

Returns string<=>bytes encoding of the session.

property events: Events

Interface for event handlers, see here

execute(command: str) str | None[source]

Executes a single SCPI command string, auto-dispatching to the matching I/O method.

The command is stripped of surrounding whitespace and routed to one of write / query / write_with_opc / query_with_opc based on:

  • whether the (sent) command contains ? -> query, otherwise -> write, and

  • an optional trailing OPC synchronization suffix, interpreted according to the optimize_opc_execute property.

Handling of the trailing OPC suffix:

  • optimize_opc_execute = False (default): a trailing ;*OPC requests OPC-synchronized execution (write_with_opc / query_with_opc) and is

    stripped from the command. A trailing ;*OPC? is kept and the whole string is sent as an ordinary query.

  • optimize_opc_execute = True: a trailing ;*OPC? or ;*OPC requests OPC-synchronized execution instead; it is stripped and the base command is run OPC-synchronized.

Examples with the default optimize_opc_execute = False:

  • execute('*IDN?') -> query('*IDN?') -> response string

  • execute('*RST') -> write('*RST') -> None

  • execute('INIT;*OPC') -> write_with_opc('INIT') -> None

  • execute('READ?;*OPC') -> query_with_opc('READ?') -> response string

  • execute('READ?;*OPC?') -> query('READ?;*OPC?') -> response string

Parameters:

command – SCPI command string, optionally ending with an OPC synchronization suffix (;*OPC or ;*OPC?, see above).

Returns:

Query response as str when the command to send contains ?; None for write commands.

file_exists(instr_file: str) bool[source]

Returns true, if the instrument file exist.

classmethod from_existing_session(session: object, options: str | None = None) RsInstrument[source]

Creates a new RsInstrument object with the entered ‘session’ reused. :param session: can be another driver or a direct pyvisa session. :param options: string tokens alternating the driver settings. More tokens are separated by comma.

property full_instrument_model_name: str

Returns the current instrument’s full name e.g. ‘FSW26’

classmethod get_driver_version() str[source]

Returns the RsInstrument package version

get_file_size(instr_file: str) int | None[source]

Return size of the instrument file, or None if the file does not exist.

classmethod get_global_logging_relative_timestamp() datetime | None[source]

Returns global common relative timestamp for log entries.

classmethod get_global_logging_target()[source]

Returns global common target stream.

get_last_sent_cmd() str[source]

Returns the last commands sent to the instrument. Only works in simulation mode.

get_lock() RLock[source]

Returns the thread lock for the current session.

By default,

  • If you create a new RsInstrument instance with new VISA session, the session gets a new thread lock. You can assign it to another RsInstrument sessions in order to share one physical instrument with a multi-thread access.

  • If you create a new RsInstrument from an existing session, the thread lock is shared automatically making both instances multi-thread safe.

You can always assign new thread lock by calling driver.utilities.assign_lock()

get_session_handle()[source]

Returns the underlying pyvisa session

get_total_execution_time() timedelta[source]

Returns total time spent by the library on communicating with the instrument. This time is always shorter than get_total_time(), since it does not include gaps between the communication. You can reset this counter with reset_time_statistics().

get_total_time() timedelta[source]

Returns delta time spent by the library between the get_total_time_startpoint() and now. This time is always longer than get_total_execution_time(), since it also includes all other activities besides the communication. You can set the total time startpoint to now with reset_time_statistics().

get_total_time_startpoint() datetime[source]

Returns time from which the execution started. This is the value that the get_total_time() calculates as its reference. Calling the reset_time_statistics() sets this time to now.

go_to_local(mixed_mode: bool = True) None[source]

Puts the instrument into local state. By default, the method uses a mechanism to keep the instrument in a mixed mode: remote and local. That means, you can remote-control your instrument, and at the same time it still allows manual control. Set the mixed_mode to False, if you want your instrument to go to remote mode as soon as it receives the first remote command.

go_to_remote() None[source]

Puts the instrument into remote state.

has_instr_option(options: str | List[str]) bool[source]

Returns true, if the entered options (case-insensitive) matches at least one of the installed options (or-logic). You can enter either a string with one option, or more options ‘/’-separated, or more options as a list of strings. If K0 is present, all the K-options are reported as present. B-options are not affected by K0. Example 1: options=’k23’ returns true, if the instrument has the option ‘K23’. Example 2: options=’k23 / K23e’ returns true, if the instrument has either the option ‘K23’ or the option ‘K23E’. Example 3: options=[‘k11’,’K22’] returns true, if the instrument has either the option ‘K11’ or the option ‘K22’.

has_instr_option_k0() bool[source]

Returns true, if the instrument has K0 installed.

has_instr_option_regex(re_options: str | List[str]) bool[source]

Returns true, if the entered regex string (case-insensitive) matches at least one of the installed options. The match must be complete, not just partial (search). You can enter either a string with one option, or more options ‘/’-separated, or more options as a list of strings. Example 1: re_options=’k10.’ returns true, if the instrument contains any option ‘K100’ … up to ‘K109’ . Example 2: re_options=’k10. / k20.*’ returns true, if the instrument contains any of the options ‘K10x’ or ‘K20xxx’. Example 3: re_options=[‘k10.’, ‘k20.*’] returns true, if the instrument contains any options ‘K10x’ or ‘K20xxx’.

property idn_string: str

Returns instrument’s identification string - the response on the SCPI command *IDN?

instr_err_suppressor(visa_tout_ms: int = 0, suppress_only_codes: int | List[int] | None = None) InstrErrorSuppressor[source]

Returns Context Manager that suppresses the instrument errors. Other exceptions types are still raised. On entering the context, this class clears all the instrument status errors. :param visa_tout_ms: VISA Timeout in milliseconds, that is set for this context. Afterward, it is changed back. Default value: do-not-change. :param suppress_only_codes: You can enter a code or list of codes for errors to be suppressed. Other errors will be reported. Example: If you enter -113 here, only the ‘Undefined Header’ error will be suppressed. Default value: suppress-all-errors.

property instrument_firmware_version: str

Returns instrument’s firmware version

property instrument_model_name: str

Returns the current instrument’s family name e.g. ‘FSW’

property instrument_options: List[str]

Returns all the instrument options. The options are sorted in the ascending order starting with K-options and continuing with B-options

property instrument_serial_number: str

Returns instrument’s serial_number

property instrument_status_checking: bool

Sets / returns Instrument Status Checking. When True (default is True), all the driver methods and properties are sending “SYSTem:ERRor?” at the end to immediately react on error that might have occurred. We recommend keeping the state checking ON all the time. Switch it OFF only in rare cases when you require maximum speed. The default state after initializing the session is ON.

is_connection_active() bool[source]

Returns true, if the VISA connection is active and the communication with the instrument still works. WARNING!!! this method queries the session’s VISA Timeout and additionally, queries the *IDN? from the instrument, hence affects the performance of your application when used regularly.

property ivi_direct_io: IviDirectIo

Interface for IVI standard Direct IO, see here

property ivi_utility: IviUtility

Interface for IVI standard Utility, see here

static list_resources(expression: str = '?*::INSTR', visa_select: str | None = None) List[str][source]

Finds all the resources defined by the expression.

  • ‘?*’ - matches all the available instruments

  • ‘USB::?*’ - matches all the USB instruments

  • ‘TCPIP::192?*’ - matches all the LAN instruments with the IP address starting with 192

Parameters:
  • expression – see the examples in the function

  • visa_select – optional parameter selecting a specific VISA. Examples: @ivi’, @rs

lock_resource(timeout: int, requested_key: str | bytes | None = None) bytes | str | None[source]

Locks the instrument to prevent it from communicating with other clients.

property logger: ScpiLogger

SCPI Logger interface, see here

property manufacturer: str

Returns manufacturer of the instrument

property opc_query_after_write: bool

Sets / returns Instrument *OPC? query sending after each command write. When True, (default is False) the driver sends *OPC? every time a write command is performed. Use this if you want to make sure your sequence is performed command-after-command.

property opc_sync_query_mechanism: OpcSyncQueryMechanism

Returns the current setting of the OPC-Sync query mechanism.

property opc_timeout: int

Sets / returns timeout in milliseconds for all the operations that use OPC synchronization.

property optimize_opc_execute: bool

Sets / returns how the execute() method interprets a trailing OPC suffix. This flag only takes effect in the execute() method. - False (default): only a trailing ;*OPC triggers OPC-synchronized execution (write_with_opc / query_with_opc); a trailing ;*OPC? is sent as an ordinary query. - True: in addition to the ;*OPC, a trailing ;*OPC? triggers OPC-synchronized execution as well. Same as the OptimizeOpcExecute init option token. The default after initializing the session is False.

process_all_commands() None[source]

SCPI command: *WAI Stops further commands processing until all commands sent before *WAI have been executed.

query(query: str) str[source]

Sends the string query to the instrument and returns the response as string. The response is trimmed of any trailing LF characters and has no length limit. This method is an alias to the query_str() method.

query_all_errors() List[str] | None[source]

Queries and clears all the errors from the instrument’s error queue. The method returns list of strings as error messages. If no error is detected, the return value is None. The process is: querying ‘SYSTem:ERRor?’ in a loop until the error queue is empty. If you want to include the error codes, call the query_all_errors_with_codes()

Note: If status-error suppression rules are configured (see status_error_suppression_add_rule()), errors matching those rules are filtered out and will not appear in the returned list.

query_all_errors_with_codes() List[Tuple[int, str]] | None[source]

Queries and clears all the errors from the instrument’s error queue. The method returns list of tuples (code: int, message: str). If no error is detected, the return value is None. The process is: querying ‘SYSTem:ERRor?’ in a loop until the error queue is empty.

Note: If status-error suppression rules are configured (see status_error_suppression_add_rule()), errors matching those rules are filtered out and will not appear in the returned list.

query_bin_block(query: str) bytes[source]

Queries binary data block to bytes. Throws an exception if the returned data was not a binary data. Returns data:bytes

query_bin_block_to_file(query: str, file_path: str, append: bool = False) None[source]

Queries binary data block to the provided file. If append is False, any existing file content is discarded. If append is True, the new content is added to the end of the existing file, or if the file does not exit, it is created. Throws an exception if the returned data was not a binary data. Example for transferring a file from Instrument -> PC: query = f”MMEM:DATA? ‘{INSTR_FILE_PATH}’”.

Alternatively, use the dedicated methods for this purpose:

  • send_file_from_pc_to_instrument()

  • read_file_from_instrument_to_pc()

query_bin_block_to_file_with_opc(query: str, file_path: str, append: bool = False, timeout: int | None = None) None[source]

Sends a OPC-synced query and writes the returned data to the provided file. If append is False, any existing file content is discarded. If append is True, the new content is added to the end of the existing file, or if the file does not exit, it is created. Throws an exception if the returned data was not a binary data.

query_bin_block_with_opc(query: str, timeout: int | None = None) bytes[source]

Sends a OPC-synced query and returns binary data block to bytes. If you do not provide timeout, the method uses current opc_timeout.

query_bin_or_ascii_float_list(query: str) List[float][source]

Queries a list of floating-point numbers that can be returned as ASCII or binary format.

  • For ASCII format, the list numbers are decoded as comma-separated values.

  • For Binary Format, the numbers are decoded based on the property BinFloatFormat, usually float 32-bit (FORM REAL,32).

query_bin_or_ascii_float_list_with_opc(query: str, timeout: int | None = None) List[float][source]

Sends a OPC-synced query and reads a list of floating-point numbers that can be returned as ASCII or binary format.

  • For ASCII format, the list numbers are decoded as comma-separated values.

  • For Binary Format, the numbers are decoded based on the property BinFloatFormat, usually float 32-bit (FORM REAL,32).

If you do not provide timeout, the method uses current opc_timeout.

query_bin_or_ascii_int_list(query: str) List[int][source]

Queries a list of floating-point numbers that can be returned as ASCII or binary format.

  • For ASCII format, the list numbers are decoded as comma-separated values.

  • For Binary Format, the numbers are decoded based on the property BinFloatFormat, usually float 32-bit (FORM REAL,32).

query_bin_or_ascii_int_list_with_opc(query: str, timeout: int | None = None) List[int][source]

Sends a OPC-synced query and reads a list of floating-point numbers that can be returned as ASCII or binary format.

  • For ASCII format, the list numbers are decoded as comma-separated values.

  • For Binary Format, the numbers are decoded based on the property BinFloatFormat, usually float 32-bit (FORM REAL,32).

If you do not provide timeout, the method uses current opc_timeout.

query_bool(query: str) bool[source]

Sends the query to the instrument and returns the response as boolean.

query_bool_list(query: str) List[bool][source]

Sends the string query to the instrument and returns the response as List of booleans, where the delimiter is comma (‘,’). Blank or empty response is returned as an empty list.

query_bool_list_with_opc(query: str, timeout: int | None = None) List[bool][source]

Sends a OPC-synced query and reads response from the instrument as csv-list of booleans. If you do not provide timeout, the method uses current opc_timeout. Blank or empty response is returned as an empty list.

query_bool_with_opc(query: str, timeout: int | None = None) bool[source]

Sends the opc-synced query to the instrument and returns the response as boolean. If you do not provide timeout, the method uses current opc_timeout.

query_float(query: str) float[source]

Sends the query to the instrument and returns the response as float.

query_float_with_opc(query: str, timeout: int | None = None) float[source]

Sends the opc-synced query to the instrument and returns the response as float. If you do not provide timeout, the method uses current opc_timeout.

query_int(query: str) int[source]

Sends the query to the instrument and returns the response as integer.

query_int_with_opc(query: str, timeout: int | None = None) int[source]

Sends the opc-synced query to the instrument and returns the response as integer. If you do not provide timeout, the method uses current opc_timeout.

query_opc(timeout: int = 0) int[source]

SCPI command: *OPC? Queries the instrument’s OPC bit and hence it waits until the instrument reports operation complete. If you define timeout > 0, the VISA timeout is set to that value just for this method call.

query_str(query: str) str[source]

Sends the string query to the instrument and returns the response as string. The response is trimmed of any trailing LF characters and has no length limit. This method is an alias to the query() method.

query_str_list(query: str, remove_blank_response: bool = False) List[str][source]

Sends the string query to the instrument and returns the response as List of strings, where the delimiter is comma (‘,’). Each element of the list is trimmed for leading and trailing quotes.

Meaning of the ‘remove_blank_response’:
  • False(default): whitespaces-only response is returned as a list with one empty element [‘’].

  • True: whitespaces-only response is returned as an empty list [].

query_str_list_with_opc(query: str, timeout: int | None = None, remove_blank_response: bool = False) List[str][source]
Sends a OPC-synced query and reads response from the instrument as csv-list.

If you do not provide timeout, the method uses current opc_timeout.

Meaning of the ‘remove_blank_response’:
  • False(default): whitespaces-only response is returned as a list with one empty element [‘’].

  • True: whitespaces-only response is returned as an empty list [].

query_str_stripped(query: str) str[source]

Sends the string query to the instrument and returns the response as string stripped of the trailing LF and leading/trailing single/double quotes. The stripping of the leading/trailing quotes is blocked, if the string contains the quotes in the middle. This method is an alias to the query_stripped() method.

query_str_with_opc(query: str, timeout: int | None = None) str[source]

Sends the opc-synced query to the instrument and returns the response as string. The response is trimmed of any trailing LF characters and has no length limit. If you do not provide timeout, the method uses current opc_timeout.

query_stripped(query: str) str[source]

Sends the string query to the instrument and returns the response as string stripped of the trailing LF and leading/trailing single/double quotes. The stripping of the leading/trailing quotes is blocked, if the string contains the quotes in the middle.

query_with_opc(query: str, timeout: int | None = None) str[source]

This method is an alias to the write_str_with_opc(). Sends the opc-synced query to the instrument and returns the response as string. The response is trimmed of any trailing LF characters and has no length limit. If you do not provide timeout, the method uses current opc_timeout.

read_file_from_instrument_to_pc(source_instr_file: str, target_pc_file: str, append_to_pc_file: bool = False) None[source]

SCPI Command: MMEM:DATA?

Reads file from instrument to the PC.

Set the append_to_pc_file to True if you want to append the read content to the end of the existing PC file.

reconnect(force_close: bool = False) bool[source]

If the connection is not active, the method tries to reconnect to the device. If the connection is active, and force_close is False, the method does nothing. If the connection is active, and force_close is True, the method closes, and opens the session again. Returns True, if the reconnection has been performed.

remove_instr_option(option: str) None[source]

Removes the option if exists.

reset(timeout: int = 0) None[source]

SCPI command: *RST Sends *RST command + calls the clear_status(). If you define timeout > 0, the VISA timeout is set to that value just for this method call.

reset_time_statistics() None[source]

Resets all execution and total time counters. Affects the results of get_total_time(), get_total_execution_time() and get_total_time_startpoint()

property resource_name: str

Returns the resource name used in the constructor.

self_test(timeout: int | None = None) Tuple[int, str][source]

SCPI command: *TST? Performs instrument’s self-test. Returns tuple (code:int, message: str). Code 0 means the self-test passed. You can define the custom timeout in milliseconds. If you do not define it, the method uses default self-test timeout (usually 60 secs).

send_file_from_pc_to_instrument(source_pc_file: str, target_instr_file: str) None[source]

SCPI Command: MMEM:DATA

Sends file from PC to the instrument.

classmethod set_global_logging_relative_time_of_first_entry() None[source]

This method sets the global flag, that takes the very first log entry start time of any global instance as a global relative timestamp. This means, after this call, the first instance that logs an entry set the relative timestamp for all global target instances, and begins with the start timestamp 0:00:00.000

classmethod set_global_logging_relative_timestamp(timestamp: datetime) None[source]

Sets global common relative timestamp for log entries. To use it, call the following: io.logger.set_relative_timestamp_global()

classmethod set_global_logging_relative_timestamp_now() None[source]

Sets global common relative timestamp for log entries to this moment. To use it, call the following: io.logger.set_relative_timestamp_global().

classmethod set_global_logging_target(target) None[source]

Sets global common target stream that each instance can use. To use it, call the following: io.logger.set_logging_target_global(). If an instance uses global logging target, it automatically uses the global relative timestamp (if set). You can set the target to None in order to invalidate it.

status_error_suppression_add_rule(rule: Tuple[int, str | Pattern[str]] | str | Pattern[str] | int) InstrumentStatusErrorRule[source]

Adds one suppression rule and returns the created rule object.

The rule argument can be one of the following:

  • (code, pattern) tuple: suppress only when the error code equals code AND the message matches pattern (logical AND).

  • pattern (str or re.Pattern): suppress by message only, for any error code.

  • code (int): suppress by error code only, for any message.

  • an already-built InstrumentStatusErrorRule: stored as-is.

A pattern is either a regex source string or an already-compiled re.Pattern. Message matching uses re.search(), so the pattern only has to be found somewhere in the message, not match it in full.

Rules are de-duplicated by value: adding a rule that is equal to one already in the list (same code and/or same pattern source string and flags) does not create a second entry, and the existing/normalized rule is returned.

Parameters:

rule – The rule definition, in any of the forms listed above.

Returns:

The normalized InstrumentStatusErrorRule that was added (or the equal one already present). Keep this reference to remove the rule later with status_error_suppression_remove_rule().

Raises:
  • TypeError – If rule (or a tuple’s code/pattern) has an unsupported type.

  • ValueError – If a tuple rule does not have exactly two elements.

status_error_suppression_add_rules(rules: Iterable[Tuple[int, str | Pattern[str]] | str | Pattern[str] | int]) None[source]

Same as the status_error_suppression_add_rule(), but the parameter is a collection of rules.

status_error_suppression_clear_all_rules() None[source]

Clears the status-error suppression rules - all instrument errors are reported.

status_error_suppression_get_rules() list[InstrumentStatusErrorRule][source]

Returns the status-error suppression rules used by automatic status checks, or query_all_errors / query_all_errors_with_codes.

status_error_suppression_remove_rule(rule: InstrumentStatusErrorRule) bool[source]

Removes the entered rule from the error suppression list. Returns true, if the rule existed in the list.

property supported_models: List[str]

Returns a list of the instrument models supported by this instrument driver

unlock_resource() None[source]

Unlocks the instrument to other clients.

property visa_manufacturer: str

Returns the manufacturer of the current VISA session.

property visa_timeout: int

Sets / returns visa IO timeout in milliseconds.

visa_tout_suppressor(visa_tout_ms: int = 0) VisaTimeoutSuppressor[source]

Returns Context Manager that suppresses the VISA timeout error. Careful!!!: Only the very first VISA Timeout exception is suppressed, and afterward the context ends. Therefore, use only one command per context manager, if you do not want to skip the following ones. :param visa_tout_ms: VISA Timeout in milliseconds, that is set for this context. Afterward, it is changed back. Default value: do-not-change.

write(cmd: str) None[source]

Writes the command to the instrument as string. This method is an alias to the write_str() method.

write_bin_block(cmd: str, payload: bytes) None[source]

Writes all the payload as binary data block to the instrument. The binary data header is added at the beginning of the transmission automatically, do not include it in the payload!!!

write_bin_block_from_file(cmd: str, file_path: str) None[source]

Writes data from the file as binary data block to the instrument using the provided command. Example for transferring a file from PC -> Instrument: cmd = f”MMEM:DATA ‘{INSTR_FILE_PATH}’,”.

Alternatively, use the dedicated methods for this purpose:

  • send_file_from_pc_to_instrument()

  • read_file_from_instrument_to_pc()

write_bool(cmd: str, param: bool) None[source]

Writes the command to the instrument followed by the boolean parameter: e.g.: cmd = ‘OUTPUT’ param = ‘True’, result command = ‘OUTPUT ON’

write_bool_with_opc(cmd: str, param: bool, timeout: int | None = None) None[source]

Writes the command with OPC to the instrument followed by the boolean parameter: e.g.: cmd = ‘OUTPUT’ param = ‘True’, result command = ‘OUTPUT ON’ If you do not provide timeout, the method uses current opc_timeout.

write_float(cmd: str, param: float) None[source]

Writes the command to the instrument followed by the boolean parameter: e.g.: cmd = ‘CENTER:FREQ’ param = ‘10E6’, result command = ‘CENTER:FREQ 10E6’

write_float_with_opc(cmd: str, param: float, timeout: int | None = None) None[source]

Writes the command with OPC to the instrument followed by the boolean parameter: e.g.: cmd = ‘CENTER:FREQ’ param = ‘10E6’, result command = ‘CENTER:FREQ 10E6’ If you do not provide timeout, the method uses current opc_timeout.

write_int(cmd: str, param: int) None[source]

Writes the command to the instrument followed by the integer parameter: e.g.: cmd = ‘SELECT:INPUT’ param = ‘2’, result command = ‘SELECT:INPUT 2’

write_int_with_opc(cmd: str, param: int, timeout: int | None = None) None[source]

Writes the command with OPC to the instrument followed by the integer parameter: e.g.: cmd = ‘SELECT:INPUT’ param = ‘2’, result command = ‘SELECT:INPUT 2’ If you do not provide timeout, the method uses current opc_timeout.

write_str(cmd: str) None[source]

Writes the command to the instrument as string. This method is an alias to write() method.

write_str_with_opc(cmd: str, timeout: int | None = None) None[source]

Writes the opc-synced command to the instrument. If you do not provide timeout, the method uses current opc_timeout.

write_with_opc(cmd: str, timeout: int | None = None) None[source]

This method is an alias to the write_str_with_opc(). Writes the opc-synced command to the instrument. If you do not provide timeout, the method uses current opc_timeout.

exception StatusException(rsrc_name: str, message: str, errors_list: List[Tuple[int, str]], first_exc: type | None = None)[source]

Bases: RsInstrException

Exception for instrument status errors. Tje field errors_list contains the complete list of all the errors with messages and codes.

exception TimeoutException(message: str)[source]

Bases: RsInstrException

Exception for timeout errors.

exception UnexpectedResponseException(rsrc_name: str, message: str)[source]

Bases: RsInstrException

Exception for instrument unexpected responses.

size_to_kb_mb_gb_string(data_size: int, as_additional_info: bool = False, allow_gb: bool = True) str[source]

Returns human-readable string with kilobytes, megabytes or gigabytes depending on the data_size range.

param data_size:

data size in bytes to convert

param allow_gb:

allow also Gigabytes size

param as_additional_info:

if True, the dynamic data appear in round bracket after the number in bytes, e.g. ‘12345678 bytes (11.7 MB)’. If False, only the dynamic data is returned, e.g. ‘11.7 MB’

size_to_kb_mb_string(data_size: int, as_additional_info: bool = False) str[source]

Returns human-readable string with kilobytes or megabytes depending on the data_size range.

Parameters:
  • data_size – data size in bytes to convert

  • as_additional_info – if True, the dynamic data appear in round bracket after the number in bytes, e.g. ‘12345678 bytes (11.7 MB)’. If False, only the dynamic data is returned, e.g. ‘11.7 MB’

value_to_si_string(value: float, fmt: str = '.12g', min_decimal_places: int = 0, str_after_number: str = ' ') str[source]

Returns the entered float value converted to string with SI notation. :param value: input float value :param fmt: formatting of the number. Default: ‘.12g’ :param min_decimal_places: assures minimal number of decimal places. If you set the number > 0, the function makes sure the decimal places are kept, and if there are less of them, they are filled with nulls (‘.000’ or ‘000’) Default: 0 .. rubric:: Examples

  • value = 1, fmt = ‘.12g’, min_decimal_places = 0, result -> ‘1’

  • value = 1, fmt = ‘.12g’, min_decimal_places = 3, result -> ‘1.000’

Parameters:

str_after_number – string after number, if the SI suffix is present. Default: ‘ ‘

Examples

  • 1.23 -> ‘1.23’

  • 1.23E3 -> ‘1.23 k’

  • 2.56E9 -> ‘2.56 G’

  • 11.3E-6 -> ‘-11.3 u’

This function is useful for user-readable string outputs.