Publishers - instro

Overview

Publishers are the mechanism by which instro handles data coming from instruments. In the example, we add two Publishers to a InstroDAQ.

Example

from instro.daq.drivers.labjack import LabJackTSeriesDriver
from instro.daq import InstroDAQ
from instro.lib.publishers import NominalCorePublisher, FilePublisher

publishers = [
    NominalCorePublisher(dataset_rid="ri.catalog.main.dataset.abc123"),
    FilePublisher(directory="/tmp", format="jsonl")
]

daq = InstroDAQ(
    name="myDAQ",
    driver=LabJackTSeriesDriver(device_id="440020473"),
    publishers=publishers,
)
# All data automatically streams to Nominal Core

Every Measurement and Command is now recorded by each Publisher in publishers. An Instrument will close() all publishers when it closes. Note: Do not attach the same ordinary Publisher instance to more than one instrument. Use SharedPublisher when multiple instruments need to publish to one underlying destination.

Built-in Publishers

instro provides the following built-in publishers.

FilePublisher

Writes measurements and commands to a local file in Avro, CSV, or JSON Lines format. Constructor

FilePublisher(
    directory: str | Path,                            # Directory where the file will be written
    format: Literal["json", "jsonl", "csv", "avro"] = "avro",  # Output file format
    custom_file_name: str | None = None,              # Optional base name (no extension)
)

Example:

from instro.daq.drivers.labjack import LabJackTSeriesDriver
from instro.daq import InstroDAQ
from instro.lib.publishers import FilePublisher

daq = InstroDAQ(
    name="myDAQ",
    driver=LabJackTSeriesDriver(device_id="440020473"),
    publishers=[FilePublisher(directory="./captures", format="avro")],
)

Format tradeoffs:

avro (default): Compact binary format. Uses the same schema as Nominal Core ingest, so captured files can be uploaded after the fact without transformation. Recommended for production captures.
csv: One row per (timestamp, channel, value, tags) tuple. Good for quick inspection in spreadsheet tools.
jsonl: Newline-delimited JSON, one record per publish. Recommended when you want human-readable output: each record is appended and flushed in constant time, and every flushed line is a complete record.

NominalCorePublisher

Sends measurement and command data to Nominal Core datasets. Constructor Parameters:

NominalCorePublisher(
    dataset_rid: str,              # Required: RID of your Nominal Core dataset
    batch_size: int | None = None, # Optional: Batch size before writing
    max_wait: timedelta | None = None,  # Optional: Max time before flush
    file_fallback: pathlib.Path | None = None,  # Optional: Path of fallback file (.avro) used if network connectivity is lost.
    profile: str | None = None,    # Optional: Nominal profile name
    api_key: str | None = None     # Optional: API key for authentication
)

Example:

from instro.daq.drivers.labjack import LabJackTSeriesDriver
from instro.daq import InstroDAQ
from instro.lib.publishers import NominalCorePublisher

DATASET_RID = "ri.catalog.main.dataset.abc123..."

daq = InstroDAQ(
    name="myDAQ",
    driver=LabJackTSeriesDriver(device_id="440020473"),
    publishers=[NominalCorePublisher(dataset_rid=DATASET_RID)],
)

A shorthand method to add publishing to Nominal Core is to directly pass in a dataset_rid kwarg. This requires a default profile to have been configured on the system.

daq = InstroDAQ(
    name="myDAQ",
    driver=LabJackTSeriesDriver(device_id="440020473"),
    dataset_rid="ri.catalog...",  # Automatically creates NominalCorePublisher
)

backup FilePublisher If you only need a local file as a fallback for when network connectivity to Nominal Core is lost, use the file_fallback parameter on NominalCorePublisher instead of attaching a separate FilePublisher.

NominalConnectPublisher

Streams real-time data to Nominal Connect for live visualization and monitoring during tests that use the Nominal Connect Desktop Application. Nominal Connect apps can also install instro for you; see Using Nominal Connect. Constructor Parameters:

