Photonscore.
Contact sales

photonscore.LINTag

photonscore.LINTag

LINTag.LINTagRemote

Python remote control interface for the LINTag time tagger.

LINTag is a multi-channel time-to-digital converter (TDC). This module provides a gRPC client to control the LINTag Capture desktop application, which must be running and reachable on the network.

The communication protocol uses a settings diff pattern: stage changes with the setup_* methods, then call LINTagRemote.sync to push them to the device and refresh the read-only status fields (such as the per-channel count rates) in a single round trip.

Configure two channels, watch the correlation, record for a minute:

from photonscore import LINTagRemote

tag = LINTagRemote("192.168.1.10:50052")
tag.setup_channel(0, threshold=-0.5, rising_edge=True)
tag.setup_channel(1, threshold=-0.5, rising_edge=True)
tag.setup_correlator(start_channel=0, stop_channel=1, window_ps=50000)
tag.sync()                  # apply config, read back device state
print(tag.cps)              # live count rate per channel (8 values)

time_ns, counts = tag.get_histogram()
time_s, cps = tag.get_counts()   # the CPS plot, last 60 s at 25 Hz

tag.start_recording("run.photons", seconds=60, chunks=True)
# ... acquisition runs on the device ...
tag.stop_recording()

Dependencies: grpc, numpy, photonscore.proto.lintag_v1_remote_pb2 (generated from photonscore/proto/lintag_v1_remote.proto)

LINTagRemote

class LINTagRemote(target = None)

Remote control handle for a LINTag time tagger.

Wraps the gRPC LINTagRemote service exposed by the LINTag Capture application. Configuration changes are staged locally in new_settings; call sync to commit them and read the current device state back into read_settings.

The channel grid is fixed at 8 channels (indices 0-7).

Unlike photonscore.LINCam.LINCamRemote.LINCamRemote, the constructor does not sync automatically, so read-only fields such as cps are empty until the first sync.

Attributes

  • grpc_target (str): Resolved host:port of the gRPC server.
  • grpc_channel (grpc.Channel): Underlying insecure gRPC channel. Send and receive message limits are raised to 1 GiB to accommodate large settings/state payloads.
  • grpc_client (LINTagRemoteStub): Generated gRPC stub.
  • new_settings (LINTagRemoteSettings): Buffer of changes staged but not yet sent to the device.
  • read_settings (LINTagRemoteSettings): Device state returned by the most recent sync.

sync

def sync()

Push pending settings to the device and refresh all status fields.

Sends the accumulated new_settings diff via the Sync RPC. The reply — the full device state as it was before this call's changes — lands in read_settings; call sync again to read the applied values back. After the call new_settings is reset so the next sync only sends newly staged changes.

Calling this with nothing staged is a harmless way to poll the current device state (e.g. to refresh cps).

cps

property

cps

numpy.ndarray: Per-channel count rate in Hz, one entry per channel (8).

Reflects the device state from the most recent sync; call sync first to refresh it.

setup_channel

def setup_channel(
    channel,
    threshold = None,
    offset_ns = None,
    dead_time_ps = None,
    prolonged_dead_time = None,
    divider = None,
    temrinate_50_ohm = None,
    rising_edge = None,
    falling_edge = None,
    static_calibration = None,
    gate_reference = None,
)

Stage configuration changes for a single input channel.

Only the parameters passed as non-None are written; the rest are left untouched on the device. Changes are buffered and take effect on the next sync.

Arguments

  • channel (int): Channel index to configure (0-7).
  • threshold (float): Discriminator threshold, in volts (-2 to 2.5).
  • offset_ns (float): Per-channel timing offset applied to timestamps, in nanoseconds (-100 to 100).
  • dead_time_ps (int): Dead time after a detected edge during which further edges are ignored, in picoseconds (0 to 100000).
  • prolonged_dead_time (bool): If True, the dead time is extended for events detected during the dead time (suppresses afterpulsing).
  • divider (int): Hardware prescaler; only every n-th event is recorded (1 to 32).
  • temrinate_50_ohm (bool): Terminate the channel input with 50 Ω (False leaves it high impedance). The parameter name matches the proto field spelling.
  • rising_edge (bool): Track and record the rising edge.
  • falling_edge (bool): Track and record the falling edge.
  • static_calibration (bool): If True, fix the channel calibration. Needed for signals correlated with the LINTag clock.
  • gate_reference (bool): Make this input a reference of the reference gate, with both its edges. Needs gate_available; see setup_gate.

setup_gate

def setup_gate(enable = None, epochs_before = None, epochs_after = None)

Stage reference gate changes; applied on the next sync.

The gate drops tags on the reference channels in the FPGA unless a hit on some other channel falls into a window around them, which is how a laser sync is kept out of the recording except where it is wanted. Mark the references with gate_reference of setup_channel; this method switches the gate on and sets the window.

Only the parameters passed as non-None are written. Requires firmware with the gate — check gate_available first, or the next sync raises FAILED_PRECONDITION.

