[python source code]

tek_hsi_connect

Module for connecting to Tektronix instruments and retrieving waveform data using gRPC.

AcqWaitOn

Bases: Enum

flowchart LR tekhsi.tek_hsi_connect.AcqWaitOn[AcqWaitOn] click tekhsi.tek_hsi_connect.AcqWaitOn href "" "tekhsi.tek_hsi_connect.AcqWaitOn"

This enumeration is used to select how to wait to access data.

AnyAcq class-attribute instance-attribute 

AnyAcq = 3

Wait for any acquisition.

NewData class-attribute instance-attribute 

NewData = 4

Wait for new data.

Using the NewData criterion for data acceptance will continue when the current data from the stored acquisition has not been read by get_data(). This is import since the underlying data is buffered because it’s stored as data on the instrument is available. If you have seen the underlying data since the last get_data() call, it will return the buffered data. If you haven’t seen the data, it will block until the next new piece of data arrives.

Examples:

>>> from tekhsi import AcqWaitOn, TekHSIConnect
>>> with TekHSIConnect("192.168.0.1:5000") as connection:
...     with connection.access_data(AcqWaitOn.NewData):
...         ...

NextAcq class-attribute instance-attribute 

NextAcq = 1

Wait for the next acquisition.

Using the NextAcq criterion for data acceptance will force the data access call to wait until the next new acquisition is available.

Examples:

>>> from tekhsi import AcqWaitOn, TekHSIConnect
>>> with TekHSIConnect("192.168.0.1:5000") as connection:
...     with connection.access_data(AcqWaitOn.NextAcq):
...         ...

Time class-attribute instance-attribute 

Time = 2

Wait for a specific time.

Using the Time criterion for data acceptance will force a time delay before accepting the next acquisition. The typical usage of this is if you are using multiple instruments and PyVISA. If you are turning on an AFG you need some time for the instrument to be set up and the data to arrive. This process is approximately the same as sleeping for half a second then calling access_data(AcqWaitOn.NextAcq).

Examples:

>>> from tekhsi import AcqWaitOn, TekHSIConnect
>>> with TekHSIConnect("192.168.0.1:5000") as connection:
...     with connection.access_data(AcqWaitOn.Time, after=0.5):
...         ...

TekHSIConnect 

TekHSIConnect(
    url: str,
    activesymbols: list[str] | None = None,
    callback: Callable | None = None,
    data_filter: Callable | None = None,
)

Support for Tektronix High-Speed Interface data API.

  • This API is intended to aid in retrieving data from instruments as fast as possible.
Parameters:
  • url (str) –

    The IP Address and port of the TekHSI server.

  • activesymbols (list[str] | None, default: None ) –

    A list of the symbols to transfer from the scope. If None, then all available symbols are transferred. Otherwise, only the selected list is transferred.

  • callback (Callable | None, default: None ) –

    An optional function to call when new data arrives. This is the fastest way to access data, and it ensures no acquisitions are missed. However, this happens in a background thread, which limits the libraries you can call from this method.

  • data_filter (Callable | None, default: None ) –

    An optional function that is used to determine if arriving data meets a custom criterion for acceptance by the client. If None, all acquisitions are accepted. However, if customer behavior is desired, then this method can be provided. Typically, these functions are used to look for specific kinds of changes, such as record length changing.

available_symbols property 

available_symbols: list[str]

Returns the list of available symbols on the instrument.

“Available” means the channel is on. What data type is returned will depend upon the probe attached or action requested by the user. This property will only return the currently available channel list. If channels are off or modes are disabled, the corresponding symbols will not be present.

Examples:

>>> from tekhsi import TekHSIConnect
>>> with TekHSIConnect("192.168.0.1:5000") as connection:
...     print(connection.available_symbols)
['ch1', 'ch1_iq', 'ch3', 'ch4_DAll']

In the above example, 'ch1' is an analog channel, 'ch1_iq' is the spectrum view channel associated with 'ch1' (when enabled). 'ch3' is another analog channel, and 'ch4_DAll' is a digital probe on 'ch4'. Types are generally determined by the name of the symbol.

Returns:
  • list[str] –

    A list of available symbols.

current_time property 

current_time: float

This property returns time relative to the connection to the gRPC client.

Returns:
  • float –

    The current time relative to the start time of the gRPC client

instrumentation_enabled property writable 

instrumentation_enabled: bool

Indicates if instrumentation is enabled.

Returns:
  • bool –

    True if instrumentation is enabled, False otherwise.

source_names property 

source_names: list[str]

Returns the list of names of sources on the instrument.

Returns:
  • list[str] –

    The list of sources.

verbose property writable 

verbose: bool

Indicates if verbose mode is enabled.

