InstroPSU

class InstroPSU(

name: str | None = None,

driver: PSUDriverBase | None = None,

num_channels: int | None = None,

publishers: list[Publisher] | None = None,

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

autostart: bool = False,

**kwargs,

)

Bases: Instrument

Power-supply instrument. Methods return Measurement/Command for publishing.

Initialize an InstroPSU.

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 PSU driver; owns its own transport:

  psu = InstroPSU(
      name="main",
      driver=BK9115("USB0::0xFFFF::0x9115::SN::INSTR"),
      num_channels=1,
  )
  

• num_channels

  Number of output channels on this PSU.

• publishers

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

• config

  A PSUConfig, a dict, or a path to a JSON config file.

• autostart

  When True, open the connection and start background polling.

• **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.
apply	Set the current limit, then the voltage, then the output state on channel.
clear_protection	Clear every latched protection trip on channel.
close	Disconnect from the device.
define_background_daemon	Replace all daemon functions with a single method (called with the given args).
get_channel	Return the most recent length samples for channel_name from the in-memory buffer.
get_current	Measure the current (amperes) flowing through channel.
get_current_setpoint	Query the configured current-limit setpoint (amperes) on channel.
get_operating_mode	Query whether channel is regulating in constant voltage, constant current, unregulated, or off (published as a string).
get_output_status	Query whether the output on channel is enabled.
get_overcurrent_protection_enabled	Query whether overcurrent protection is enabled on channel.
get_overcurrent_protection_level	Query the overcurrent protection threshold (amperes) on channel.
get_overcurrent_protection_tripped	Query whether overcurrent protection has tripped on channel and not been cleared.
get_overvoltage_protection_delay	Query the overvoltage protection trip delay (seconds) on channel.
get_overvoltage_protection_enabled	Query whether overvoltage protection is enabled on channel.
get_overvoltage_protection_level	Query the overvoltage protection threshold (volts) on channel.
get_overvoltage_protection_tripped	Query whether overvoltage protection has tripped on channel and not been cleared.
get_remote_sense_enabled	Query whether remote sense is enabled on channel.
get_single_channel_value	Return the most recent sample for channel_name from the in-memory buffer.
get_voltage	Measure the voltage (volts) sensed at channel terminals.
get_voltage_setpoint	Query the configured voltage setpoint (volts) on channel.
open	Establish connection to the device.
output_enable	Enable or disable the output on channel.
publish	Fan data out to every configured publisher; kwargs pass through.
set_current_limit	Set the current limit (amperes) on channel.
set_overcurrent_protection_enabled	Enable or disable overcurrent protection on channel.
set_overcurrent_protection_level	Set the overcurrent protection threshold (amperes) on channel.
set_overvoltage_protection_delay	Set the overvoltage protection trip delay (seconds) on channel.
set_overvoltage_protection_enabled	Enable or disable overvoltage protection on channel.
set_overvoltage_protection_level	Set the overvoltage protection threshold (volts) on channel.
set_remote_sense_enabled	Enable or disable remote sense on channel.
set_voltage	Set the output voltage (volts) on channel.
start	Start the background daemon thread.
stop	Signal the background daemon to stop and join it.

property background_interval

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

open()

Establish connection to the device.

close()

Disconnect from the device.

apply(

*,

current_limit: float,

voltage: float,

enable: bool = False,

channel: int,

**kwargs,

) → list[Command]

Set the current limit, then the voltage, then the output state on channel.

The output stays off unless enable=True is passed, and on the default path it is disabled before the new setpoints are written, so setpoints are never applied to a live output. Every argument is keyword-only, so a value can’t land on the wrong setpoint and the output can’t be energized without enable=True spelled out at the call site.

Holds the resource lock across all steps so nothing observes a half-applied channel, and publishes only after releasing it so publisher I/O never blocks the background daemon or other threads waiting on the instrument.

Not atomic on the instrument: if a step raises, earlier steps have already been committed and the channel is left in an unknown state. Commands for the steps that did land are still published before the exception propagates.

set_voltage(

voltage: float,

channel: int,

**kwargs,

) → Command

Set the output voltage (volts) on channel.