Arguments

  • enable (bool): Run the gate. False keeps the references and the window as set but the hardware sees no references.
  • epochs_before (int): Window before a hit, in epochs of 32 ns (0 to 16).
  • epochs_after (int): Window after a hit, in epochs of 32 ns (0 to 16). With both windows 0 every reference tag is dropped.

gate_available

property

gate_available

bool: Whether the firmware has the reference gate, from the most recent sync. Writing any gate field without it is refused.

setup_clock

def setup_clock(source)

Stage a TDC clock source change; applied on the next sync.

Arguments

  • source (str): "internal", "external_10mhz" or "external_50mhz".

clock_source

property

clock_source

str: Clock source from the most recent sync (see setup_clock for values), or "unknown" before the first one.

clock_frequency_mhz

property

clock_frequency_mhz

float: Measured TDC clock frequency in MHz from the most recent sync.

setup_correlator

def setup_correlator(
    is_correlating = None,
    start_channel = None,
    start_falling = None,
    stop_channel = None,
    stop_falling = None,
    stop_offset_ps = None,
    window_ps = None,
    bin_size_ps = None,
    correlator_type = None,
    rolling_window_ms = None,
)

Stage configuration changes for the online correlator.

Only the parameters passed as non-None are changed. Changes are buffered and take effect on the next sync.

Arguments

  • is_correlating (bool): Run the online correlator.
  • start_channel (int): Start channel index (0-7).
  • start_falling (bool): Correlate the start channel's falling edge instead of the rising one.
  • stop_channel (int): Stop channel index (0-7).
  • stop_falling (bool): Correlate the stop channel's falling edge instead of the rising one.
  • stop_offset_ps (int): Offset added to stop timestamps, in picoseconds.
  • window_ps (int): Histogram half-width in picoseconds (3000 to 1000000); the histogram spans -window..+window.
  • bin_size_ps (int): Histogram bin size, one of BIN_SIZES_PS.
  • correlator_type (str): "start_stop" or "multi_start_stop".
  • rolling_window_ms (int): Temporal accumulation window, one of ROLLING_WINDOWS_MS; 0 accumulates forever.

correlator_type

property

correlator_type

str: Correlator type from the most recent sync (see setup_correlator for values).

get_histogram

def get_histogram(clear_after = False) -> tuple

Fetch the online correlator histogram.

Arguments

  • clear_after (bool): Clear the correlator right after the histogram is taken, starting a fresh accumulation.

Returns: tuple A (time_ns, counts) tuple: bin centers in nanoseconds as

Returns: tuple numpy.float64 and bin counts as numpy.uint32, spanning

Returns: tuple -window..+window.

get_counts

def get_counts(clear_after = False) -> tuple

Fetch the count-rate history behind the GUI's CPS plot.

Independent of sync: the device keeps a rolling buffer per channel, the last 60 s at 25 Hz, and this reads it whole. Use it where cps — a single instantaneous value per channel — is too coarse.

Arguments

  • clear_after (bool): Clear the buffers right after the read, as the GUI's Clear Countrates does.

Returns: tuple (time_s, cps) - sample times in seconds relative to the read as numpy.float64, negative into the past, and an 8 by len(time_s) array of count rates in Hz as numpy.uint32, one row per channel, oldest sample first.

queues

property

queues

dict: Backpressure from the most recent sync — the GUI's "Enqueued for" numbers, as decoder, correlator and recorder.

A queue that climbs and does not come back is the acquisition falling behind; correlator_overflow and recorder_overflow are these crossing their limits.

clear_correlator

def clear_correlator()

Clear the correlation histogram, like the GUI's Clear Correlator.

Any other staged changes are committed in the same call.

clear_cps

def clear_cps()

Clear the count-rate history, like the GUI's Clear Countrates.

Any other staged changes are committed in the same call.

start_recording

def start_recording(
    filename = None,
    seconds = None,
    chunks = False,
    chunk_seconds = None,
)

Start recording tagged events to a file on the device.

Calls sync immediately; any changes staged beforehand are committed in the same call. The GUI mirrors the running recording.

Arguments

  • filename (str): Destination .photons path on the machine running LINTag Capture. Omit to reuse the already-set path (the file number keeps incrementing); an empty string stops recording (deprecated, use stop_recording).
  • seconds (int): Record for this long. Without chunks the recording stops by itself; with chunks it rotates to a new numbered file instead. Omit for one endless recording.
  • chunks (bool): Rotate to a new file every seconds seconds until stop_recording. Requires seconds.
  • chunk_seconds (int): Deprecated alias for seconds=..., chunks=True.

stop_recording

def stop_recording()

Stop the current recording; the file path and numbering are kept.

is_recording

property

is_recording

bool: Whether a recording ran at the most recent sync.

file_number

property

file_number

int: Number of the recording file being written (or next to write).

file_photon_count

property

file_photon_count

int: Events written to the current recording file so far.

acquisition_seconds

property

acquisition_seconds

float: Time the current recording file has been open, 0 when idle.

recorder_overflow

property

recorder_overflow

bool: The recorder queue is nearly full and may drop events.

correlator_overflow

property

correlator_overflow

bool: The correlator queues overflow and drop events.