from connect_python import Client

NominalConnectPublisher(
    client: Client,     # Nominal Connect client instance
    stream_id: str      # Stream identifier for Connect
)

Example, from a typical Connect app:

from instro.daq.drivers.labjack import LabJackTSeriesDriver
from instro.daq import InstroDAQ
from instro.lib.publishers import NominalConnectPublisher

import connect_python

@connect_python.main
def main(client: connect_python.Client):
    stream_id = "myDAQ-stream"

    daq = InstroDAQ(
        name="myDAQ",
        driver=LabJackTSeriesDriver(device_id="440020473"),
        publishers=[NominalConnectPublisher(client=client, stream_id=stream_id)],
    )

Attaching Publishers

Publishers can be attached after Instrument construction also.

...
daq = InstroDAQ(name="myDAQ", driver=LabJackTSeriesDriver(device_id="440020473"))
daq.add_publisher(NominalCorePublisher(dataset_rid="ri.catalog..."))

Custom Publisher

All Publisher implementations follow a simple protocol with two required methods publish and close:

from typing import Protocol

from instro.lib.types import Measurement, Command


class Publisher(Protocol):
    def publish(self, data: Measurement | Command, **kwargs) -> None:
        """Called each time the instrument generates data"""
        ...

    def close(self) -> None:
        """Called when the owning instrument is closed"""
        ...

publish: returns Measurements and Commands.
close: release resources owned by the Publisher. The parent instrument calls close() during teardown.

Example The example below prints to the console every time a Measurement is published corresponding to a predefined name.

from instro.lib.types import Measurement, Command


class PrintChannelPublisher:
    def __init__(self, channel_name: str):
        self.channel_name = channel_name

    def publish(self, data: Measurement | Command, **kwargs) -> None:
        try:
            if isinstance(data, Measurement):
                if data.channel_data.get(self.channel_name, None):
                    print(f"{self.channel_name}: {data.latest}")
        except Exception as e:
            print(f"Error publishing data: {e}")

    def close(self) -> None:
        pass

Publisher Wrappers

Publisher wrappers modify the behavior of other publishers by adding buffering, asynchronous processing, or explicit shared ownership. They are composable.

SharedPublisher

SharedPublisher coordinates shared ownership of one underlying publisher across multiple instruments. The default model is still exclusive ownership: one publisher instance belongs to one instrument. Use SharedPublisher only when multiple instruments must publish to the same file, stream, client, or other sink. Each instrument receives its own SharedPublisher handle. Closing one instrument closes only that handle. The underlying publisher closes after the last shared handle closes. When to use:

Multiple instruments should write to the same local capture file
Multiple instruments should publish to the same custom sink
The underlying publisher owns a resource that should stay open until all instruments finish

Parameters:

SharedPublisher(
    publisher: Publisher  # The underlying publisher to share
)

Example:

from instro.lib.publishers import FilePublisher, SharedPublisher

file_publisher = FilePublisher(directory="./captures", format="avro")
shared_publisher = SharedPublisher(file_publisher)

primary_psu.add_publisher(shared_publisher)
secondary_psu.add_publisher(shared_publisher.clone())

primary_psu.close()    # Leaves the underlying FilePublisher open.
secondary_psu.close()  # Closes the underlying FilePublisher.

See Publishers: publish shared for a full example.

BasicBufferedPublisher

BasicBufferedPublisher collects data in memory and publishes in batches, reducing the overhead of frequent publish calls. It is the concrete implementation of the abstract BufferedPublisher base, so instantiate BasicBufferedPublisher. When to use:

You’re making many small publish calls and want to reduce overhead
You want to batch data before sending to a remote service
Network or I/O latency is impacting performance

Parameters:

BasicBufferedPublisher(
    publisher: Publisher,    # The publisher to wrap
    buffer_size: int = 1000  # Flush when buffer reaches this size
)

Example:

