# I2CInterface

## class I2CInterface( name: str, driver: I2CDriverBase, system_definition: SystemDefinition, publishers: list[Publisher] | None = None, **kwargs, )[source]

Bases: [`Instrument`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument "instro.lib.instrument.Instrument")

I2C interface supporting register-based and command-based devices.

Commands are enum values OR’d together and sent as a single byte stream.
The `SystemDefinition` is read-only — `I2CInterface` resolves names to
transactions on every call; the vendor driver never sees it.

Initialize an I2CInterface.

Parameters:
:   - **name**

      Channel-name prefix for published data.
    - **driver**

      Concrete I2C-adapter driver; owns its own transport:

      ```
      i2c = I2CInterface(
          "main",
          driver=Aardvark(serial_number="2239-764425"),
          system_definition=system,
      )
      ```
    - **system_definition**

      Bus description (devices, register maps, comms params).
    - **publishers**

      Publishers that receive emitted Measurement/Command data.
    - ****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`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.background_interval "instro.i2c.I2CInterface.background_interval") | Seconds between background daemon iterations (0 = no wait). |
| --- | --- |
| [`channel_names`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.channel_names "instro.i2c.I2CInterface.channel_names") | Return an insertion-ordered snapshot of all published channel names. |

Methods

| [`add_background_daemon_function`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.add_background_daemon_function "instro.i2c.I2CInterface.add_background_daemon_function") | Append `method` to the daemon's call list. |
| --- | --- |
| [`add_publisher`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.add_publisher "instro.i2c.I2CInterface.add_publisher") | Register a publisher to receive this instrument's Measurement/Command data. |
| [`close`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.close "instro.i2c.I2CInterface.close") | Close the underlying driver and stop the daemon. |
| [`define_background_daemon`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.define_background_daemon "instro.i2c.I2CInterface.define_background_daemon") | Replace all daemon functions with a single `method` (called with the given args). |
| [`get_channel`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.get_channel "instro.i2c.I2CInterface.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/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.get_single_channel_value "instro.i2c.I2CInterface.get_single_channel_value") | Return the most recent sample for `channel_name` from the in-memory buffer. |
| [`open`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.open "instro.i2c.I2CInterface.open") | Open the underlying I2C-adapter driver. |
| [`publish`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.publish "instro.i2c.I2CInterface.publish") | Fan `data` out to every configured publisher; `kwargs` pass through. |
| [`query`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.query "instro.i2c.I2CInterface.query") | Send a batch command (OR of its enum members) to `peripheral` and read the response. |
| [`read`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.read "instro.i2c.I2CInterface.read") | Read a register (or one of its named fields) on `peripheral`. |
| [`read_raw`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.read_raw "instro.i2c.I2CInterface.read_raw") | Read `length` bytes from the 7-bit I2C `address` and decode as an integer. |
| [`reset_reg`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.reset_reg "instro.i2c.I2CInterface.reset_reg") | Write a register's defined default value back to hardware (initialization/recovery). |
| [`start`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.start "instro.i2c.I2CInterface.start") | Start the background daemon thread. |
| [`stop`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.stop "instro.i2c.I2CInterface.stop") | Signal the background daemon to stop and join it. |
| [`write`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.write "instro.i2c.I2CInterface.write") | Write a register (or one of its named fields) on `peripheral`. |
| [`write_raw`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.write_raw "instro.i2c.I2CInterface.write_raw") | Write `data` to the 7-bit I2C `address`. |
| [`write_read_raw`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.write_read_raw "instro.i2c.I2CInterface.write_read_raw") | Write-then-read without an intermediate I2C STOP (uses repeated START); decode as integer. |
| [`write_then_read_raw`](https://dev.nominal.io/instro/ref/instruments/instro.i2c.I2CInterface.md#instro.i2c.I2CInterface.write_then_read_raw "instro.i2c.I2CInterface.write_then_read_raw") | Write-then-read with an I2C STOP between operations (required by devices that need STOP to process input). |

### write_read_raw( address: int, payload: bytes, length: int, endianness: Literal['little', 'big'], ) → int[source]

Write-then-read without an intermediate I2C STOP (uses repeated START); decode as integer.

Typical use: write a register address, then read its contents.

Parameters:
:   - **address**

      7-bit I2C device address.
    - **payload**

      Bytes to write (typically the register address).
    - **length**

      Number of bytes to read back.
    - **endianness**

      Byte order for the integer decode.

### write_raw(address: int, data: bytes)[source]

Write `data` to the 7-bit I2C `address`.

### read_raw( address: int, length: int, endianness: Literal['little', 'big'], ) → int[source]

Read `length` bytes from the 7-bit I2C `address` and decode as an integer.

### write_then_read_raw( address: int, payload: bytes, length: int, endianness: Literal['little', 'big'], ) → int[source]

Write-then-read with an I2C STOP between operations (required by devices that need STOP to process input).

### read( peripheral: str, register_alias: str, field: str = '', **kwargs, ) → Measurement[source]

Read a register (or one of its named fields) on `peripheral`.

With `field`, only the named bit field is extracted (masked and shifted).
Raises `ValueError` if `peripheral` is not a register-based device.

### write( peripheral: str, register_alias: str, value: int, field: str = '', **kwargs, ) → Command[source]

Write a register (or one of its named fields) on `peripheral`.

With `field`, performs a thread-safe read-modify-write of just that
field (mask, insert, write). Raises `ValueError` if `peripheral` is
not a register-based device.

### reset_reg(peripheral: str, register_alias: str, **kwargs)[source]

Write a register’s defined default value back to hardware (initialization/recovery).

### query( peripheral: str, batch_command: str, **kwargs, ) → Measurement[source]

Send a batch command (OR of its enum members) to `peripheral` and read the response.

Uses write-then-read with an I2C STOP between operations. Raises
`ValueError` if `peripheral` is not a command-based device, or if
`batch_command` references undefined command enums.

### open()[source]

Open the underlying I2C-adapter driver.

### close()[source]

Close the underlying driver and stop the daemon.

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

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