# 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`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument "instro.lib.instrument.Instrument"), [`Generic`](https://docs.python.org/3/library/typing.html#typing.Generic "(in Python v3.15)")[`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`](https://dev.nominal.io/instro/ref/library/instro.lib.types.Command.md#instro.lib.types.Command "instro.lib.types.Command") on `<name>.<channel>.cmd`; every property read publishes a
[`Measurement`](https://dev.nominal.io/instro/ref/library/instro.lib.types.Measurement.md#instro.lib.types.Measurement "instro.lib.types.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`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.close "instro.pymeasure.PymeasureInstrument.close") (or [`detach`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.detach "instro.pymeasure.PymeasureInstrument.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`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.ChannelMap.md#instro.pymeasure.ChannelMap "instro.pymeasure.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`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.start "instro.pymeasure.PymeasureInstrument.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`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.close "instro.pymeasure.PymeasureInstrument.close"). [`close`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.close "instro.pymeasure.PymeasureInstrument.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**](https://docs.python.org/3/builtins/exceptions.html#RuntimeError "(in Python v3.15)") – `instrument` is already attached to another [`PymeasureInstrument`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument "instro.pymeasure.PymeasureInstrument").
    - [**ValueError**](https://docs.python.org/3/builtins/exceptions.html#ValueError "(in Python v3.15)") – 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`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.background_interval "instro.pymeasure.PymeasureInstrument.background_interval") | Seconds between background daemon iterations (0 = no wait). |
| --- | --- |
| [`background_running`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.background_running "instro.pymeasure.PymeasureInstrument.background_running") | Whether the background daemon thread is running. |
| [`channel_map`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.channel_map "instro.pymeasure.PymeasureInstrument.channel_map") | Attribute path to channel name (without the instrument prefix) for every published property. |
| [`channel_names`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.channel_names "instro.pymeasure.PymeasureInstrument.channel_names") | Return an insertion-ordered snapshot of all published channel names. |
| [`device`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.device "instro.pymeasure.PymeasureInstrument.device") | The wrapped pymeasure instrument; use it exactly as without instro. |

Methods

| [`add_background_daemon_function`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.add_background_daemon_function "instro.pymeasure.PymeasureInstrument.add_background_daemon_function") | Append `method` to the daemon's call list. |
| --- | --- |
| [`add_publisher`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.add_publisher "instro.pymeasure.PymeasureInstrument.add_publisher") | Register a publisher to receive this instrument's Measurement/Command data. |
| [`close`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.close "instro.pymeasure.PymeasureInstrument.close") | Stop the daemon, close the publishers, detach, and close the device if `close_device`. |
| [`define_background_daemon`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.define_background_daemon "instro.pymeasure.PymeasureInstrument.define_background_daemon") | Replace all daemon functions with a single `method` (called with the given args). |
| [`detach`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.detach "instro.pymeasure.PymeasureInstrument.detach") | Stop the background daemon, stop publishing, and restore the original classes. |
| [`get_channel`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.get_channel "instro.pymeasure.PymeasureInstrument.get_channel") | Return the most recent `length` samples for `channel_name` from the in-memory buffer. |
| [`get_single_channel_value`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.get_single_channel_value "instro.pymeasure.PymeasureInstrument.get_single_channel_value") | Return the most recent sample for `channel_name` from the in-memory buffer. |
| [`open`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.open "instro.pymeasure.PymeasureInstrument.open") | No-op base hook. |
| [`publish`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.publish "instro.pymeasure.PymeasureInstrument.publish") | Fan `data` out to every configured publisher; `kwargs` pass through. |
| [`start`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.start "instro.pymeasure.PymeasureInstrument.start") | Start the background daemon thread. |
| [`stop`](https://dev.nominal.io/instro/ref/library/instro.pymeasure.PymeasureInstrument.md#instro.pymeasure.PymeasureInstrument.stop "instro.pymeasure.PymeasureInstrument.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**](https://docs.python.org/3/builtins/exceptions.html#RuntimeError "(in Python v3.15)") – 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**](https://docs.python.org/3/builtins/exceptions.html#RuntimeError "(in Python v3.15)") – No background buffer; `start()` was not called.
    - [**ChannelNotFoundError**](https://dev.nominal.io/instro/ref/library/instro.lib.publishers.channel_buffer.ChannelNotFoundError.md#instro.lib.publishers.channel_buffer.ChannelNotFoundError "instro.lib.publishers.channel_buffer.ChannelNotFoundError") – channel had no values and no data appeared before `timeout`.
      `wait_for_new_samples=True` and channel did not appear within `timeout`.
    - [**ChannelValueTimeoutError**](https://dev.nominal.io/instro/ref/library/instro.lib.publishers.channel_buffer.ChannelValueTimeoutError.md#instro.lib.publishers.channel_buffer.ChannelValueTimeoutError "instro.lib.publishers.channel_buffer.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**](https://docs.python.org/3/builtins/exceptions.html#RuntimeError "(in Python v3.15)") – 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.