InstroScope

class InstroScope(

name: str | None = None,

driver: ScopeDriverBase | None = None,

num_channels: int | None = None,

publishers: list[Publisher] | None = None,

config: ScopeConfig | dict | Path | str | None = None,

autostart: bool = False,

**kwargs,

)

Bases: Instrument

Oscilloscope instrument. Tracks scope state locally; call sync_configuration() after open() to refresh.

Initialize an InstroScope.

Provide either config or driver/num_channels together, not both.

Parameters:

• name

  Channel-name prefix for published data. Falls back to config.device.name when config is given.

• driver

  Concrete scope driver; owns its own transport:

  scope = InstroScope(
      "scope",
      driver=Keysight1200X("<visa_resource>"),
      num_channels=4,
  )
  

• num_channels

  Analog-input channel count.

• publishers

  Publishers that receive emitted Measurement/Command data. Combined with any publishers declared in config.

• config

  A ScopeConfig, a dict, or a path to a JSON config file. Its channels, acquisition, and trigger blocks are applied through the public setters on open(); each channel’s measurements are registered as background daemon functions.

• autostart

  When True, open the connection and start background polling. Requires config with at least one measurements entry, since the daemon has nothing to poll otherwise.

• **kwargs

  Default tags applied to every emitted Measurement/Command. Pass dataset_rid="<rid>" to auto-create a NominalCorePublisher (uses the on-disk ‘default’ Nominal credential).

Attributes

background_interval	Seconds between background daemon iterations (0 = no wait).
channel_names	Return an insertion-ordered snapshot of all published channel names.

Methods

add_background_daemon_function	Append method to the daemon's call list.
add_publisher	Register a publisher to receive this instrument's Measurement/Command data.
close	Stop the daemon and publishers, then close the underlying driver.
define_background_daemon	Replace all daemon functions with a single method (called with the given args).
fetch_waveform	Fetch the acquired waveform from channel.
force_trigger	Force a trigger event immediately.
get_acquisition_mode	Query the current acquisition mode (published as a string).
get_acquisition_state	Query RUNNING/STOPPED.
get_average_count	Query the average count.
get_channel	Return the most recent length samples for channel_name from the in-memory buffer.
get_coupling	Query the input coupling mode on channel (published as a string).
get_horizontal_scale	Query the timebase (seconds/div).
get_probe_attenuation	Query the probe attenuation ratio on channel.
get_sample_rate	Query the current sample rate (Sa/s).
get_single_channel_value	Return the most recent sample for channel_name from the in-memory buffer.
get_trigger_status	Query the trigger status (published as a string).
get_vertical_offset	Query the vertical offset (volts) on channel.
get_vertical_scale	Query the vertical scale (V/div) on channel.
load_settings	Recall a scope setup.
measure	Take a built-in measurement on channel.
open	Open the underlying driver and apply any configured initial state.
publish	Fan data out to every configured publisher; kwargs pass through.
run	Start continuous acquisition.
save_screenshot	Capture a screenshot and save it.
save_settings	Save scope setup.
set_acquisition_mode	Set the acquisition mode (NORMAL/AVERAGE/HIRES/PEAK_DETECT/ENVELOPE).
set_average_count	Set the average count (waveforms averaged) used in AVERAGE acquisition mode.
set_coupling	Set AC/DC input coupling on channel.
set_horizontal_scale	Set the timebase (seconds/div).
set_probe_attenuation	Set the probe attenuation ratio (e.g. 1, 10, 100) on channel.
set_trigger_level	Set the trigger level (volts).
set_trigger_mode	Set the trigger mode (AUTO/NORMAL).
set_trigger_slope	Set the trigger edge slope (RISING/FALLING/EITHER).
set_trigger_source	Set the trigger source to channel.
set_trigger_type	Set the trigger type (EDGE, PULSE, …).
set_vertical_offset	Set the vertical offset (volts) on channel.
set_vertical_scale	Set the vertical scale (V/div) on channel.
single	Arm a single-shot acquisition.
start	Start the background daemon; warns when nothing is registered to poll.
stop	Signal the background daemon to stop and join it.
stop_acquisition	Stop acquisition; records the acquisition timestamp if one was armed.
sync_configuration	Bulk-query the instrument and overwrite the tracked ScopeState with live state.

property background_interval

Seconds between background daemon iterations (0 = no wait).

open() → None

Open the underlying driver and apply any configured initial state.

start() → None

Start the background daemon; warns when nothing is registered to poll.

close() → None

Stop the daemon and publishers, then close the underlying driver.

sync_configuration() → ScopeState

Bulk-query the instrument and overwrite the tracked ScopeState with live state.

Note that trigger source/type/level/slope/mode aren’t bulk-queried — not all scopes expose every trigger field, so call the per-field getters where supported.

set_vertical_scale(

volts_per_div: float,

channel: int,

**kwargs,

) → Command

Set the vertical scale (V/div) on channel.

get_vertical_scale(

channel: int,

**kwargs,

) → Measurement | None

Query the vertical scale (V/div) on channel.

set_vertical_offset(

offset: float,

channel: int,

**kwargs,

) → Command

Set the vertical offset (volts) on channel.

get_vertical_offset(

channel: int,

**kwargs,

) → Measurement | None

Query the vertical offset (volts) on channel.

set_coupling(

coupling: Coupling,

channel: int,

**kwargs,

) → Command

Set AC/DC input coupling on channel.

get_coupling(

channel: int,

**kwargs,

) → Measurement

Query the input coupling mode on channel (published as a string).

set_probe_attenuation(

factor: float,

channel: int,

**kwargs,

) → Command

