I2CInterface#

class I2CInterface(

name: str,

driver: I2CDriverBase,

system_definition: SystemDefinition,

publishers: list[Publisher] | None = None,

**kwargs,

)[source]#

Bases: 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`	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_single_channel_value`	Return the most recent sample for `channel_name` from the in-memory buffer.
`open`	Open the underlying I2C-adapter driver.
`publish`	Fan `data` out to every configured publisher; `kwargs` pass through.
`query`	Send a batch command (OR of its enum members) to `peripheral` and read the response.
`read`	Read a register (or one of its named fields) on `peripheral`.
`read_raw`	Read `length` bytes from the 7-bit I2C `address` and decode as an integer.
`reset_reg`	Write a register's defined default value back to hardware (initialization/recovery).
`start`	Start the background daemon thread.
`stop`	Signal the background daemon to stop and join it.
`write`	Write a register (or one of its named fields) on `peripheral`.
`write_raw`	Write `data` to the 7-bit I2C `address`.
`write_read_raw`	Write-then-read without an intermediate I2C STOP (uses repeated START); decode as integer.
`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 – 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.