InstroELoad#

class InstroELoad(
name: str | None = None,
driver: ELoadDriverBase | None = None,
publishers: list[Publisher] | None = None,
config: ELoadConfig | dict | Path | str | None = None,
autostart: bool = False,
**kwargs,
)[source]#

Bases: Instrument

Electronic-load instrument. Methods return Measurement/Command for publishing.

Initialize an InstroELoad.

Provide either config or driver, not both.

Parameters:
  • name

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

  • driver

    Concrete E-Load driver; owns its own transport:

    eload = InstroELoad(
        "main",
        driver=BK85XXB("ASRL19::INSTR"),
    )
    

  • publishers

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

  • config

    An ELoadConfig, a dict, or a path to a JSON config file. Its load block is applied through the public setters on open(); the input is never enabled from config.

  • 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.

close

Close the underlying driver and stop the daemon.

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) sensed 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 on channel.

open

Open the underlying driver and apply any configured load state.

output_enable

Enable or disable the output on channel.

publish

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

set_level

Set the operating level in the active mode's units (CC: A, CV: V, CP: W, CR: Ω).

set_mode

Set the channel's operation mode: CC, CR, CP, or CV.

set_range

Set the range from the expected maximum operating value in the active mode's units (CC: A, CV: V, CP: W, CR: Ω).

set_slewrate

Set the per-edge current slew rate (A/μs) for direction (RISE/FALL/BOTH).

short_output

Enable or disable the channel short.

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() → None[source]#

Open the underlying driver and apply any configured load state.

close() → None[source]#

Close the underlying driver and stop the daemon.

set_mode(
mode: LoadMode,
channel: int = 1,
**kwargs,
) → Command[source]#

Set the channel’s operation mode: CC, CR, CP, or CV. Not all models support every mode.

set_level(
value: float,
channel: int = 1,
curr_limit: float | None = None,
**kwargs,
) → Command[source]#

Set the operating level in the active mode’s units (CC: A, CV: V, CP: W, CR: Ω).

curr_limit applies only in CV mode. Raises ValueError if no mode has been set.

short_output(
enable: bool,
channel: int = 1,
**kwargs,
) → Command[source]#

Enable or disable the channel short.

Warning

Takes effect IMMEDIATELY. enable=True shorts the channel output.

set_range(
value: float,
channel: int = 1,
**kwargs,
) → Command[source]#

Set the range from the expected maximum operating value in the active mode’s units (CC: A, CV: V, CP: W, CR: Ω).

Raises ValueError if no mode has been set.

set_slewrate(
direction: SlewRateDirection,
rate: float,
channel: int = 1,
**kwargs,
) → Command[source]#

Set the per-edge current slew rate (A/μs) for direction (RISE/FALL/BOTH).

output_enable(
enable: bool,
channel: int = 1,
**kwargs,
) → Command[source]#

Enable or disable the output on channel.

get_current(
channel: int = 1,
**kwargs,
) → Measurement | None[source]#

Measure the current (amperes) sensed on channel. Returns None if unavailable.

get_voltage(
channel: int = 1,
**kwargs,
) → Measurement | None[source]#

Measure the voltage (volts) sensed 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.