import time
from instro.psu.drivers import SimulatedPSU
from instro.psu import InstroPSU
from instro.lib.publishers import BasicBufferedPublisher
from instro.lib.types import Measurement, Command

# Custom publisher that prints to console
class PrintChannelPublisher:
    def __init__(self, channel_name: str):
        self.channel_name = channel_name

    def publish(self, data: Measurement | Command, **kwargs):
        if isinstance(data, Measurement):
            if data.channel_data.get(self.channel_name, None):
                print(f"{self.channel_name}: {data.latest}")

    def close(self):
        print("Publisher closed")

# Wrap the custom publisher with buffering
print_publisher = PrintChannelPublisher(channel_name="myPSU.ch1.current")
buffered_publisher = BasicBufferedPublisher(print_publisher, buffer_size=5)

psu = InstroPSU(
    name="myPSU",
    driver=SimulatedPSU("TCPIP0::127.0.0.1::5025::SOCKET"),
    num_channels=2,
)
psu.add_publisher(buffered_publisher)

psu.open()
psu.output_enable(True, channel=1)

# Console output only appears every 5 measurements (when buffer flushes)
for v in range(12):
    psu.set_voltage(v * 0.1, channel=1)
    psu.get_current(channel=1)
    time.sleep(0.1)

psu.close()  # Remaining buffer is automatically flushed

The buffer is automatically flushed when it reaches capacity or when close() is called, ensuring no data is lost.

QueuedPublisher

Offloads publishing to a background thread, making publish calls non-blocking and preventing slow publishers from impacting instrument operations. When to use:

Publishing is slow (network latency, disk I/O) and blocking your test loop
You want instrument operations to proceed immediately without waiting for publish
You need guaranteed throughput for time-critical measurements

Parameters:

QueuedPublisher(
    publisher: Publisher,           # The publisher to wrap
    max_queue_size: int = 1000,     # Maximum queue size
    wait_for_queue: bool = False    # Wait for queue to empty on close
)

Example:

import time
from instro.psu.drivers import SimulatedPSU
from instro.psu import InstroPSU
from instro.lib.publishers import QueuedPublisher
from instro.lib.types import Measurement, Command

# Custom publisher with artificial delay to simulate slow I/O
class SlowPublisher:
    def publish(self, data: Measurement | Command, **kwargs):
        time.sleep(0.1)  # Simulate slow network or disk I/O
        print(f"Published {len(data.channel_data)} channels")

    def close(self):
        print("Publisher closed")

# Wrap the slow publisher with queuing
slow_publisher = SlowPublisher()
queued_publisher = QueuedPublisher(
    slow_publisher,
    max_queue_size=100,
    wait_for_queue=True  # Ensure all data is sent before closing
)

psu = InstroPSU(
    name="myPSU",
    driver=SimulatedPSU("TCPIP0::127.0.0.1::5025::SOCKET"),
    num_channels=2,
)
psu.add_publisher(queued_publisher)

psu.open()
psu.output_enable(True, channel=1)

# Publishing happens in background - no blocking!
for v in range(10):
    start = time.time()
    psu.set_voltage(v, channel=1)
    psu.get_current(channel=1)
    print(f"Loop iteration: {time.time() - start:.4f}s")  # Very fast!

psu.close()  # Waits for queue to empty before closing

Using QueuedPublisher can dramatically improve test performance when publishing is a bottleneck. The background thread handles all I/O while your test continues uninterrupted.

​Overview

​Example

​Built-in Publishers

​FilePublisher

​NominalCorePublisher

​NominalConnectPublisher

​Attaching Publishers

​Custom Publisher

​Publisher Wrappers

​SharedPublisher

​BasicBufferedPublisher

​QueuedPublisher

Overview

Example

Built-in Publishers

FilePublisher

NominalCorePublisher

NominalConnectPublisher

Attaching Publishers

Custom Publisher

Publisher Wrappers

SharedPublisher

BasicBufferedPublisher

QueuedPublisher