Set the probe attenuation ratio (e.g. 1, 10, 100) on channel.

get_probe_attenuation(

channel: int,

**kwargs,

) → Measurement | None

Query the probe attenuation ratio on channel.

set_horizontal_scale(

seconds_per_div: float,

**kwargs,

) → Command

Set the timebase (seconds/div).

get_horizontal_scale(

**kwargs,

) → Measurement | None

Query the timebase (seconds/div).

get_sample_rate(

**kwargs,

) → Measurement | None

Query the current sample rate (Sa/s).

set_acquisition_mode(

mode: AcquisitionMode,

**kwargs,

) → Command

Set the acquisition mode (NORMAL/AVERAGE/HIRES/PEAK_DETECT/ENVELOPE).

get_acquisition_mode(

**kwargs,

) → Measurement

Query the current acquisition mode (published as a string).

set_average_count(

count: int,

**kwargs,

) → Command

Set the average count (waveforms averaged) used in AVERAGE acquisition mode.

get_average_count(

**kwargs,

) → Measurement | None

Query the average count.

run(**kwargs) → Command

Start continuous acquisition.

stop_acquisition(**kwargs) → Command

Stop acquisition; records the acquisition timestamp if one was armed.

single(**kwargs) → Command

Arm a single-shot acquisition.

get_acquisition_state(

**kwargs,

) → Measurement

Query RUNNING/STOPPED. If STOPPED while armed, records the acquisition timestamp.

fetch_waveform(

channel: int,

timeout: float = 5.0,

**kwargs,

) → Measurement

Fetch the acquired waveform from channel.

Timestamps are ns relative to the trigger point (negative = pre-trigger). If an acquisition is armed, blocks up to timeout seconds for it to complete; raises TimeoutError if it doesn’t.

measure(

measurement_type: ScopeMeasurementType,

channel: int = 1,

timeout: float = 5.0,

**kwargs,

) → Measurement | None

Take a built-in measurement on channel.

Timestamps default to the last recorded acquisition time (falls back to time.time_ns() if none). If an acquisition is armed, blocks up to timeout seconds for it to complete; raises TimeoutError if it doesn’t.

set_trigger_source(

channel: int,

**kwargs,

) → Command

Set the trigger source to channel.

set_trigger_type(

trigger_type: TriggerType,

**kwargs,

) → Command

Set the trigger type (EDGE, PULSE, …).

set_trigger_level(

level: float,

**kwargs,

) → Command

Set the trigger level (volts).

set_trigger_slope(

slope: TriggerSlope,

**kwargs,

) → Command

Set the trigger edge slope (RISING/FALLING/EITHER).

set_trigger_mode(

mode: TriggerMode,

**kwargs,

) → Command

Set the trigger mode (AUTO/NORMAL).

force_trigger(**kwargs) → Command

Force a trigger event immediately.

get_trigger_status(

**kwargs,

) → Measurement

Query the trigger status (published as a string).

save_screenshot(

filepath: str,

to_instrument: bool = False,

**kwargs,

) → Command

Capture a screenshot and save it. to_instrument=True writes to the scope’s filesystem; otherwise to the host.

save_settings(

name: str,

to_instrument: bool = False,

**kwargs,

) → Command

Save scope setup. to_instrument=True writes to the scope’s filesystem; otherwise to the host.

load_settings(

name: str,

from_instrument: bool = False,

**kwargs,

) → Command

Recall a scope setup. from_instrument=True reads from the scope’s filesystem; otherwise from the host.

Invalidates the tracked ScopeState since instrument state changed externally; call sync_configuration() to refresh.

add_background_daemon_function(

method: Callable,

*args,

**kwargs,

)

Append method to the daemon’s call list. Use define_background_daemon to replace instead.

add_publisher(

publisher: Publisher,

)

Register a publisher to receive this instrument’s Measurement/Command data.

property channel_names: tuple[str, ...]

Return an insertion-ordered snapshot of all published channel names.

Names are complete after one full daemon iteration: {name}.loop_time publishes every iteration, so call get_channel(“loop_time”, wait_for_new_samples=True) then read channel_names to ensure the complete set has arrived.

Raises:

RuntimeError – No background buffer; start() was not called.

define_background_daemon(

method: Callable,

*args,

**kwargs,

)

Replace all daemon functions with a single method (called with the given args).

get_channel(

channel_name: str,

length: int = 1,

wait_for_new_samples: bool = False,

timeout: float = 10.0,

) → Measurement

Return the most recent length samples for channel_name from the in-memory buffer.

If the channel does not exist yet, or no sample is available, the code will always block until timeout expires.

Parameters:

• channel_name

  Name of the channel to retrieve.

• length

  Number of trailing samples to return.

• wait_for_new_samples

  Block until at least length new values arrive.

• timeout

  Seconds to wait when insufficient data exists or wait_for_new_samples=True.

Raises:

• RuntimeError – No background buffer; start() was not called.

• ChannelNotFoundError – channel had no values and no data appeared before timeout. wait_for_new_samples=True and channel did not appear within timeout.

• ChannelValueTimeoutError – wait_for_new_samples=True and values did not arrive within timeout.

get_single_channel_value(

channel_name: str,

) → float | str | None

Return the most recent sample for channel_name from the in-memory buffer.

This will not wait, if data is not available then None is returned.

Parameters:

channel_name

Name of the channel to retrieve.

Raises:

RuntimeError – No background buffer; start() was not called.

publish(

data: Measurement | Command,

**kwargs,

)

Fan data out to every configured publisher; kwargs pass through.

stop()

Signal the background daemon to stop and join it. No-op if not running.