get_voltage(

channel: int,

**kwargs,

) → Measurement | None

Measure the voltage (volts) sensed at channel terminals. Output may vary from voltage setpoint outside of constant voltage mode. Returns None if unavailable.

set_current_limit(

current_limit: float,

channel: int,

**kwargs,

) → Command

Set the current limit (amperes) on channel.

get_current(

channel: int,

**kwargs,

) → Measurement | None

Measure the current (amperes) flowing through channel. Output may vary from current-limit setpoint outside of constant current mode. Returns None if unavailable.

output_enable(

enable: bool,

channel: int,

**kwargs,

) → Command

Enable or disable the output on channel.

get_output_status(

channel: int,

**kwargs,

) → Measurement | None

Query whether the output on channel is enabled. Returns None if unavailable.

get_voltage_setpoint(

channel: int,

**kwargs,

) → Measurement | None

Query the configured voltage setpoint (volts) on channel. Output may vary from actual measured voltage outside of constant voltage mode. Returns None if unavailable.

get_operating_mode(

channel: int,

**kwargs,

) → Measurement | None

Query whether channel is regulating in constant voltage, constant current, unregulated, or off (published as a string).

get_current_setpoint(

channel: int,

**kwargs,

) → Measurement | None

Query the configured current-limit setpoint (amperes) on channel. Output may vary from actual measured current outside of constant current mode. Returns None if unavailable.

set_overvoltage_protection_level(

voltage: float,

channel: int,

**kwargs,

) → Command

Set the overvoltage protection threshold (volts) on channel.

get_overvoltage_protection_level(

channel: int,

**kwargs,

) → Measurement | None

Query the overvoltage protection threshold (volts) on channel. Returns None if unavailable.

set_overvoltage_protection_enabled(

enabled: bool,

channel: int,

**kwargs,

) → Command

Enable or disable overvoltage protection on channel.

get_overvoltage_protection_enabled(

channel: int,

**kwargs,

) → Measurement | None

Query whether overvoltage protection is enabled on channel. Returns None if unavailable.

set_overvoltage_protection_delay(

delay: float,

channel: int,

**kwargs,

) → Command

Set the overvoltage protection trip delay (seconds) on channel.

get_overvoltage_protection_delay(

channel: int,

**kwargs,

) → Measurement | None

Query the overvoltage protection trip delay (seconds) on channel. Returns None if unavailable.

set_overcurrent_protection_level(

current: float,

channel: int,

**kwargs,

) → Command

Set the overcurrent protection threshold (amperes) on channel.

get_overcurrent_protection_level(

channel: int,

**kwargs,

) → Measurement | None

Query the overcurrent protection threshold (amperes) on channel. Returns None if unavailable.

set_overcurrent_protection_enabled(

enabled: bool,

channel: int,

**kwargs,

) → Command

Enable or disable overcurrent protection on channel.

get_overcurrent_protection_enabled(

channel: int,

**kwargs,

) → Measurement | None

Query whether overcurrent protection is enabled on channel. Returns None if unavailable.

clear_protection(

channel: int,

**kwargs,

) → Command

Clear every latched protection trip on channel. Publishes True on the protection.clear command channel.

The output stays off after the clear; re-enable it with output_enable once the cause is fixed.

get_overvoltage_protection_tripped(

channel: int,

**kwargs,

) → Measurement | None

Query whether overvoltage protection has tripped on channel and not been cleared.

Raises NotImplementedError if the driver does not implement the query, or FeatureNotSupportedError if the supply has no overvoltage protection.

get_overcurrent_protection_tripped(

channel: int,

**kwargs,

) → Measurement | None

Query whether overcurrent protection has tripped on channel and not been cleared.

Raises NotImplementedError if the driver does not implement the query, or FeatureNotSupportedError if the supply has no overcurrent protection.

set_remote_sense_enabled(

enabled: bool,

channel: int,

**kwargs,

) → Command

Enable or disable remote sense on channel.

get_remote_sense_enabled(

channel: int,

**kwargs,

) → Measurement | None

Query whether remote sense is enabled on channel. Returns None if unavailable.

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.

start()

Start the background daemon thread. No-op if already running.

stop()

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