PymeasureInstrument#

class PymeasureInstrument(
name: str,
instrument: T,
publishers: list[Publisher] | None = None,
*,
channel_map: ChannelMap | dict[str, Any] | str | Path | None = None,
poll: list[str] | None = None,
device_attr: str = 'device',
close_device: bool = True,
background_config: BackgroundDaemonConfig | None = None,
**kwargs: Any,
)[source]#

Bases: Instrument, Generic[T]

An instro Instrument that publishes the traffic of a pymeasure instrument.

Every property assignment on the wrapped instrument (and its channels) publishes a Command on <name>.<channel>.cmd; every property read publishes a Measurement on <name>.<channel>. This includes property access inside the driver’s own methods, such as apply_voltage().

The pymeasure object keeps working exactly as before and stays an instance of its class: the wrapper swaps in a generated subclass, and close (or detach) swaps the original back.

Wrap instrument so its commands and readings reach publishers.

Example:

from pymeasure.instruments.keysight import KeysightE36312A
from instro.lib import FilePublisher
from instro.pymeasure import PymeasureInstrument

with PymeasureInstrument(
    name="psu",
    instrument=KeysightE36312A("USB0::0x2A8D::0x1102::MY123::INSTR"),
    publishers=[FilePublisher(format="jsonl")],
    channel_map="psu_map.json",
    poll=["ch_1.voltage", "ch_1.current"],
) as psu:
    psu.device.ch_1.voltage_setpoint = 5  # publishes psu.ch1.voltage_setpoint.cmd
    psu.start()                           # polls ch_1.voltage/current every second
    print(psu.get_channel("ch1.current", wait_for_new_samples=True).latest)
Parameters:

  • name

    Channel-name prefix for everything published.

  • instrument

    The pymeasure instrument, already constructed (connected).

  • publishers

    Publishers that receive the commands and measurements.

  • channel_map

    Which properties publish and under which names: a ChannelMap, its dict form, or the path of a JSON file from instro pymeasure map. None publishes every property under the names the inspection rules give it.

  • poll

    Attribute paths ("ch_1.voltage") the background daemon reads every background_interval seconds once start is called.

  • device_attr

    Extra attribute name for the pymeasure instrument, in addition to device; e.g. "smu" makes wrapper.smu work.

  • close_device

    Close the pymeasure instrument’s adapter in close. close doesn’t call the driver’s shutdown(), which on many drivers turns outputs off; call device.shutdown() first if you want the instrument left in its safe state.

  • background_config

    Background-daemon settings.

  • **kwargs

    Default tags, or dataset_rid= (see Instrument).

Raises:

  • RuntimeError – instrument is already attached to another PymeasureInstrument.

  • ValueError – The channel map does not fit instrument, a poll path is unknown or not readable, or device_attr collides with an existing attribute.

Attributes

background_interval

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

background_running

Whether the background daemon thread is running.

channel_map

Attribute path to channel name (without the instrument prefix) for every published property.

channel_names

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

device

The wrapped pymeasure instrument; use it exactly as without instro.

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, close the publishers, detach, and close the device if close_device.

define_background_daemon

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

detach

Stop the background daemon, stop publishing, and restore the original classes.

get_channel

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

get_single_channel_value

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

open

No-op base hook.

publish

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

start

Start the background daemon thread.

stop

Signal the background daemon to stop and join it.
property device: T#

The wrapped pymeasure instrument; use it exactly as without instro.

property channel_map: dict[str, str]#

Attribute path to channel name (without the instrument prefix) for every published property.

detach() → None[source]#

Stop the background daemon, stop publishing, and restore the original classes. Idempotent.

close() → None[source]#

Stop the daemon, close the publishers, detach, and close the device if close_device.

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 background_interval#

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

property background_running: bool#

Whether the background daemon thread is running.

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.

open() → None#

No-op base hook. Category subclasses override to open their transport.

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.