Returns:
  • bool –

    True if verbose mode is enabled, False otherwise.

access_data 

access_data(on: AcqWaitOn = NewData, after: float = -1) -> Self

Grants access to data.

Must be used as a context manager to grant access for get_data() method calls.

The access_data() context manager is used to get access to the available data. It holds access to the current acquisition (as a blocking method) for the duration of the current context. This is how you ensure that all data you get is from the same acquisition. It does not matter if the scope is running continuously or using single sequence, all the data is from the same acquisition when inside the access_data() context manager code block.

This also means you are potentially holding off scope acquisitions when inside the access_data() code block. So, it’s recommended you only get the data in the context manager, and then do any processing outside the context manager block.

Examples:

>>> from tm_data_types import AnalogWaveform
>>> from tekhsi import TekHSIConnect
>>> with TekHSIConnect("192.168.0.1:5000") as connection:
...     # Request access to data
...     with connection.access_data():
...         # Access granted
...         ch1: AnalogWaveform = connection.get_data("ch1")
...         ch3: AnalogWaveform = connection.get_data("ch3")
Parameters:
  • on (AcqWaitOn, default: NewData ) –

    Criterion for acceptance of data. See AcqWaitOn for details on each available criterion option.

  • after (float, default: -1 ) –

    Additional criterion when the on input parameter is set to AcqWaitOn.Time.

active_symbols 

active_symbols(symbols: list[str]) -> None

Sets symbols to consider moving from instrument into data cache.

Parameters:
  • symbols (list[str]) –

    list of symbols to be moved

any_acq staticmethod 

any_acq(
    previous_header: dict[str, WaveformHeader], current_header: dict[str, WaveformHeader]
) -> bool

Prebuilt acq acceptance filter that accepts all new acqs.

Parameters:
  • previous_header (dict[str, WaveformHeader]) –

    Previous header dictionary.

  • current_header (dict[str, WaveformHeader]) –

    Current header dictionary.
Returns:
  • bool –

    True if the acquisition is accepted, False otherwise.

any_horizontal_change staticmethod 

any_horizontal_change(
    previous_header: dict[str, WaveformHeader], current_header: dict[str, WaveformHeader]
) -> bool

Acq acceptance filter that accepts only acqs with changes to horizontal settings.

Parameters:
  • previous_header (dict[str, WaveformHeader]) –

    Previous header dictionary.

  • current_header (dict[str, WaveformHeader]) –

    Current header dictionary.
Returns:
  • bool –

    True if the acquisition is accepted, False otherwise.

any_vertical_change staticmethod 

any_vertical_change(
    previous_header: dict[str, WaveformHeader], current_header: dict[str, WaveformHeader]
) -> bool

Prebuilt acq acceptance filter that accepts only acqs with changes to vertical settings.

Parameters:
  • previous_header (dict[str, WaveformHeader]) –

    Previous header dictionary.

  • current_header (dict[str, WaveformHeader]) –

    Current header dictionary.
Returns:
  • bool –

    True if the acquisition is accepted, False otherwise.

close 

close() -> None

Close and clean up gRPC connection.

data_arrival staticmethod 

data_arrival(waveforms: list[AnyWaveform]) -> None

Available to be overridden if user wants to create a derived class.

This method will be called on every accepted acq.

Parameters:
  • waveforms (list[AnyWaveform]) –

    list of waveforms.

done_with_data 

done_with_data() -> None

Releases the acquisition after accessing the required data.

force_sequence 

force_sequence() -> None

force_sequence asks the instrument to please give us access.

to the current acquisition data. This is useful when connecting to a stopped instrument to get access to the currently available data. Otherwise, the API will wait until the next acquisition.

get_data 

get_data(name: str) -> AnyWaveform | None

Gets the saved data of the previous acquisition with the data item of the requested name.

The provided name parameter must correspond to the names returned from the available_symbols property, however, the names are case-insensitive.

Parameters:
  • name (str) –

    Name of the data item.
Returns:
  • AnyWaveform | None –

    The waveform data or None if caching is off or data is not found.

set_acq_filter 

set_acq_filter(acq_filter: Callable) -> None

Sets rules for acquisitions that are accepted and forwarded.

This is to allow only import data changes to be passed to the callback or saved to backing store.

Parameters:
  • acq_filter (function) –

    A function that takes two headers and returns True if the data

wait_for_data 

wait_for_data(on: AcqWaitOn = NewData, after: float = -1) -> None

Waits until specified acquisition criterion is met.

Parameters:
  • on (AcqWaitOn, default: NewData ) –

    Criterion for acceptance of data.

  • after (float, default: -1 ) –

    Additional criterion when the on input parameter is set to AcqWaitOn.Time.